import { DatabaseSync } from 'node:sqlite'; import { randomUUID } from 'node:crypto'; import { mkdir, chmod } from 'node:fs/promises'; import { join } from 'node:path'; import { Worker } from 'node:worker_threads'; import { NETWORK, CONSENSUS, nftTransferMessage, testnetOwnerOf, transactionCardIds, type TestnetSnapshot, type NftTransfer } from '../../shared/testnet.ts'; import { ADDITIONAL_MARKET_FEE, marketCommandSchema, marketMessage, type MarketCommand, type MarketListing, type MarketOrder, type MarketInfo } from '../../shared/market.ts'; import { isXmrListing, isXmrOrder, marketCommandV3Schema, marketMessageV3, type MarketActivityV3, type MarketCommandV3, type MarketListingRecord, type MarketListingV3, type MarketOrderRecord, type MarketOrderV3 } from '../../shared/market-v3.ts'; import { CHAIN_ID, GENESIS_HASH, transactionId, transferSchema, verifySignature } from '../testnet/consensus.ts'; import { mockQuote, mockXmrQuote, paymentMode } from './payments.ts'; import { RpcError } from '../resources.ts'; import type { CanonicalEvents, IndexedNft } from '../testnet/canonical-index.ts'; import { z } from 'zod'; import { operatorMarketPolicy, resolvedMarketPolicy, permitsMarketAction, validateMarketPolicy, type MarketPolicy } from './policy.ts'; import { PROFILE } from '../../shared/network.ts'; type Chain = { snapshot(): TestnetSnapshot; summary?(): { tip: string; revision: number; height?: number }; submitTransfer(raw: unknown): Promise; nft?(token: string): IndexedNft | undefined; canonicalTransaction?(id: string): { height: number; blockHash: string } | undefined; canonicalHead?(): number; canonicalEvents?(after: number, limit?: number): CanonicalEvents; pendingTransactions?(): TestnetSnapshot['pendingTransfers']; reconciliationDatabasePath?(): string }; type ChainView = { height: number; pendingTransfers: TestnetSnapshot['pendingTransfers']; nft(token: string): IndexedNft | undefined; transaction(id: string): { height: number; blockHash: string } | undefined }; type Config = { feeBps: number; confirmations: number } & MarketPolicy; export function marketConfig(env: NodeJS.ProcessEnv = process.env): Config { paymentMode(env); const integer = (name: string, fallback: number, min: number, max: number) => { const text = env[name] ?? String(fallback); if (!/^\d+$/.test(text)) throw new Error(`Invalid ${name}`); const n = Number(text); if (!Number.isSafeInteger(n) || n < min || n > max) throw new Error(`Invalid ${name}`); return n; }; return { ...operatorMarketPolicy(env), feeBps: integer('WIKIDECK_MARKET_FEE_BPS', 250, 0, 5000), confirmations: integer('WIKIDECK_MARKET_CONFIRMATIONS', 10, 2, 1000) }; } const PAYMENT_WINDOW = 30 * 60_000; const DELIVERY_WINDOW = 48 * 60 * 60_000; const RECONCILIATION_FRESH_MS = 15_000; const cursorSchema = z.object({ version: z.literal(1), streamId: z.string().uuid(), cursor: z.number().int().nonnegative().safe(), anchor: z.string().regex(/^[a-f0-9]{64}$/) }).strict(); const canonicalEventSchema = z.object({ sequence: z.number().int().positive().safe(), type: z.enum(['connect', 'disconnect']), height: z.number().int().positive().safe(), hash: z.string().regex(/^[a-f0-9]{64}$/), previousHash: z.string().regex(/^[a-f0-9]{64}$/), transactionIds: z.array(z.string().regex(/^[a-f0-9]{64}$/)).max(64), eventHash: z.string().regex(/^[a-f0-9]{64}$/) }).strict(); const canonicalEventPageSchema = z.object({ streamId: z.string().uuid(), previousAnchor: z.string().regex(/^[a-f0-9]{64}$/), anchor: z.string().regex(/^[a-f0-9]{64}$/), events: z.array(canonicalEventSchema).max(64), cursor: z.number().int().nonnegative().safe(), head: z.number().int().nonnegative().safe(), caughtUp: z.boolean() }).strict(); const reconciliationJobSchema = z.object({ version: z.literal(1), stage: z.enum(['orders', 'listings', 'finalize']), afterId: z.string().nullable(), observedAt: z.number().int().nonnegative().safe(), tip: z.string().regex(/^[a-f0-9]{64}$/).nullable(), revision: z.number().int().nonnegative().safe().nullable(), eventPage: canonicalEventPageSchema.nullable() }).strict(); // Keep each synchronous SQLite slice small enough to limit request-loop stalls. export const MARKET_RECONCILIATION_BATCH_SIZE = 32; const RECONCILIATION_JOB_KEY = 'canonical_reconciliation_job'; /** Node-local marketplace, NOT consensus escrow. Its SQLite journal is backed up with the ledger. */ export class Market { private queue: Promise = Promise.resolve(); private queued = 0; private reconciled: { tip: string; revision: number; at: number } | undefined; private eventCursor = 0; private eventCheckpoint: z.infer | undefined; private eventBacklog = false; private reconciliationJob: z.infer | undefined; private readonly reconciliationWorker?: Worker; private readonly workerReadsLedger: boolean; private workerRequestId = 0; private readonly workerRequests = new Map): void; reject(error: Error): void }>(); private reconciliationWorkerBusy = false; private reconciliationWorkerError: string | undefined; private readonly db: DatabaseSync; private readonly chain: Chain; readonly config: Config; readonly id: string; private readonly now: () => number; private constructor(db: DatabaseSync, chain: Chain, config: Config, id: string, now: () => number, worker?: Worker) { this.db = db; this.chain = chain; this.config = config; this.id = id; this.now = now; this.reconciliationWorker = worker; this.workerReadsLedger = !!chain.reconciliationDatabasePath; worker?.on('message', (message: { id: number; ok: boolean; error?: string }) => { const pending = this.workerRequests.get(message.id); if (!pending) return; this.workerRequests.delete(message.id); if (message.ok) pending.resolve(message); else pending.reject(new Error(message.error ?? 'Reconciliation worker failed')); }); worker?.on('error', error => { this.reconciliationWorkerError = 'Reconciliation worker failed'; for (const pending of this.workerRequests.values()) pending.reject(error); this.workerRequests.clear(); }); worker?.on('exit', code => { if (!this.workerRequests.size) return; const error = new Error(`Reconciliation worker exited with code ${code}`); this.reconciliationWorkerError = 'Reconciliation worker stopped unexpectedly'; for (const pending of this.workerRequests.values()) pending.reject(error); this.workerRequests.clear(); }); const cursor = db.prepare("SELECT value FROM market_meta WHERE key='canonical_event_cursor'").get() as { value: string } | undefined; if (cursor) { this.eventCheckpoint = cursorSchema.parse(JSON.parse(cursor.value)); this.eventCursor = this.eventCheckpoint.cursor; } const job = db.prepare('SELECT value FROM market_meta WHERE key=?').get(RECONCILIATION_JOB_KEY) as { value: string } | undefined; if (job) { this.reconciliationJob = reconciliationJobSchema.parse(JSON.parse(job.value)); this.eventBacklog = true; } else this.eventBacklog = !!chain.canonicalHead && this.eventCursor < chain.canonicalHead(); } static async open(directory: string, chain: Chain, config = marketConfig(), now = Date.now, reconciliation: { worker?: boolean } = {}): Promise { await validateMarketPolicy(config); const folder = join(directory, PROFILE); await mkdir(folder, { recursive: true, mode: 0o700 }); const file = join(folder, 'market.sqlite'); const db = new DatabaseSync(file); let worker: Worker | undefined; try { db.exec(`PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL; PRAGMA busy_timeout=5000; CREATE TABLE IF NOT EXISTS market_meta(key TEXT PRIMARY KEY, value TEXT NOT NULL); CREATE TABLE IF NOT EXISTS listings(id TEXT PRIMARY KEY, card TEXT NOT NULL, status TEXT NOT NULL, data TEXT NOT NULL); CREATE UNIQUE INDEX IF NOT EXISTS one_active_listing ON listings(card) WHERE status IN ('open','reserved'); CREATE TABLE IF NOT EXISTS orders(id TEXT PRIMARY KEY, data TEXT NOT NULL); CREATE TABLE IF NOT EXISTS receipts(id TEXT PRIMARY KEY, digest TEXT NOT NULL, expires INTEGER NOT NULL, data TEXT NOT NULL);`); db.exec(`CREATE INDEX IF NOT EXISTS listings_by_status ON listings(status); CREATE INDEX IF NOT EXISTS listings_by_seller ON listings(json_extract(data,'$.seller')); CREATE INDEX IF NOT EXISTS orders_by_seller ON orders(json_extract(data,'$.seller')); CREATE INDEX IF NOT EXISTS orders_by_buyer ON orders(json_extract(data,'$.buyer'));`); const meta = db.prepare('SELECT value FROM market_meta WHERE key=?').get('identity') as { value: string } | undefined; let id: string; if (meta) { const identity = JSON.parse(meta.value); if (identity.network !== NETWORK || identity.chainId !== CHAIN_ID || identity.version !== 1 || typeof identity.id !== 'string') throw new Error('Marketplace network/schema mismatch'); id = identity.id; } else { id = randomUUID(); db.prepare('INSERT INTO market_meta VALUES(?,?)').run('identity', JSON.stringify({ version: 1, network: NETWORK, chainId: CHAIN_ID, id })); } if (process.platform !== 'win32') await chmod(file, 0o600); worker = reconciliation.worker ? new Worker(new URL('./reconciliation-commit-worker.js', import.meta.url), { workerData: { marketFile: file, ledgerFile: chain.reconciliationDatabasePath?.() ?? null }, execArgv: [] }) : undefined; worker?.unref(); return new Market(db, chain, config, id, now, worker); } catch (error) { if (worker) await worker.terminate(); db.close(); throw error; } } async close(): Promise { if (this.reconciliationWorker) await this.reconciliationWorker.terminate(); this.db.close(); } isReconciliationWorkerBusy(): boolean { return this.reconciliationWorkerBusy; } isReconciliationWorkerFailed(): boolean { return !!this.reconciliationWorkerError; } diagnostics() { const states=this.db.prepare("SELECT json_extract(data,'$.status') AS status,COUNT(*) AS count FROM orders GROUP BY json_extract(data,'$.status')").all(); return { queued:this.queued,lastReconciledAt:this.reconciled?.at??null,orderStates:states, canonicalEventCursor:this.eventCursor,canonicalEventHead:this.chain.canonicalHead?.()??null,eventBacklog:this.eventBacklog, reconciliationWorkerBusy:this.reconciliationWorkerBusy,reconciliationWorkerFailed:!!this.reconciliationWorkerError }; } hasReconciliationBacklog(): boolean { return this.eventBacklog || !!this.reconciliationJob; } info(): MarketInfo { return { network: NETWORK, chainId: CHAIN_ID, marketId: this.id, paymentMode: 'mock', marketVersion: 3, feeBps: this.config.feeBps, ...resolvedMarketPolicy(this.config), additionalFee: { ...ADDITIONAL_MARKET_FEE }, deliveryConfirmations: this.config.confirmations, serverTime: this.now() }; } private nextReconciliationRows(table: 'listings' | 'orders', afterId: string | null): { id: string; data: string }[] { return this.db.prepare(`SELECT id,data FROM ${table} WHERE (? IS NULL OR id>?) ORDER BY id LIMIT ?`) .all(afterId, afterId, MARKET_RECONCILIATION_BATCH_SIZE) as { id: string; data: string }[]; } private listing(id: string): MarketListingRecord { const row = this.db.prepare('SELECT data FROM listings WHERE id=?').get(id) as { data: string } | undefined; if (!row) throw new Error('Listing not found'); return JSON.parse(row.data) as MarketListingRecord; } private order(id: string): MarketOrderRecord { const row = this.db.prepare('SELECT data FROM orders WHERE id=?').get(id) as { data: string } | undefined; if (!row) throw new Error('Order not found'); return JSON.parse(row.data) as MarketOrderRecord; } private saveListing(l: MarketListingRecord) { this.db.prepare('INSERT INTO listings VALUES(?,?,?,?) ON CONFLICT(id) DO UPDATE SET status=excluded.status,data=excluded.data WHERE listings.data<>excluded.data') .run(l.id, l.cardHash, l.status, JSON.stringify(l)); } private saveOrder(o: MarketOrderRecord) { this.db.prepare('INSERT INTO orders VALUES(?,?) ON CONFLICT(id) DO UPDATE SET data=excluded.data WHERE orders.data<>excluded.data').run(o.id, JSON.stringify(o)); } private atomic(fn: () => T): T { this.db.exec('BEGIN IMMEDIATE'); try { const result = fn(); this.db.exec('COMMIT'); return result; } catch (error) { this.db.exec('ROLLBACK'); throw error; } } private view(): ChainView { if (this.chain.nft && this.chain.canonicalTransaction && this.chain.pendingTransactions && this.chain.summary) { const height = this.chain.summary().height; if (height === undefined) throw new Error('Indexed chain height missing'); return { height, pendingTransfers: this.chain.pendingTransactions(), nft: token => this.chain.nft!(token), transaction: id => this.chain.canonicalTransaction!(id) }; } // Compatibility boundary for unit-test/legacy adapters only. The actual node always supplies indexed queries. const s = this.chain.snapshot(); const txs = new Map(s.blocks.flatMap(b => b.transactions.map(tx => [transactionId(tx), { height: b.height, blockHash: b.hash }] as const))); return { height: s.blocks.length, pendingTransfers: s.pendingTransfers, transaction: id => txs.get(id), nft: token => { const block = s.blocks.find(b => b.tokenId === token); return block ? { block, owner: testnetOwnerOf(s, token)!, sequence: s.transfers.filter(tx => transactionCardIds(tx).includes(token)).length, confirmations: s.blocks.length + 1 - block.height } : undefined; } }; } private card(view: ChainView, hash: string, owner: string, sequence: number): boolean { const nft = view.nft(hash); return !!nft && nft.owner === owner && nft.sequence === sequence && nft.confirmations >= CONSENSUS.coinbaseMaturity; } private deliveryId(order: MarketOrderRecord): string { // Identity excludes signature bytes. Recognize exact sale delivery through ANY submission route, // including peer gossip or a direct /api/transfer, even before the market records delivery intent. return transactionId({ network: NETWORK, chainId: CHAIN_ID, cardHash: order.cardHash, from: order.seller, to: order.buyer, sequence: order.sequence, publicKey: '', signature: '' }); } private workerRequest(request: Record): Promise> { const worker = this.reconciliationWorker; if (!worker) return Promise.reject(new Error('Reconciliation worker is unavailable')); const id = ++this.workerRequestId; return new Promise((resolve, reject) => { this.workerRequests.set(id, { resolve, reject }); try { worker.postMessage({ ...request, id }); } catch (error) { this.workerRequests.delete(id); reject(error instanceof Error ? error : new Error('Worker request failed')); } }); } private scheduleWorkerContinuation(): void { setImmediate(() => { if (!this.reconciliationWorkerBusy && this.hasReconciliationBacklog()) this.reconcile(); }).unref(); } private finalizeWorkerJob(job: z.infer): void { const summary = this.chain.summary?.(), eventPage = job.eventPage as CanonicalEvents | null; const currentHead = this.chain.canonicalHead?.() ?? eventPage?.head ?? 0; if ((summary && (job.tip !== summary.tip || job.revision !== summary.revision)) || (eventPage && currentHead > eventPage.head)) { const events = this.chain.canonicalEvents?.(this.eventCursor, 64); if (events && this.eventCheckpoint && (events.streamId !== this.eventCheckpoint.streamId || events.previousAnchor !== this.eventCheckpoint.anchor)) throw new RpcError(409, 'Canonical event history changed; restore matching ledger and market backups.'); const restarted = reconciliationJobSchema.parse({ version: 1, stage: 'orders', afterId: null, observedAt: this.now(), tip: summary?.tip ?? null, revision: summary?.revision ?? null, eventPage: events ?? null }); this.atomic(() => this.db.prepare('UPDATE market_meta SET value=? WHERE key=?') .run(JSON.stringify(restarted), RECONCILIATION_JOB_KEY)); this.reconciliationJob = restarted; this.eventBacklog = true; this.reconciliationWorkerError = undefined; this.scheduleWorkerContinuation(); return; } if (this.reconciliationWorkerBusy) return; this.reconciliationWorkerBusy = true; const cursor = eventPage ? JSON.stringify({ version: 1, streamId: eventPage.streamId, cursor: eventPage.cursor, anchor: eventPage.anchor }) : null; void this.workerRequest({ operation: 'finalize', expectedJob: job, cursor, observedAt: job.observedAt }) .then(() => { this.reconciliationJob = undefined; if (eventPage) { this.eventCursor = eventPage.cursor; this.eventCheckpoint = { version: 1, streamId: eventPage.streamId, cursor: eventPage.cursor, anchor: eventPage.anchor }; } const current = this.chain.summary?.(); const head = this.chain.canonicalHead?.() ?? eventPage?.head ?? 0; this.eventBacklog = (!!eventPage && (!eventPage.caughtUp || head > eventPage.cursor)) || (!!current && (current.tip !== job.tip || current.revision !== job.revision)); if (current && !this.eventBacklog) this.reconciled = { ...current, at: this.now() }; this.reconciliationWorkerError = undefined; }) .catch(error => { this.reconciliationWorkerError = error instanceof Error ? error.message : 'Worker finalize failed'; this.eventBacklog = true; }) .finally(() => { this.reconciliationWorkerBusy = false; if (!this.reconciliationWorkerError && this.hasReconciliationBacklog()) this.scheduleWorkerContinuation(); }); } private dispatchWorkerBatch(job: z.infer): void { if (!this.reconciliationWorker || this.reconciliationWorkerBusy || job.stage === 'finalize') return; this.reconciliationWorkerBusy = true; try { let fallbackFacts: Record | null = null; if (!this.workerReadsLedger) { const rows = this.nextReconciliationRows(job.stage, job.afterId), s = this.view(); const pendingTransactions = s.pendingTransfers.map(tx => ({ id: transactionId(tx), cardIds: transactionCardIds(tx) })); const pendingIds = pendingTransactions.map(tx => tx.id); const pendingCardIds = [...new Set(pendingTransactions.flatMap(tx => tx.cardIds))]; const deliveryIds: Record = Object.create(null), targetCards = new Set(); if (job.stage === 'orders') for (const row of rows) { const order = JSON.parse(row.data) as MarketOrderRecord; deliveryIds[order.id] = this.deliveryId(order); targetCards.add(order.cardHash); } else for (const row of rows) targetCards.add((JSON.parse(row.data) as MarketListingRecord).cardHash); const deliveryTransactions: Record = Object.create(null); const nfts: Record = Object.create(null); if (job.stage === 'orders') for (const id of Object.values(deliveryIds)) deliveryTransactions[id] = s.transaction(id) ?? null; for (const card of targetCards) { const nft = s.nft(card); nfts[card] = nft ? { owner: nft.owner, sequence: nft.sequence, confirmations: nft.confirmations } : null; } fallbackFacts = { height: s.height, pendingTransactions, pendingIds, pendingCardIds, deliveryIds, deliveryTransactions, nfts }; } const disconnectedIds = job.eventPage?.events.filter(event => event.type === 'disconnect').flatMap(event => event.transactionIds) ?? []; void this.workerRequest({ operation: 'reconcile-batch', expectedJob: job, stage: job.stage, afterId: job.afterId, batchSize: MARKET_RECONCILIATION_BATCH_SIZE, fallbackFacts, genesisHash: GENESIS_HASH, network: NETWORK, chainId: CHAIN_ID, disconnectedIds, maturity: CONSENSUS.coinbaseMaturity, additionalFee: ADDITIONAL_MARKET_FEE, paymentWindow: PAYMENT_WINDOW }) .then(message => { const nextJob = reconciliationJobSchema.parse(message.nextJob); this.reconciliationJob = nextJob; this.reconciliationWorkerError = undefined; this.eventBacklog = true; if (this.workerReadsLedger) this.refenceWorkerSnapshot(job, message.snapshot as { tip: string; eventHead: number; pendingDigest: string }); }) .catch(error => { this.reconciliationWorkerError = error instanceof Error ? error.message : 'Worker reconciliation failed'; this.eventBacklog = true; }) .finally(() => { this.reconciliationWorkerBusy = false; if (!this.reconciliationWorkerError) this.scheduleWorkerContinuation(); }); } catch (error) { this.reconciliationWorkerBusy = false; this.reconciliationWorkerError = error instanceof Error ? error.message : 'Worker reconciliation preparation failed'; this.eventBacklog = true; } } private refenceWorkerSnapshot(job: z.infer, snapshot: { tip: string; eventHead: number; pendingDigest: string }): void { const summary = this.chain.summary?.(); if (!summary) return; const pendingDigest = this.chain.pendingTransactions ? JSON.stringify(this.chain.pendingTransactions().map(transactionId).sort()) : ''; const pageHead = job.eventPage?.head ?? snapshot.eventHead; if (summary.tip === snapshot.tip && (job.revision === null || summary.revision === job.revision) && snapshot.eventHead <= pageHead && pendingDigest === snapshot.pendingDigest) return; const events = this.chain.canonicalEvents?.(this.eventCursor, 64); if (events && this.eventCheckpoint && (events.streamId !== this.eventCheckpoint.streamId || events.previousAnchor !== this.eventCheckpoint.anchor)) throw new RpcError(409, 'Canonical event history changed; restore matching ledger and market backups.'); const restarted = reconciliationJobSchema.parse({ version: 1, stage: 'orders', afterId: null, observedAt: this.now(), tip: summary.tip, revision: summary.revision, eventPage: events ?? null }); this.atomic(() => this.db.prepare('UPDATE market_meta SET value=? WHERE key=?') .run(JSON.stringify(restarted), RECONCILIATION_JOB_KEY)); this.reconciliationJob = restarted; this.eventBacklog = true; } /** Reconcile one durable keyset batch from canonical transaction identity, never mempool acknowledgement. */ private reconcile() { if (this.reconciliationWorkerBusy || this.reconciliationWorkerError) return; const now = this.now(); const summary = this.chain.summary?.(); if (!this.reconciliationJob) { const events = this.chain.canonicalEvents?.(this.eventCursor, 64); if (events && this.eventCheckpoint && (events.streamId !== this.eventCheckpoint.streamId || events.previousAnchor !== this.eventCheckpoint.anchor)) throw new RpcError(409, 'Canonical event history changed; restore matching ledger and market backups.'); const job = reconciliationJobSchema.parse({ version: 1, stage: 'orders', afterId: null, observedAt: now, tip: summary?.tip ?? null, revision: summary?.revision ?? null, eventPage: events ?? null }); this.atomic(() => this.db.prepare('INSERT INTO market_meta(key,value) VALUES(?,?)') .run(RECONCILIATION_JOB_KEY, JSON.stringify(job))); this.reconciliationJob = job; this.eventBacklog = true; } let job = this.reconciliationJob; if (summary && (job.tip !== summary.tip || job.revision !== summary.revision)) { // Canonical state changed while a multi-batch scan was running. Rewind the // keyset and bind a fresh event page so every row is checked against one // observed chain revision before the canonical cursor can advance. const events = this.chain.canonicalEvents?.(this.eventCursor, 64); if (events && this.eventCheckpoint && (events.streamId !== this.eventCheckpoint.streamId || events.previousAnchor !== this.eventCheckpoint.anchor)) throw new RpcError(409, 'Canonical event history changed; restore matching ledger and market backups.'); job = reconciliationJobSchema.parse({ version: 1, stage: 'orders', afterId: null, observedAt: now, tip: summary.tip, revision: summary.revision, eventPage: events ?? null }); this.atomic(() => this.db.prepare('UPDATE market_meta SET value=? WHERE key=?') .run(JSON.stringify(job), RECONCILIATION_JOB_KEY)); this.reconciliationJob = job; this.eventBacklog = true; } if (job.stage === 'finalize') { if (this.reconciliationWorker) { this.finalizeWorkerJob(job); return; } const eventPage = job.eventPage as CanonicalEvents | null; this.atomic(() => { this.db.prepare('DELETE FROM receipts WHERE expires < ?').run(job.observedAt - 86400_000); if (eventPage) this.db.prepare('INSERT INTO market_meta(key,value) VALUES(?,?) ON CONFLICT(key) DO UPDATE SET value=excluded.value') .run('canonical_event_cursor', JSON.stringify({ version: 1, streamId: eventPage.streamId, cursor: eventPage.cursor, anchor: eventPage.anchor })); this.db.prepare('DELETE FROM market_meta WHERE key=?').run(RECONCILIATION_JOB_KEY); }); this.reconciliationJob = undefined; if (eventPage) { this.eventCursor = eventPage.cursor; this.eventCheckpoint = { version: 1, streamId: eventPage.streamId, cursor: eventPage.cursor, anchor: eventPage.anchor }; } const head = this.chain.canonicalHead?.() ?? eventPage?.head ?? 0; this.eventBacklog = !!eventPage && (!eventPage.caughtUp || head > eventPage.cursor); const current = this.chain.summary?.(); if (current && !this.eventBacklog) this.reconciled = { ...current, at: now }; return; } if (this.reconciliationWorker) { this.dispatchWorkerBatch(job); return; } const s = this.view(), pendingIds = new Set(s.pendingTransfers.map(transactionId)); const eventPage = job.eventPage as CanonicalEvents | null; const disconnected = new Set(eventPage?.events.filter(e => e.type === 'disconnect').flatMap(e => e.transactionIds) ?? []); const rows = this.nextReconciliationRows(job.stage, job.afterId); let nextJob: z.infer | undefined; let completed = false; this.atomic(() => { if (job.stage === 'orders' && rows.length) { for (const row of rows) { const o = JSON.parse(row.data) as MarketOrderRecord; const l = this.listing(o.listingId); // A disconnect of completed delivery is sticky even if the same transaction has reappeared. if (o.status === 'complete' && disconnected.has(this.deliveryId(o))) { o.status = 'review'; o.payout = 'review'; if (o.additionalFee) o.additionalFee.payout = 'review'; this.saveOrder(o); } if (o.status === 'refunded' || o.status === 'expired') continue; const deliveryId = this.deliveryId(o); const delivery = s.transaction(deliveryId); const confirmations = delivery ? s.height - delivery.height + 1 : 0; o.deliveryConfirmations = confirmations; if (o.status === 'complete' || o.status === 'review') { if (!delivery || confirmations < o.requiredConfirmations) { o.status = 'review'; o.payout = 'review'; if (o.additionalFee) o.additionalFee.payout = 'review'; } this.saveOrder(o); continue; // Never automatically pay a second time after a reorganization. } if (delivery && o.status === 'awaiting-payment') { o.status = 'review'; o.payout = 'review'; if (o.additionalFee) o.additionalFee.payout = 'review'; } else if (delivery) { o.transferId = deliveryId; o.status = 'delivery-pending'; if (confirmations >= o.requiredConfirmations) { o.status = 'complete'; o.payout = 'mock-paid'; l.status = 'sold'; if (o.additionalFee) o.additionalFee.payout = 'mock-paid'; } } else if (o.status === 'awaiting-payment' && job.observedAt >= o.payBy) { o.status = 'expired'; l.status = l.kind === 'fixed' ? 'open' : 'expired'; } else if (o.status !== 'awaiting-payment') { const pending = pendingIds.has(deliveryId); const conflict = s.pendingTransfers.some(tx => transactionCardIds(tx).includes(o.cardHash) && transactionId(tx) !== deliveryId); if (!this.card(s, o.cardHash, o.seller, o.sequence) || conflict || (job.observedAt >= o.deliverBy && !pending)) o.status = 'refund-required'; else if (o.status === 'delivery-pending' && !pending) o.status = 'paid'; } this.saveOrder(o); this.saveListing(l); } } else if (job.stage === 'listings' && rows.length) { for (const row of rows) { const l = JSON.parse(row.data) as MarketListingRecord; if (l.status !== 'open') continue; if (l.additionalFee?.bps !== ADDITIONAL_MARKET_FEE.bps || l.additionalFee?.address !== ADDITIONAL_MARKET_FEE.address || !this.card(s, l.cardHash, l.seller, l.sequence) || s.pendingTransfers.some(tx => transactionCardIds(tx).includes(l.cardHash))) l.status = 'invalid'; else if (l.kind === 'auction' && job.observedAt >= l.endsAt! + PAYMENT_WINDOW) l.status = 'expired'; else if (l.kind === 'auction' && job.observedAt >= l.endsAt! && !l.highestBid) l.status = 'expired'; this.saveListing(l); } } if (job.stage === 'orders') { if (rows.length < MARKET_RECONCILIATION_BATCH_SIZE) nextJob = { ...job, stage: 'listings', afterId: null }; else nextJob = { ...job, afterId: rows.at(-1)!.id }; } else { if (rows.length < MARKET_RECONCILIATION_BATCH_SIZE) { this.db.prepare('DELETE FROM receipts WHERE expires < ?').run(job.observedAt - 86400_000); if (eventPage) { this.db.prepare('INSERT INTO market_meta(key,value) VALUES(?,?) ON CONFLICT(key) DO UPDATE SET value=excluded.value') .run('canonical_event_cursor', JSON.stringify({ version: 1, streamId: eventPage.streamId, cursor: eventPage.cursor, anchor: eventPage.anchor })); } this.db.prepare('DELETE FROM market_meta WHERE key=?').run(RECONCILIATION_JOB_KEY); completed = true; } else nextJob = { ...job, afterId: rows.at(-1)!.id }; } if (nextJob) this.db.prepare('UPDATE market_meta SET value=? WHERE key=?').run(JSON.stringify(nextJob), RECONCILIATION_JOB_KEY); }); if (completed) { this.reconciliationJob = undefined; if (eventPage) { this.eventCursor = eventPage.cursor; this.eventCheckpoint = { version: 1, streamId: eventPage.streamId, cursor: eventPage.cursor, anchor: eventPage.anchor }; // The ledger can append events while large journals are processed over many bounded calls. // Do not mark the market reconciled to a newer tip until those post-page events are consumed. const currentHead = this.chain.canonicalHead?.() ?? eventPage.head; this.eventBacklog = !eventPage.caughtUp || currentHead > eventPage.cursor; } else this.eventBacklog = false; const current = this.chain.summary?.(); if (current && !this.eventBacklog) this.reconciled = { ...current, at: now }; } else { this.reconciliationJob = nextJob; this.eventBacklog = true; } // Empty/short order pages can advance to the listing stage in the same bounded call. This // avoids rejecting the first command on an empty market while still capping work at 2×128 rows. if (!completed && job.stage === 'orders' && rows.length < MARKET_RECONCILIATION_BATCH_SIZE) this.reconcile(); } browse(offset = 0): { listings: MarketListingRecord[]; next: number | null; info: MarketInfo } { if (!Number.isSafeInteger(offset) || offset < 0 || offset > 10000) throw new Error('Invalid marketplace page'); // Reconciliation must not interleave with delivery submission while its chain await is outstanding. const summary = this.chain.summary?.(); const last = this.reconciled; if (!this.queued && (this.hasReconciliationBacklog() || !summary || !last || summary.tip !== last.tip || summary.revision !== last.revision || this.now() - last.at >= RECONCILIATION_FRESH_MS)) this.reconcile(); if (!resolvedMarketPolicy(this.config).marketEnabled) return { listings: [], next: null, info: this.info() }; const rows = this.db.prepare("SELECT data FROM listings WHERE status IN ('open','reserved') ORDER BY rowid DESC LIMIT 25 OFFSET ?") .all(offset) as { data: string }[]; return { listings: rows.slice(0,24).map(r=>JSON.parse(r.data)), next: rows.length>24?offset+24:null, info: this.info() }; } execute(raw: unknown): Promise { const current = marketCommandV3Schema.safeParse(raw); const cmd = current.success ? current.data : marketCommandSchema.parse(raw); if (this.queued >= 64) return Promise.reject(new Error('Marketplace busy; retry shortly')); this.queued++; const result = this.queue.then(async () => { try { return await this.run(cmd); } finally { this.queued--; } }); this.queue = result.catch(() => undefined); return result; } private async run(c: MarketCommand | MarketCommandV3): Promise { const now = this.now(); const isV3 = 'version' in c; const message = isV3 ? marketMessageV3(c) : marketMessage(c); if (c.chainId !== CHAIN_ID || c.marketId !== this.id || c.expiresAt < now || c.expiresAt > now + 5 * 60_000 || !verifySignature(c.publicKey, c.actor, message, c.signature)) throw new Error('Invalid or expired marketplace signature'); const key = `${c.actor}:${c.nonce}`, digest = message; const old = this.db.prepare('SELECT digest,data FROM receipts WHERE id=?').get(key) as { digest: string; data: string } | undefined; if (old) { if (old.digest !== digest) throw new Error('Nonce already used for different terms'); return JSON.parse(old.data); } if (!permitsMarketAction(this.config, c.action.type)) throw new Error('New market trading is disabled by this host; existing activity and resolution remain available'); const a = c.action; const summary = this.chain.summary?.(), last = this.reconciled; const stale = !summary || !last || summary.tip !== last.tip || summary.revision !== last.revision || this.now() - last.at >= RECONCILIATION_FRESH_MS; if (this.hasReconciliationBacklog() || stale || (!this.reconciliationWorker && a.type !== 'activity')) this.reconcile(); if (this.eventBacklog && a.type !== 'activity') throw new RpcError(503, 'Market canonical reconciliation is catching up; retry shortly.'); if (a.type === 'activity') return { orders: (this.db.prepare("SELECT data FROM orders WHERE json_extract(data,'$.seller')=? OR json_extract(data,'$.buyer')=? ORDER BY rowid DESC") .all(c.actor,c.actor) as { data: string }[]).map(r=>JSON.parse(r.data) as MarketOrderRecord), listings: (this.db.prepare("SELECT data FROM listings WHERE json_extract(data,'$.seller')=? ORDER BY rowid DESC") .all(c.actor) as { data: string }[]).map(r=>JSON.parse(r.data) as MarketListingRecord) }; if (a.type === 'deliver') { const o = this.order(a.orderId), tx = transferSchema.parse(a.transfer); if (isV3 && !isXmrOrder(o) || !isV3 && isXmrOrder(o) || o.seller !== c.actor || !['paid', 'delivery-pending'].includes(o.status)) throw new Error('Only the paid order seller can deliver under the order terms'); if (tx.cardHash !== o.cardHash || tx.sequence !== o.sequence || tx.from !== o.seller || tx.to !== o.buyer || !verifySignature(tx.publicKey, tx.from, nftTransferMessage(tx), tx.signature)) throw new Error('Delivery does not match this sale'); // Persist intent before the ledger await. A crash/restart reconciles by deterministic tx ID. o.transfer = tx; o.transferId = transactionId(tx); o.status = 'delivery-pending'; this.saveOrder(o); await this.chain.submitTransfer(tx); return this.receipt(key, digest, c.expiresAt, o); } return this.atomic(() => { const result = isV3 ? this.mutateV3(c) : this.mutateV2(c); return this.receipt(key, digest, c.expiresAt, result); }); } private receipt(key: string, digest: string, expires: number, value: T): T { this.db.prepare('INSERT INTO receipts VALUES(?,?,?,?)').run(key, digest, expires, JSON.stringify(value)); return value; } private mutateV2(c: MarketCommand): MarketActivityV3 | MarketListingRecord | MarketOrderRecord { const a = c.action, now = this.now(), s = this.view(); if (a.type === 'list') { throw new Error('USD listing terms are retired; create a new native-XMR listing'); } if (a.type === 'cancel') { const l = this.listing(a.listingId); if (isXmrListing(l) || l.status !== 'open') throw new Error('USD legacy listing cannot be changed under this command'); if (l.seller !== c.actor || l.highestBid) throw new Error('Only the seller can cancel, and auctions with bids cannot be cancelled'); l.status = 'cancelled'; this.saveListing(l); return l; } if (a.type === 'mock-pay' || a.type === 'refund') { const o = this.order(a.orderId); if (isXmrOrder(o)) throw new Error('Use native-XMR signed terms for this order'); if (o.buyer !== c.actor) throw new Error('Only the order buyer can perform this action'); if (a.type === 'mock-pay') { if (o.status !== 'awaiting-payment' || now >= o.payBy) throw new Error('Invoice is not payable'); if (!this.card(s, o.cardHash, o.seller, o.sequence) || s.pendingTransfers.some(tx => transactionCardIds(tx).includes(o.cardHash))) throw new Error('Seller card changed; do not pay'); o.status = 'paid'; o.deliverBy = now + DELIVERY_WINDOW; } else { if (o.status !== 'refund-required' || s.pendingTransfers.some(tx => transactionCardIds(tx).includes(o.cardHash)) || s.transaction(this.deliveryId(o))) throw new Error('Refund requires failed delivery and no pending/canonical sale transfer'); o.status = 'refunded'; const l = this.listing(o.listingId); l.status = 'cancelled'; this.saveListing(l); } this.saveOrder(o); return o; } if (a.type === 'activity') return { orders: (this.db.prepare("SELECT data FROM orders WHERE json_extract(data,'$.seller')=? OR json_extract(data,'$.buyer')=? ORDER BY rowid DESC") .all(c.actor,c.actor) as { data: string }[]).map(r=>JSON.parse(r.data) as MarketOrderRecord), listings: (this.db.prepare("SELECT data FROM listings WHERE json_extract(data,'$.seller')=? ORDER BY rowid DESC") .all(c.actor) as { data: string }[]).map(r=>JSON.parse(r.data) as MarketListingRecord) }; if (a.type === 'deliver') throw new Error('Legacy delivery is handled by the common signed-order delivery path'); throw new Error('Legacy USD bids and checkout are retired; only already-created orders may proceed'); } private mutateV3(c: MarketCommandV3): MarketActivityV3 | MarketListingRecord | MarketOrderRecord { const a = c.action, now = this.now(), s = this.view(); if (a.type === 'list') { if (a.feeBps !== this.config.feeBps) throw new Error('Market fee changed; review and sign again'); const hostFeeAddress = resolvedMarketPolicy(this.config).hostFeeAddress; if ((a.hostFeeAddress ?? ADDITIONAL_MARKET_FEE.address) !== hostFeeAddress) throw new Error('Host fee recipient changed; review and sign again'); if (!this.card(s, a.cardHash, c.actor, a.sequence) || s.pendingTransfers.some(tx => transactionCardIds(tx).includes(a.cardHash))) throw new Error('Card is missing, stale, immature, or already pending'); if (a.kind === 'fixed' ? a.endsAt !== null : a.endsAt === null || a.endsAt < now + 60_000 || a.endsAt > now + 30 * 86400_000) throw new Error('Auction duration must be 1 minute to 30 days; fixed listings have no end'); const activeListings = this.db.prepare(`SELECT COUNT(*) AS n FROM listings WHERE json_extract(data,'$.seller')=? AND status IN ('open','reserved')`).get(c.actor) as { n: number }; if (activeListings.n >= 100) throw new Error('Limit: 100 active listings per seller'); if ((this.db.prepare('SELECT COUNT(*) AS n FROM listings').get() as { n: number }).n >= 10000) throw new Error('Marketplace journal limit reached; operator archival required'); if (this.db.prepare("SELECT 1 FROM listings WHERE card=? AND status IN ('open','reserved')").get(a.cardHash)) throw new Error('Card already listed'); const mint = s.nft(a.cardHash)!.block; const l: MarketListingV3 = { id: randomUUID(), termsVersion: 3, cardHash: a.cardHash, sequence: a.sequence, seller: c.actor, title: mint.article.title, image: mint.article.image!, kind: a.kind, priceAtomicXmr: a.priceAtomicXmr, feeBps: a.feeBps, hostFeeAddress, additionalFee: { ...ADDITIONAL_MARKET_FEE }, createdAt: now, endsAt: a.endsAt, status: 'open', highestBid: null }; this.saveListing(l); return l; } if (a.type === 'retire-legacy-listing') { const listing = this.listing(a.listingId); if (isXmrListing(listing)) throw new Error('Native-XMR listings use the signed cancel action'); if (listing.seller !== c.actor || !['open','invalid'].includes(listing.status)) throw new Error('Only the seller can retire an unreserved legacy listing'); if (listing.highestBid) throw new Error('Legacy auction has bids; resolve or wait for its original terms before retirement'); if (this.db.prepare("SELECT 1 FROM orders WHERE json_extract(data,'$.listingId')=? AND json_extract(data,'$.status') IN ('awaiting-payment','paid','delivery-pending','refund-required','review') LIMIT 1").get(listing.id)) { throw new Error('Legacy listing has an in-flight order; preserve that order under its original terms'); } listing.status = 'cancelled'; this.saveListing(listing); return listing; } if (a.type === 'bid' || a.type === 'buy' || a.type === 'cancel') { const l = this.listing(a.listingId); if (!isXmrListing(l)) throw new Error('Legacy USD listings are read-only; seller must retire and relist with fresh XMR authorization'); if (l.status !== 'open') throw new Error('Listing is no longer available'); if (a.type === 'cancel') { if (l.seller !== c.actor || l.highestBid) throw new Error('Only the seller can cancel, and auctions with bids cannot be cancelled'); l.status = 'cancelled'; this.saveListing(l); return l; } if (l.seller === c.actor) throw new Error('Cannot buy or bid on your own card'); if (a.type === 'bid') { if (l.kind !== 'auction' || now >= l.endsAt!) throw new Error('Auction is not accepting bids'); const minimum = l.highestBid ? BigInt(l.highestBid.priceAtomicXmr) + 1n : BigInt(l.priceAtomicXmr); if (BigInt(a.priceAtomicXmr) < minimum) throw new Error('Bid must meet the start price and exceed the current bid by one atomic unit'); l.highestBid = { bidder: c.actor, priceAtomicXmr: a.priceAtomicXmr }; this.saveListing(l); return l; } const price = l.kind === 'fixed' ? l.priceAtomicXmr : l.highestBid?.priceAtomicXmr; const hostFeeAddress = l.hostFeeAddress ?? ADDITIONAL_MARKET_FEE.address; if ((a.hostFeeAddress ?? ADDITIONAL_MARKET_FEE.address) !== hostFeeAddress) throw new Error('Host fee recipient does not match signed listing terms'); if (l.kind === 'auction' && (now < l.endsAt! || now >= l.endsAt! + PAYMENT_WINDOW || l.highestBid?.bidder !== c.actor)) throw new Error('Only the auction winner can check out within 30 minutes of closing'); if (price !== a.priceAtomicXmr || l.feeBps !== a.feeBps) throw new Error('Sale terms changed; review and sign again'); if (!this.card(s, l.cardHash, l.seller, l.sequence) || s.pendingTransfers.some(tx => transactionCardIds(tx).includes(l.cardHash))) throw new Error('Card is no longer available'); if ((this.db.prepare('SELECT COUNT(*) AS n FROM orders').get() as { n: number }).n >= 10000) throw new Error('Marketplace order limit reached; operator archival required'); const payBy = l.kind === 'auction' ? l.endsAt! + PAYMENT_WINDOW : now + PAYMENT_WINDOW; const order: MarketOrderV3 = { id: randomUUID(), termsVersion: 3, listingId: l.id, seller: l.seller, buyer: c.actor, cardHash: l.cardHash, sequence: l.sequence, priceAtomicXmr: price!, feeBps: l.feeBps, hostFeeAddress, createdAt: now, payBy, deliverBy: payBy + DELIVERY_WINDOW, status: 'awaiting-payment', ...mockXmrQuote(price!, l.feeBps), payout: 'not-started', deliveryConfirmations: 0, requiredConfirmations: this.config.confirmations }; l.status = 'reserved'; this.saveListing(l); this.saveOrder(order); return order; } if (a.type === 'mock-pay' || a.type === 'refund') { const order = this.order(a.orderId); if (!isXmrOrder(order)) throw new Error('Legacy USD orders keep their original terms and cannot be changed through native-XMR actions'); if (order.buyer !== c.actor) throw new Error('Only the order buyer can perform this action'); if (a.type === 'mock-pay') { if (order.status !== 'awaiting-payment' || now >= order.payBy) throw new Error('Invoice is not payable'); if (!this.card(s, order.cardHash, order.seller, order.sequence) || s.pendingTransfers.some(tx => transactionCardIds(tx).includes(order.cardHash))) throw new Error('Seller card changed; do not pay'); order.status = 'paid'; order.deliverBy = now + DELIVERY_WINDOW; } else { if (order.status !== 'refund-required' || s.pendingTransfers.some(tx => transactionCardIds(tx).includes(order.cardHash)) || s.transaction(this.deliveryId(order))) throw new Error('Refund requires failed delivery and no pending/canonical sale transfer'); order.status = 'refunded'; const listing = this.listing(order.listingId); listing.status = 'cancelled'; this.saveListing(listing); } this.saveOrder(order); return order; } throw new Error('Unsupported native-XMR market action'); } }