wikideck/server/market/reconciliation-commit-worker.js

import { DatabaseSync } from 'node:sqlite';
import { parentPort, workerData } from 'node:worker_threads';
import { isDeepStrictEqual } from 'node:util';

const db = new DatabaseSync(workerData.marketFile);
db.exec('PRAGMA busy_timeout=5000; PRAGMA synchronous=FULL;');
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 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 {
    if (request.operation === 'reconcile-batch') {
      atomic(() => {
        assertCurrentJob(request.expectedJob);
        const selected = rows(request.stage, request.afterId, request.batchSize);
        const result = request.stage === 'orders' ? processOrders(request, selected) : processListings(request, selected);
        saveResults(result);
        db.prepare("UPDATE market_meta SET value=? WHERE key='canonical_reconciliation_job'")
          .run(JSON.stringify(request.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 });
  } catch (error) {
    port.postMessage({ id: request.id, ok: false, error: error instanceof Error ? error.message : 'Worker batch failed' });
  }
});
port.on('close', () => db.close());