Repository Public
1import { NETWORKS, isTerminal, type NetworkDef } from '@spw/core';2import { adapterFor, type Health } from './chains/index.ts';34const NETS = Object.values(NETWORKS) as NetworkDef[];5import { config } from './config.ts';6import { audit, sql } from './db.ts';7import { signer } from './signer-client.ts';8import { advanceSwap, recordSwapOutcome } from './swap-exec.ts';9import { advanceBridgeBurn, advanceBridges } from './bridge.ts';10import { broadcastNow, move, type ProposalRow } from './transfers.ts';11import { advanceXswapSend, advanceXswaps, xswapMayBroadcast } from './xswap.ts';12import { advancePlans } from './plan-exec.ts'; // v7 any-to-any plans13import { advanceWithdrawals } from './agents/withdrawals.ts'; // v81415export const healthCache = new Map<string, Health>();16const log = (m: string) => console.log(`[worker] ${m}`);1718async function expireStale(): Promise<void> {19 const rows = await sql<ProposalRow[]>`select * from proposals where state in ('draft','pending_approval') and expires_at < now()`;20 for (const p of rows) await move(p, 'expired', { error: 'expired before approval' }).catch(() => {});21 // an "approved" row older than a minute means the process died mid-sign: the signer's replay cache makes a retry impossible, so fail it22 const stuck = await sql<ProposalRow[]>`select * from proposals where state = 'approved' and updated_at < now() - interval '1 minute'`;23 for (const p of stuck) await move(p, 'failed', { error: 'interrupted during signing; nothing was broadcast' }).catch(() => {});24}2526async function reconcile(): Promise<void> {27 const rows = await sql<ProposalRow[]>`select * from proposals where state in ('signed','broadcast','confirming','reorged')`;28 for (let p of rows) {29 try {30 if (p.state === 'signed') {31 // never relay a transaction whose approval window is long gone32 if (p.expires_at && Date.now() - p.expires_at.getTime() > 10 * 60_000) { await move(p, 'failed', { error: 'could not be broadcast in time; nothing was sent' }); continue; }33 p = p.kind === 'swap' ? await advanceSwap(p) : p.kind === 'bridge' ? await advanceBridgeBurn(p) : p.kind === 'xswap' ? await advanceXswapSend(p)34 : p.kind === 'gmx' ? await (await import('./derivatives/gmx-live.ts')).gmx().advanceSend(p) /* v8: approval mined first, then the order transaction, simulated */ : await broadcastNow(p);35 if (p.state !== 'broadcast') continue;36 }37 if (!config.broadcast || !p.txid || !p.intent) continue;38 const a = adapterFor(p.network);39 // a swap preceded by an approval uses the next nonce40 const tracked = (p.kind === 'swap' || p.kind === 'bridge' || p.kind === 'xswap' || p.kind === 'gmx') && p.pre_raw && p.intent.evm ? { ...p.intent, evm: { ...p.intent.evm, nonce: p.intent.evm.nonce + 1 } } : p.intent;41 const t = await a.track({ txid: p.txid, intent: tracked, blockRef: p.block_ref, broadcastAt: (p.broadcast_at?.getTime() ?? Date.now()) / 1000 });42 if (t.rebroadcast && p.raw && Date.now() - (p.last_broadcast_at?.getTime() ?? 0) > 45_000 && (p.kind !== 'xswap' || xswapMayBroadcast(p))) { // a vault payment is never relayed again late in its window43 await a.broadcast(p.raw).catch(() => {});44 await sql`update proposals set last_broadcast_at = now() where id = ${p.id}`;45 }46 if (t.state !== p.state) {47 const patch: Record<string, unknown> = { confirmations: t.confirmations, block_ref: t.blockRef ?? null, error: t.error ?? null };48 if (t.state === 'finalized') patch.finalized_at = new Date();49 const moved = await move(p, t.state, patch);50 if (t.state === 'finalized' || t.state === 'confirming') await recordSwapOutcome(moved).catch(() => {});51 } else if (t.confirmations !== p.confirmations) await sql`update proposals set confirmations = ${t.confirmations}, block_ref = ${t.blockRef ?? p.block_ref} where id = ${p.id}`;52 } catch (e) { log(`reconcile ${p.id}: ${(e as Error).message}`); }53 }54}5556async function scanDeposits(): Promise<void> {57 const st = await signer.status().catch(() => null);58 if (!st) return;59 for (const n of NETS) {60 const address = st.addresses[n.id];61 if (!address) continue;62 try {63 const [c] = await sql<{ cursor: string }[]>`select cursor from scan_cursor where network = ${n.id}`;64 const r = await adapterFor(n.id).deposits(address, c ? Number(c.cursor) : null);65 for (const d of r.items) {66 const ins = await sql`insert into deposits (network, txid, idx, asset, amount, from_addr, block, block_time)67 values (${n.id}, ${d.txid}, ${d.idx}, ${d.asset}, ${d.amount.toString()}, ${d.from}, ${d.block}, ${d.time ? new Date(d.time * 1000) : null})68 on conflict do nothing returning txid`;69 if (ins.length) await audit('deposit.seen', 'chain', { network: n.id, txid: d.txid, asset: d.asset, amount: d.amount.toString() });70 }71 await sql`insert into scan_cursor (network, cursor, finalized) values (${n.id}, ${r.cursor}, ${Math.min(r.finalizedHeight, Number.MAX_SAFE_INTEGER)})72 on conflict (network) do update set cursor = excluded.cursor, finalized = excluded.finalized`;73 } catch (e) { log(`deposits ${n.id}: ${(e as Error).message.slice(0, 120)}`); }74 }75}7677async function refreshHealth(): Promise<void> {78 await Promise.all(NETS.map(async (n) => {79 try { healthCache.set(n.id, await adapterFor(n.id).health()); }80 catch { healthCache.set(n.id, { network: n.id, ok: false, readable: false, identityVerified: false, height: null, endpoints: [], checkedAt: new Date().toISOString() }); }81 }));82}8384// v6-alerts: heartbeat of every worker loop (read by the monitoring dashboard and the "worker failing" alert) — additive85import { loopStats, type LoopStat } from './loop-stats.ts'; // v8: the map lives in its own module (the agent engines report there too)86export { loopStats, type LoopStat };8788export function loop(name: string, fn: () => Promise<void>, everyMs: number): void {89 let running = false;90 const st: LoopStat = { name, everyMs, runs: 0, lastStartAt: null, lastOkAt: null, lastErrorAt: null, lastError: null, failures: 0 };91 loopStats.set(name, st);92 const tick = async () => { if (running) return; running = true; st.lastStartAt = Date.now(); st.runs += 1;93 try { await fn(); st.lastOkAt = Date.now(); st.failures = 0; }94 catch (e) { log(`${name}: ${(e as Error).message}`); st.lastErrorAt = Date.now(); st.lastError = (e as Error).message.slice(0, 200); st.failures += 1; }95 finally { running = false; } };96 void tick();97 setInterval(tick, everyMs).unref();98}99100export function startWorkers(): void {101 loop('health', refreshHealth, 45_000);102 loop('expire', expireStale, 15_000);103 loop('reconcile', reconcile, 8_000);104 loop('bridges', advanceBridges, 5_000);105 loop('xswaps', advanceXswaps, 10_000);106 loop('plans', advancePlans, 6_000); // v7: reads the proposals of open plans and moves the plans; never signs107 loop('deposits', scanDeposits, 75_000);108 loop('withdrawals', () => advanceWithdrawals(), 8_000); // v8: agent-vault withdrawals — receipt and arrival at the owner's address; never signs109}110export { refreshHealth, reconcile, expireStale, isTerminal, advanceBridges };111