Files
2026-10-22 14:40:00 -04:00

665 lines
47 KiB
TypeScript
Executable File
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
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<string>; 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<unknown> = Promise.resolve();
private queued = 0;
private reconciled: { tip: string; revision: number; at: number } | undefined;
private eventCursor = 0;
private eventCheckpoint: z.infer<typeof cursorSchema> | undefined;
private eventBacklog = false;
private reconciliationJob: z.infer<typeof reconciliationJobSchema> | undefined;
private readonly reconciliationWorker?: Worker;
private readonly workerReadsLedger: boolean;
private workerRequestId = 0;
private readonly workerRequests = new Map<number, { resolve(message: Record<string, unknown>): 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<Market> {
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<void> {
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<T>(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<string, unknown>): Promise<Record<string, unknown>> {
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<typeof reconciliationJobSchema>): 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<typeof reconciliationJobSchema>): void {
if (!this.reconciliationWorker || this.reconciliationWorkerBusy || job.stage === 'finalize') return;
this.reconciliationWorkerBusy = true;
try {
let fallbackFacts: Record<string, unknown> | 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<string, string> = Object.create(null), targetCards = new Set<string>();
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<string, { height: number; blockHash: string } | null> = Object.create(null);
const nfts: Record<string, { owner: string; sequence: number; confirmations: number } | null> = 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<typeof reconciliationJobSchema>, 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<typeof reconciliationJobSchema> | 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<MarketActivityV3 | MarketListingRecord | MarketOrderRecord> {
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<MarketActivityV3 | MarketListingRecord | MarketOrderRecord> {
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<T>(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');
}
}