185 lines
10 KiB
JavaScript
Executable File
185 lines
10 KiB
JavaScript
Executable File
import { DatabaseSync } from 'node:sqlite';
|
|
import { parentPort, workerData } from 'node:worker_threads';
|
|
import { isDeepStrictEqual } from 'node:util';
|
|
import { createHash } from 'node:crypto';
|
|
|
|
const db = new DatabaseSync(workerData.marketFile);
|
|
db.exec('PRAGMA busy_timeout=5000; PRAGMA synchronous=FULL;');
|
|
const ledger = workerData.ledgerFile ? new DatabaseSync(workerData.ledgerFile, { readOnly: true }) : null;
|
|
ledger?.exec('PRAGMA busy_timeout=5000; PRAGMA query_only=ON;');
|
|
const port = parentPort;
|
|
if (!port) throw new Error('Reconciliation commit worker requires a parent port');
|
|
|
|
function atomic(fn) {
|
|
db.exec('BEGIN IMMEDIATE');
|
|
try { const result = fn(); db.exec('COMMIT'); return result; }
|
|
catch (error) { db.exec('ROLLBACK'); throw error; }
|
|
}
|
|
function assertCurrentJob(expected) {
|
|
const row = db.prepare("SELECT value FROM market_meta WHERE key='canonical_reconciliation_job'").get();
|
|
if (!row || !isDeepStrictEqual(JSON.parse(row.value), expected)) {
|
|
throw new Error('Reconciliation job changed before worker commit');
|
|
}
|
|
}
|
|
function rows(stage, afterId, limit) {
|
|
const table = stage === 'orders' ? 'orders' : 'listings';
|
|
return db.prepare(`SELECT id,data FROM ${table} WHERE (? IS NULL OR id>?) ORDER BY id LIMIT ?`)
|
|
.all(afterId, afterId, limit);
|
|
}
|
|
function readLedgerFacts(request, selected) {
|
|
if (!ledger) return request.fallbackFacts;
|
|
ledger.exec('BEGIN');
|
|
try {
|
|
const tip = ledger.prepare('SELECT hash FROM canonical_blocks ORDER BY height DESC LIMIT 1').get()?.hash ?? request.genesisHash;
|
|
const height = ledger.prepare('SELECT COALESCE(MAX(height),0) AS n FROM canonical_blocks').get().n;
|
|
const eventHead = ledger.prepare('SELECT COALESCE(MAX(sequence),0) AS n FROM canonical_events').get().n;
|
|
const pendingTransactions = ledger.prepare('SELECT id,transaction_json FROM pending_transfers ORDER BY position').all()
|
|
.map(row => {
|
|
const tx = JSON.parse(row.transaction_json);
|
|
const cardIds = tx.type === 'trade' ? [...tx.offered, ...tx.requested].map(leg => leg.cardHash) : [tx.cardHash];
|
|
return { id: row.id, cardIds };
|
|
});
|
|
const pendingIds = pendingTransactions.map(tx => tx.id);
|
|
const deliveryIds = Object.create(null), transactionIds = new Set(), cardIds = new Set();
|
|
for (const row of selected) {
|
|
const record = JSON.parse(row.data);
|
|
if (request.stage === 'orders') {
|
|
const id = createHash('sha256').update(JSON.stringify([request.network, 'nft-transfer', request.chainId,
|
|
record.cardHash, record.seller, record.buyer, record.sequence])).digest('hex');
|
|
deliveryIds[record.id] = id; transactionIds.add(id); cardIds.add(record.cardHash);
|
|
} else cardIds.add(record.cardHash);
|
|
}
|
|
const deliveryTransactions = Object.create(null);
|
|
const readTx = ledger.prepare('SELECT height,block_hash FROM canonical_transactions WHERE id=?');
|
|
for (const id of transactionIds) deliveryTransactions[id] = readTx.get(id) ?? null;
|
|
const nfts = Object.create(null);
|
|
const readNft = ledger.prepare(`SELECT n.owner,n.sequence,n.minted_height,b.block_json FROM canonical_nfts n
|
|
JOIN blocks b ON b.hash=n.block_hash WHERE n.token=?`);
|
|
for (const card of cardIds) {
|
|
const nft = readNft.get(card);
|
|
nfts[card] = nft ? { owner: nft.owner, sequence: nft.sequence, confirmations: height - nft.minted_height + 1 } : null;
|
|
}
|
|
ledger.exec('COMMIT');
|
|
return { height,
|
|
pendingTransactions, pendingIds, pendingCardIds: [...new Set(pendingTransactions.flatMap(tx => tx.cardIds))],
|
|
deliveryIds, deliveryTransactions, nfts,
|
|
snapshot: { tip, eventHead, pendingDigest: JSON.stringify([...pendingIds].sort()) } };
|
|
} catch (error) { ledger.exec('ROLLBACK'); throw error; }
|
|
}
|
|
function validCard(facts, cardHash, owner, sequence, maturity) {
|
|
const nft = facts.nfts[cardHash];
|
|
return !!nft && nft.owner === owner && nft.sequence === sequence && nft.confirmations >= maturity;
|
|
}
|
|
function deliveryId(request, order) { return request.deliveryIds[order.id]; }
|
|
function processOrders(request, selected) {
|
|
const disconnected = new Set(request.disconnectedIds);
|
|
const pendingIds = new Set(request.pendingIds);
|
|
const orders = [], listings = [];
|
|
for (const row of selected) {
|
|
const order = JSON.parse(row.data);
|
|
const listingRow = db.prepare('SELECT data FROM listings WHERE id=?').get(order.listingId);
|
|
if (!listingRow) throw new Error('Order listing missing during reconciliation');
|
|
const listing = JSON.parse(listingRow.data);
|
|
const id = deliveryId(request, order), delivery = request.deliveryTransactions[id];
|
|
if (order.status === 'complete' && disconnected.has(id)) {
|
|
order.status = 'review'; order.payout = 'review';
|
|
if (order.additionalFee) order.additionalFee.payout = 'review';
|
|
}
|
|
if (order.status === 'refunded' || order.status === 'expired') continue;
|
|
const confirmations = delivery ? request.height - delivery.height + 1 : 0;
|
|
order.deliveryConfirmations = confirmations;
|
|
if (order.status === 'complete' || order.status === 'review') {
|
|
if (!delivery || confirmations < order.requiredConfirmations) {
|
|
order.status = 'review'; order.payout = 'review';
|
|
if (order.additionalFee) order.additionalFee.payout = 'review';
|
|
}
|
|
orders.push(order);
|
|
continue;
|
|
}
|
|
if (delivery && order.status === 'awaiting-payment') {
|
|
order.status = 'review'; order.payout = 'review';
|
|
if (order.additionalFee) order.additionalFee.payout = 'review';
|
|
} else if (delivery) {
|
|
order.transferId = id; order.status = 'delivery-pending';
|
|
if (confirmations >= order.requiredConfirmations) {
|
|
order.status = 'complete'; order.payout = 'mock-paid'; listing.status = 'sold';
|
|
if (order.additionalFee) order.additionalFee.payout = 'mock-paid';
|
|
}
|
|
} else if (order.status === 'awaiting-payment' && request.expectedJob.observedAt >= order.payBy) {
|
|
order.status = 'expired'; listing.status = listing.kind === 'fixed' ? 'open' : 'expired';
|
|
} else if (order.status !== 'awaiting-payment') {
|
|
const pending = pendingIds.has(id);
|
|
const conflict = request.pendingTransactions.some(tx => tx.cardIds.includes(order.cardHash) && tx.id !== id);
|
|
if (!validCard(request, order.cardHash, order.seller, order.sequence, request.maturity) ||
|
|
conflict || (request.expectedJob.observedAt >= order.deliverBy && !pending)) order.status = 'refund-required';
|
|
else if (order.status === 'delivery-pending' && !pending) order.status = 'paid';
|
|
}
|
|
orders.push(order);
|
|
listings.push(listing);
|
|
}
|
|
return { orders, listings };
|
|
}
|
|
function processListings(request, selected) {
|
|
const pendingCards = new Set(request.pendingCardIds), listings = [];
|
|
for (const row of selected) {
|
|
const listing = JSON.parse(row.data);
|
|
if (listing.status !== 'open') continue;
|
|
if (listing.additionalFee?.bps !== request.additionalFee.bps || listing.additionalFee?.address !== request.additionalFee.address ||
|
|
!validCard(request, listing.cardHash, listing.seller, listing.sequence, request.maturity) || pendingCards.has(listing.cardHash)) {
|
|
listing.status = 'invalid';
|
|
} else if (listing.kind === 'auction' && request.expectedJob.observedAt >= listing.endsAt + request.paymentWindow) {
|
|
listing.status = 'expired';
|
|
} else if (listing.kind === 'auction' && request.expectedJob.observedAt >= listing.endsAt && !listing.highestBid) {
|
|
listing.status = 'expired';
|
|
}
|
|
listings.push(listing);
|
|
}
|
|
return { orders: [], listings };
|
|
}
|
|
function saveResults(result) {
|
|
const order = db.prepare(`INSERT INTO orders(id,data) VALUES(?,?)
|
|
ON CONFLICT(id) DO UPDATE SET data=excluded.data WHERE orders.data<>excluded.data`);
|
|
for (const item of result.orders) order.run(item.id, JSON.stringify(item));
|
|
const listing = db.prepare(`INSERT INTO listings(id,card,status,data) VALUES(?,?,?,?)
|
|
ON CONFLICT(id) DO UPDATE SET status=excluded.status,data=excluded.data WHERE listings.data<>excluded.data`);
|
|
for (const item of result.listings) listing.run(item.id, item.cardHash, item.status, JSON.stringify(item));
|
|
}
|
|
|
|
port.on('message', request => {
|
|
try {
|
|
let snapshot, nextJob;
|
|
if (request.operation === 'reconcile-batch') {
|
|
atomic(() => {
|
|
assertCurrentJob(request.expectedJob);
|
|
const selected = rows(request.stage, request.afterId, request.batchSize);
|
|
const facts = readLedgerFacts(request, selected);
|
|
snapshot = facts.snapshot;
|
|
const prepared = { ...request, ...facts };
|
|
const result = request.stage === 'orders' ? processOrders(prepared, selected) : processListings(prepared, selected);
|
|
saveResults(result);
|
|
nextJob = request.stage === 'orders'
|
|
? selected.length < request.batchSize ? { ...request.expectedJob, stage: 'listings', afterId: null }
|
|
: { ...request.expectedJob, afterId: selected.at(-1).id }
|
|
: selected.length < request.batchSize ? { ...request.expectedJob, stage: 'finalize', afterId: null }
|
|
: { ...request.expectedJob, afterId: selected.at(-1).id };
|
|
db.prepare("UPDATE market_meta SET value=? WHERE key='canonical_reconciliation_job'")
|
|
.run(JSON.stringify(nextJob));
|
|
});
|
|
} else if (request.operation === 'finalize') {
|
|
atomic(() => {
|
|
assertCurrentJob(request.expectedJob);
|
|
db.prepare('DELETE FROM receipts WHERE expires < ?').run(request.observedAt - 86400_000);
|
|
if (request.cursor) {
|
|
db.prepare(`INSERT INTO market_meta(key,value) VALUES('canonical_event_cursor',?)
|
|
ON CONFLICT(key) DO UPDATE SET value=excluded.value`).run(request.cursor);
|
|
}
|
|
db.prepare("DELETE FROM market_meta WHERE key='canonical_reconciliation_job'").run();
|
|
});
|
|
} else throw new Error('Unknown reconciliation worker operation');
|
|
port.postMessage({ id: request.id, ok: true, snapshot, nextJob });
|
|
} catch (error) {
|
|
port.postMessage({ id: request.id, ok: false, error: error instanceof Error ? error.message : 'Worker batch failed' });
|
|
}
|
|
});
|
|
port.on('close', () => { ledger?.close(); db.close(); });
|