WikiDeck/wikideck
3
0
Fork 1
Code Issues 8 Pull Requests Actions Packages Projects 2 Releases Wiki Activity

Canonical indexes and recoverable market observation

Browse Source
This commit is contained in:
eric
2026-10-20 14:35:45 -04:00
parent fc65e497a8
commit e48081f041
11 changed files with 583 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/market/store.ts
+74 -18
View File
@@ -11,8 +11,17 @@ import { isXmrListing, isXmrOrder, marketCommandV3Schema, marketMessageV3,
type MarketOrderRecord, type MarketOrderV3 } from '../../shared/market-v3.ts';
import { CHAIN_ID, 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';
type Chain = { snapshot(): TestnetSnapshot; summary?(): { tip: string; revision: number }; submitTransfer(raw: unknown): Promise<string> };
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'] };
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 };
export function marketConfig(env: NodeJS.ProcessEnv = process.env): Config {
paymentMode(env);
@@ -28,12 +37,17 @@ export function marketConfig(env: NodeJS.ProcessEnv = process.env): Config {
}
const PAYMENT_WINDOW = 30 * 60_000;
const DELIVERY_WINDOW = 48 * 60 * 60_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();
/** 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 readonly db: DatabaseSync;
private readonly chain: Chain;
readonly config: Config;
@@ -41,6 +55,11 @@ export class Market {
private readonly now: () => number;
private constructor(db: DatabaseSync, chain: Chain, config: Config, id: string, now: () => number) {
this.db = db; this.chain = chain; this.config = config; this.id = id; this.now = now;
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;
}
}
static async open(directory: string, chain: Chain, config = marketConfig(), now = Date.now): Promise<Market> {
@@ -77,7 +96,8 @@ export class Market {
close() { this.db.close(); }
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 };
return { queued:this.queued,lastReconciledAt:this.reconciled?.at??null,orderStates:states,
canonicalEventCursor:this.eventCursor,canonicalEventHead:this.chain.canonicalHead?.()??null,eventBacklog:this.eventBacklog };
}
info(): MarketInfo {
return { network: NETWORK, chainId: CHAIN_ID, marketId: this.id, paymentMode: 'mock', marketVersion: 3, feeBps: this.config.feeBps,
@@ -110,11 +130,25 @@ export class Market {
try { const result = fn(); this.db.exec('COMMIT'); return result; }
catch (error) { this.db.exec('ROLLBACK'); throw error; }
}
private card(snapshot: TestnetSnapshot, hash: string, owner: string, sequence: number): boolean {
const block = snapshot.blocks.find(b => b.tokenId === hash);
return !!block && testnetOwnerOf(snapshot, hash) === owner &&
snapshot.transfers.filter(tx => transactionCardIds(tx).includes(hash)).length === sequence &&
snapshot.blocks.length + 1 - block.height >= CONSENSUS.coinbaseMaturity;
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,
@@ -124,17 +158,34 @@ export class Market {
}
/** Reconcile from canonical transaction identity, never from a mempool acknowledgement. */
private reconcile() {
const s = this.chain.snapshot(), now = this.now();
const deliveries = new Map<string, TestnetSnapshot['blocks'][number]>();
for (const b of s.blocks) for (const tx of b.transactions) deliveries.set(transactionId(tx), b);
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 s = this.view(), now = this.now();
const pendingIds = new Set(s.pendingTransfers.map(transactionId));
this.atomic(() => {
// Process every disconnect, even if the same delivery reappeared before this consumer caught up.
// Never infer that a current tip erases an intermediate loss of previously settled evidence.
const disconnected = new Set(events?.events.filter(e => e.type === 'disconnect').flatMap(e => e.transactionIds) ?? []);
if (events) {
for (const o of this.rows<MarketOrderRecord>('orders')) {
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);
}
}
this.db.prepare('INSERT INTO market_meta VALUES(?,?) ON CONFLICT(key) DO UPDATE SET value=excluded.value')
.run('canonical_event_cursor', JSON.stringify({ version: 1, streamId: events.streamId, cursor: events.cursor, anchor: events.anchor }));
// Catch-up is bounded per pass. No new payout, timeout/refund decision or mutation while events remain.
if (!events.caughtUp) return;
}
for (const o of this.rows<MarketOrderRecord>('orders')) {
const l = this.listing(o.listingId);
if (o.status === 'refunded' || o.status === 'expired') continue;
const deliveryId = this.deliveryId(o);
const delivery = deliveries.get(deliveryId);
const confirmations = delivery ? s.blocks.length - delivery.height + 1 : 0;
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) {
@@ -174,8 +225,12 @@ export class Market {
}
this.db.prepare('DELETE FROM receipts WHERE expires < ?').run(now - 86400_000);
});
if (events) {
this.eventCursor = events.cursor; this.eventBacklog = !events.caughtUp;
this.eventCheckpoint = { version: 1, streamId: events.streamId, cursor: events.cursor, anchor: events.anchor };
}
const summary = this.chain.summary?.();
if (summary) this.reconciled = { ...summary, at: now };
if (summary && !this.eventBacklog) this.reconciled = { ...summary, at: now };
}
browse(offset = 0): { listings: MarketListingRecord[]; next: number | null; info: MarketInfo } {
if (!Number.isSafeInteger(offset) || offset < 0 || offset > 10000) throw new Error('Invalid marketplace page');
@@ -215,6 +270,7 @@ export class Market {
const summary = this.chain.summary?.(), last = this.reconciled;
if (a.type !== 'activity' || !summary || !last || summary.tip !== last.tip || summary.revision !== last.revision ||
this.now() - last.at >= 1000) 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),
@@ -240,7 +296,7 @@ export class Market {
return value;
}
private mutateV2(c: MarketCommand): MarketActivityV3 | MarketListingRecord | MarketOrderRecord {
const a = c.action, now = this.now(), s = this.chain.snapshot();
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');
}
@@ -260,7 +316,7 @@ export class Market {
o.status = 'paid'; o.deliverBy = now + DELIVERY_WINDOW;
} else {
if (o.status !== 'refund-required' || s.pendingTransfers.some(tx => transactionCardIds(tx).includes(o.cardHash)) ||
s.blocks.some(b => b.transactions.some(tx => transactionId(tx) === this.deliveryId(o)))) throw new Error('Refund requires failed delivery and no pending/canonical sale transfer');
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;
@@ -275,7 +331,7 @@ export class Market {
}
private mutateV3(c: MarketCommandV3): MarketActivityV3 | MarketListingRecord | MarketOrderRecord {
const a = c.action, now = this.now(), s = this.chain.snapshot();
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');
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');
@@ -283,7 +339,7 @@ export class Market {
if (this.rows<MarketListingRecord>('listings').filter(l => l.seller === c.actor && ['open','reserved'].includes(l.status)).length >= 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.blocks.find(b => b.tokenId === a.cardHash)!;
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, additionalFee: { ...ADDITIONAL_MARKET_FEE },
@@ -337,7 +393,7 @@ export class Market {
order.status = 'paid'; order.deliverBy = now + DELIVERY_WINDOW;
} else {
if (order.status !== 'refund-required' || s.pendingTransfers.some(tx => transactionCardIds(tx).includes(order.cardHash)) ||
s.blocks.some(b => b.transactions.some(tx => transactionId(tx) === this.deliveryId(order)))) throw new Error('Refund requires failed delivery and no pending/canonical sale transfer');
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;