# v0.2.17 # { "Depends": "py-genlayer:1jb45aa8ynh2a9c9xn3b7qqh8sm5q93hwfp7jqmwsfhh8jpz09h6" } from genlayer import * from genlayer.gl.vm import UserError import json DEFAULT_MARKETPLACE_FEE_BPS = 250 @gl.evm.contract_interface class NativeRecipient: class View: pass class Write: pass class GenesisMarketplace(gl.Contract): """ GENESIS Marketplace contract for GenLayer. Handles fixed-price NFT listings, cancellations, purchases, royalty accounting, marketplace fee accounting, ownership tracking, sale history, native GEN payment validation, withdrawal accounting, and collection activity. IMPORTANT PAYMENT UNIT: - All prices and payment values are stored in wei. - GenLayer native GEN uses 18 decimals: 1 GEN = 10^18 wei. - list_item.price must be a decimal-free integer string in wei. Storage model: - owners: token_key -> owner wallet - creators: collection_id -> creator wallet - royalty_bps: collection_id -> royalty basis points string - pending_withdrawals: wallet:currency -> wei string - listings: listing_id -> JSON listing object - listing_ids: numeric index -> listing_id - sales: sale_id -> JSON sale object - sale_ids: numeric index -> sale_id """ owner: str marketplace_fee_bps: u256 owners: TreeMap[str, str] creators: TreeMap[str, str] royalty_bps: TreeMap[str, str] pending_withdrawals: TreeMap[str, str] listings: TreeMap[str, str] listing_ids: TreeMap[str, str] sales: TreeMap[str, str] sale_ids: TreeMap[str, str] total_listings: u256 total_sales: u256 def __init__(self) -> None: self.owner = str(gl.message.sender_address) self.marketplace_fee_bps = u256(DEFAULT_MARKETPLACE_FEE_BPS) self.owners = TreeMap() self.creators = TreeMap() self.royalty_bps = TreeMap() self.pending_withdrawals = TreeMap() self.listings = TreeMap() self.listing_ids = TreeMap() self.sales = TreeMap() self.sale_ids = TreeMap() self.total_listings = u256(0) self.total_sales = u256(0) # --------------------------------------------------------------------- # Admin / token registry # --------------------------------------------------------------------- @gl.public.write def set_marketplace_fee_bps(self, fee_bps: int) -> int: self.require_contract_owner() fee = self.bps_value(fee_bps, "marketplace_fee_bps") self.marketplace_fee_bps = u256(fee) return fee @gl.public.write def register_token( self, collection_id: str, token_id: str, token_owner: str, creator: str, royalty_bps: int, ) -> str: """ Registers or updates token ownership in this marketplace's local source of truth. For production, replace this with verified NFT collection ownership integration. For current GENESIS testing, only the marketplace owner can register/update tokens. """ self.require_contract_owner() self.require_text(collection_id, "collection_id") self.require_text(token_id, "token_id") self.require_text(token_owner, "token_owner") self.require_text(creator, "creator") royalty = self.bps_value(royalty_bps, "royalty_bps") key = self.token_key(collection_id, token_id) self.owners[key] = token_owner self.creators[collection_id] = creator self.royalty_bps[collection_id] = str(royalty) result = { "collection_id": collection_id, "token_id": token_id, "owner": token_owner, "creator": creator, "royalty_bps": royalty, } return self.dumps(result) @gl.public.write def transfer_registered_token(self, collection_id: str, token_id: str, new_owner: str) -> str: """ Manual owner update for demos/tests. Current registered owner or contract owner can update it. """ self.require_text(collection_id, "collection_id") self.require_text(token_id, "token_id") self.require_text(new_owner, "new_owner") key = self.token_key(collection_id, token_id) current_owner = self.require_token_owner(collection_id, token_id) sender = str(gl.message.sender_address) if sender != current_owner and sender != self.owner: raise UserError("Only token owner or contract owner can transfer registered token") self.owners[key] = new_owner result = { "collection_id": collection_id, "token_id": token_id, "previous_owner": current_owner, "new_owner": new_owner, } return self.dumps(result) # --------------------------------------------------------------------- # Marketplace writes # --------------------------------------------------------------------- @gl.public.write def list_item(self, collection_id: str, token_id: str, price: str, currency: str) -> str: self.require_text(collection_id, "collection_id") self.require_text(token_id, "token_id") self.require_text(currency, "currency") self.require_native_currency(currency) parsed_price = self.positive_amount(price, "price") token_owner = self.require_token_owner(collection_id, token_id) seller = str(gl.message.sender_address) if seller != token_owner: raise UserError("Only NFT owner can list") if self.has_active_listing(collection_id, token_id): raise UserError("Token already has an active listing") listing_number = int(self.total_listings) + 1 listing_id = self.make_listing_id(listing_number) listing = { "listing_id": listing_id, "collection_id": collection_id, "token_id": token_id, "seller": seller, "price": str(parsed_price), "currency": currency, "status": "active", "created_at": listing_number, "expires_at": 0, "sold_at": 0, } self.listings[listing_id] = self.dumps(listing) self.listing_ids[str(int(self.total_listings))] = listing_id self.total_listings = u256(listing_number) return self.dumps(listing) @gl.public.write def cancel_listing(self, listing_id: str) -> str: listing = self.require_listing(listing_id) sender = str(gl.message.sender_address) if str(listing.get("seller", "")) != sender: raise UserError("Only seller can cancel") if str(listing.get("status", "")) != "active": raise UserError("Cannot cancel inactive listing") listing["status"] = "cancelled" self.listings[listing_id] = self.dumps(listing) return self.dumps(listing) @gl.public.write.payable def buy_item(self, listing_id: str) -> str: listing = self.require_listing(listing_id) buyer = str(gl.message.sender_address) if str(listing.get("status", "")) != "active": raise UserError("Cannot buy inactive listing") if buyer == str(listing.get("seller", "")): raise UserError("Seller cannot buy own listing") collection_id = str(listing.get("collection_id", "")) token_id = str(listing.get("token_id", "")) seller = str(listing.get("seller", "")) current_owner = self.require_token_owner(collection_id, token_id) if current_owner != seller: listing["status"] = "expired" self.listings[listing_id] = self.dumps(listing) raise UserError("Stale listing owner mismatch") price = self.positive_amount(listing.get("price", "0"), "price") self.require_exact_payment(price) sale = self.record_sold_listing(listing) self.credit_sale_distribution(sale) self.owners[self.token_key(collection_id, token_id)] = buyer listing["status"] = "sold" listing["sold_at"] = int(self.total_sales) self.listings[listing_id] = self.dumps(listing) return self.dumps(sale) @gl.public.write def mark_listing_expired(self, listing_id: str) -> str: listing = self.require_listing(listing_id) if str(listing.get("status", "")) != "active": raise UserError("Only active listings can be expired") listing["status"] = "expired" self.listings[listing_id] = self.dumps(listing) return self.dumps(listing) @gl.public.write def withdraw_balance(self, currency: str) -> str: self.require_text(currency, "currency") self.require_native_currency(currency) wallet = str(gl.message.sender_address) key = self.withdrawal_key(wallet, currency) amount = self.get_pending_withdrawal_internal(wallet, currency) if amount <= 0: raise UserError("No balance to withdraw") self.pending_withdrawals[key] = "0" self.transfer_native(wallet, amount) receipt = { "wallet": wallet, "currency": currency, "amount": str(amount), "status": "withdrawn", } return self.dumps(receipt) # --------------------------------------------------------------------- # Views # --------------------------------------------------------------------- @gl.public.view def get_marketplace_fee_bps(self) -> int: return int(self.marketplace_fee_bps) @gl.public.view def get_owner(self, collection_id: str, token_id: str) -> str: return self.require_token_owner(collection_id, token_id) @gl.public.view def get_creator(self, collection_id: str) -> str: self.require_text(collection_id, "collection_id") return self.creators[collection_id] if self.creator_exists(collection_id) else "" @gl.public.view def get_royalty_bps(self, collection_id: str) -> int: self.require_text(collection_id, "collection_id") if not self.royalty_exists(collection_id): return 0 return int(self.royalty_bps[collection_id]) @gl.public.view def get_listing(self, listing_id: str) -> str: listing = self.require_listing(listing_id) return self.dumps(listing) @gl.public.view def get_listing_id(self, index: int) -> str: if index < 0 or index >= int(self.total_listings): raise UserError("Listing index out of range") return self.listing_ids[str(index)] @gl.public.view def get_total_listings(self) -> int: return int(self.total_listings) @gl.public.view def get_total_sales(self) -> int: return int(self.total_sales) @gl.public.view def get_sale(self, sale_id: str) -> str: self.require_text(sale_id, "sale_id") if not self.sale_exists(sale_id): raise UserError("Unknown sale") return self.sales[sale_id] @gl.public.view def get_sale_id(self, index: int) -> str: if index < 0 or index >= int(self.total_sales): raise UserError("Sale index out of range") return self.sale_ids[str(index)] @gl.public.view def get_active_listings(self, collection_id: str) -> str: self.require_text(collection_id, "collection_id") result = [] total = int(self.total_listings) for i in range(total): listing_id = self.listing_ids[str(i)] listing = self.loads(self.listings[listing_id]) if ( str(listing.get("collection_id", "")) == collection_id and str(listing.get("status", "")) == "active" ): result.append(listing) return self.dumps(result) @gl.public.view def get_sale_history(self, collection_id: str) -> str: """ Pass an empty string to return all sales. """ result = [] total = int(self.total_sales) for i in range(total): sale_id = self.sale_ids[str(i)] sale = self.loads(self.sales[sale_id]) if collection_id == "" or str(sale.get("collection_id", "")) == collection_id: result.append(sale) return self.dumps(result) @gl.public.view def get_pending_withdrawal(self, wallet: str, currency: str) -> str: self.require_text(wallet, "wallet") self.require_text(currency, "currency") self.require_native_currency(currency) return str(self.get_pending_withdrawal_internal(wallet, currency)) # --------------------------------------------------------------------- # Internal helpers # --------------------------------------------------------------------- def require_exact_payment(self, expected_value: int) -> int: if expected_value < 0: raise UserError("expected payment value cannot be negative") actual_value = self.message_value(expected_value) if actual_value != expected_value: raise UserError( "Incorrect payment value: expected " + str(expected_value) + ", received " + str(actual_value) ) return actual_value def message_value(self, expected_value: int) -> int: """ Reads native value attached to the transaction. If the GenLayer runtime exposes a different field name, update this helper only. Purchases must not silently pass when native value is unavailable. """ candidates = ["value", "amount", "transaction_value"] for field in candidates: try: raw_value = getattr(gl.message, field) return self.non_negative_amount(raw_value, "message value") except Exception: pass if expected_value == 0: return 0 raise UserError("Native transaction value is not available in this GenLayer runtime") def transfer_native(self, recipient: str, amount: int) -> None: if amount <= 0: raise UserError("transfer amount must be positive") self.require_text(recipient, "recipient") NativeRecipient(Address(recipient)).emit_transfer(value=u256(amount)) def credit_sale_distribution(self, sale: dict) -> None: currency = str(sale.get("currency", "")) seller = str(sale.get("seller", "")) collection_id = str(sale.get("collection_id", "")) seller_proceeds = self.non_negative_amount(sale.get("seller_proceeds", "0"), "seller_proceeds") royalty_amount = self.non_negative_amount(sale.get("royalty_amount", "0"), "royalty_amount") marketplace_fee = self.non_negative_amount(sale.get("marketplace_fee", "0"), "marketplace_fee") self.credit_withdrawal(seller, currency, seller_proceeds) if royalty_amount > 0 and self.creator_exists(collection_id): creator = self.creators[collection_id] self.credit_withdrawal(creator, currency, royalty_amount) elif royalty_amount > 0: self.credit_withdrawal(seller, currency, royalty_amount) if marketplace_fee > 0: self.credit_withdrawal(self.owner, currency, marketplace_fee) def credit_withdrawal(self, wallet: str, currency: str, amount: int) -> None: self.require_text(wallet, "wallet") self.require_text(currency, "currency") if amount < 0: raise UserError("withdrawal amount cannot be negative") if amount == 0: return key = self.withdrawal_key(wallet, currency) current = self.get_pending_withdrawal_internal(wallet, currency) self.pending_withdrawals[key] = str(current + amount) def get_pending_withdrawal_internal(self, wallet: str, currency: str) -> int: key = self.withdrawal_key(wallet, currency) try: value = self.pending_withdrawals[key] except Exception: return 0 if not isinstance(value, str) or value == "": return 0 try: parsed = int(value) except Exception: return 0 if parsed < 0: return 0 return parsed def withdrawal_key(self, wallet: str, currency: str) -> str: return wallet + ":" + currency def record_sold_listing(self, listing: dict) -> dict: listing_id = str(listing.get("listing_id", "")) if str(listing.get("status", "")) != "active": raise UserError("Cannot record sale for inactive listing") collection_id = str(listing.get("collection_id", "")) token_id = str(listing.get("token_id", "")) buyer = str(gl.message.sender_address) seller = str(listing.get("seller", "")) price = self.positive_amount(listing.get("price", "0"), "price") currency = str(listing.get("currency", "")) royalty = 0 if self.royalty_exists(collection_id): royalty = int(self.royalty_bps[collection_id]) royalty_amount = price * royalty // 10000 marketplace_fee = price * int(self.marketplace_fee_bps) // 10000 seller_proceeds = price - royalty_amount - marketplace_fee if seller_proceeds < 0: raise UserError("Fees exceed sale price") sale_number = int(self.total_sales) + 1 sale_id = self.make_sale_id(sale_number) sale = { "sale_id": sale_id, "listing_id": listing_id, "collection_id": collection_id, "token_id": token_id, "buyer": buyer, "seller": seller, "price": str(price), "currency": currency, "royalty_bps": royalty, "royalty_amount": str(royalty_amount), "marketplace_fee": str(marketplace_fee), "seller_proceeds": str(seller_proceeds), } self.sales[sale_id] = self.dumps(sale) self.sale_ids[str(int(self.total_sales))] = sale_id self.total_sales = u256(sale_number) return sale def require_listing(self, listing_id: str) -> dict: self.require_text(listing_id, "listing_id") if not self.listing_exists(listing_id): raise UserError("Unknown listing") parsed = self.loads(self.listings[listing_id]) if not isinstance(parsed, dict): raise UserError("Stored listing is invalid") return parsed def listing_exists(self, listing_id: str) -> bool: try: value = self.listings[listing_id] return isinstance(value, str) and value != "" except Exception: return False def sale_exists(self, sale_id: str) -> bool: try: value = self.sales[sale_id] return isinstance(value, str) and value != "" except Exception: return False def creator_exists(self, collection_id: str) -> bool: try: value = self.creators[collection_id] return isinstance(value, str) and value != "" except Exception: return False def royalty_exists(self, collection_id: str) -> bool: try: value = self.royalty_bps[collection_id] return isinstance(value, str) and value != "" except Exception: return False def require_token_owner(self, collection_id: str, token_id: str) -> str: self.require_text(collection_id, "collection_id") self.require_text(token_id, "token_id") key = self.token_key(collection_id, token_id) try: owner = self.owners[key] except Exception: raise UserError("Unknown token") if not isinstance(owner, str) or owner == "": raise UserError("Unknown token") return owner def has_active_listing(self, collection_id: str, token_id: str) -> bool: total = int(self.total_listings) for i in range(total): listing_id = self.listing_ids[str(i)] listing = self.loads(self.listings[listing_id]) if ( str(listing.get("collection_id", "")) == collection_id and str(listing.get("token_id", "")) == token_id and str(listing.get("status", "")) == "active" ): return True return False def require_contract_owner(self) -> None: if str(gl.message.sender_address) != self.owner: raise UserError("Only contract owner can perform this action") def require_text(self, value: str, label: str) -> None: if not isinstance(value, str) or value.strip() == "": raise UserError(label + " is required") def positive_amount(self, value, label: str) -> int: parsed = self.non_negative_amount(value, label) if parsed <= 0: raise UserError(label + " must be positive") return parsed def non_negative_amount(self, value, label: str) -> int: try: parsed = int(value) except Exception: raise UserError(label + " must be an integer amount") if parsed < 0: raise UserError(label + " must be non-negative") return int(parsed) def require_native_currency(self, currency: str) -> None: if currency not in ("GEN", "NATIVE"): raise UserError("Only native GEN payments are supported in this version") def bps_value(self, value: int, label: str) -> int: if not isinstance(value, int): raise UserError(label + " must be an integer") if value < 0 or value > 10000: raise UserError(label + " must be between 0 and 10000") return int(value) def token_key(self, collection_id: str, token_id: str) -> str: return collection_id + ":" + token_id def make_listing_id(self, number: int) -> str: raw = str(number) while len(raw) < 6: raw = "0" + raw return "listing_" + raw def make_sale_id(self, number: int) -> str: raw = str(number) while len(raw) < 6: raw = "0" + raw return "sale_" + raw def loads(self, value: str): try: parsed = json.loads(value) except Exception: raise UserError("Invalid JSON") if not isinstance(parsed, dict) and not isinstance(parsed, list): raise UserError("JSON must encode an object or list") return parsed def dumps(self, value) -> str: return json.dumps(value, sort_keys=True, separators=(",", ":"))