# 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. Option B architecture: - Each NFT collection may have its own deployed NFT contract. - Marketplace listings are keyed by collection_id + token_id for app UX. - Each registered token also stores nft_collection_address for indexing, verification, and future direct ownership checks. 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: collection_id:token_id -> owner wallet - token_contracts: collection_id:token_id -> NFT collection contract address - 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] token_contracts: 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.token_contracts = 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: return self.register_token_with_contract(collection_id, "", token_id, token_owner, creator, royalty_bps) @gl.public.write def register_token_with_contract( self, collection_id: str, nft_collection_address: 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 Option B, pass the real nft_collection_address for the collection. This keeps the existing GENESIS marketplace payment flow working while preserving the link to the deployed NFT collection contract. """ 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.token_contracts[key] = nft_collection_address self.creators[collection_id] = creator self.royalty_bps[collection_id] = str(royalty) result = { "collection_id": collection_id, "nft_collection_address": nft_collection_address, "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: 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, "nft_collection_address": self.get_token_contract_internal(collection_id, token_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, "nft_collection_address": self.get_token_contract_internal(collection_id, token_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_token_contract(self, collection_id: str, token_id: str) -> str: self.require_token_owner(collection_id, token_id) return self.get_token_contract_internal(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: return self.dumps(self.require_listing(listing_id)) @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: 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: 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", "")) nft_collection_address = str(listing.get("nft_collection_address", "")) 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, "nft_collection_address": nft_collection_address, "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 get_token_contract_internal(self, collection_id: str, token_id: str) -> str: key = self.token_key(collection_id, token_id) try: value = self.token_contracts[key] if isinstance(value, str): return value except Exception: pass return "" 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=(",", ":"))