Repository Public
1import { ProviderRuntime, type ProviderStats } from './runtime.ts';2import { ProviderError, sameAsset, type SwapQuote, type SwapQuoteRequest } from './types.ts';34export type QuoteEvent =5 | { type: 'quote'; quote: SwapQuote; ms: number }6 | { type: 'provider'; provider: string; status: 'no_route' | 'skipped' | 'error'; code?: string; detail?: string; ms: number }7 | { type: 'done'; ms: number };89/** Structural validation applied to every provider answer before anyone else sees it. A failing quote is dropped and counted as an error. */10export function checkQuote(q: SwapQuote, req: SwapQuoteRequest, now: number): string | null {11 if (!sameAsset(q.request.from, req.from) || !sameAsset(q.request.to, req.to)) return 'quote is for another pair';12 if (q.request.amountIn !== req.amountIn) return 'quote is for another amount';13 if (q.request.recipient !== req.recipient || q.request.sender !== req.sender) return 'quote is for other addresses';14 if (typeof q.amountOut !== 'bigint' || typeof q.minAmountOut !== 'bigint') return 'amounts are not integers';15 if (q.amountOut <= 0n || q.minAmountOut <= 0n) return 'non-positive output';16 if (q.minAmountOut > q.amountOut) return 'minimum output above expected output';17 if (!(q.expiresAt > now)) return 'quote already expired';18 if (q.steps.length === 0 || q.steps.length > 6) return 'unreasonable number of steps';19 if (!sameAsset(q.steps[0]!.from, req.from) || !sameAsset(q.steps[q.steps.length - 1]!.to, req.to)) return 'steps do not connect the requested assets';20 for (let k = 1; k < q.steps.length; k++) if (!sameAsset(q.steps[k - 1]!.to, q.steps[k]!.from)) return 'steps are not contiguous';21 for (const f of q.fees) if (typeof f.amount !== 'bigint' || f.amount < 0n) return 'invalid fee amount';22 if (q.estimatedSeconds !== null && !(q.estimatedSeconds >= 0 && q.estimatedSeconds < 7 * 86400)) return 'invalid time estimate';23 if (q.priceImpactBps !== null && !(q.priceImpactBps >= -10_000 && q.priceImpactBps <= 10_000)) return 'invalid price impact';24 return null;25}2627export class QuoteAggregator {28 constructor(private runtimes: ProviderRuntime[], private now: () => number = Date.now) {}29 providers(): ProviderStats[] { return this.runtimes.map((r) => r.stats()); }30 runtime(id: string): ProviderRuntime | undefined { return this.runtimes.find((r) => r.provider.id === id); }3132 /** Query every compatible provider concurrently and yield results as they arrive (fastest first). */33 async *stream(req: SwapQuoteRequest, opts: { deadlineMs?: number; signal?: AbortSignal; only?: string[]; role?: 'direct' | 'leg' } = {}): AsyncGenerator<QuoteEvent> {34 const t0 = this.now();35 const ac = new AbortController();36 const kill = setTimeout(() => ac.abort(), opts.deadlineMs ?? 4000);37 opts.signal?.addEventListener('abort', () => ac.abort(), { once: true });38 const queue: QuoteEvent[] = []; let wake: (() => void) | null = null; let pending = 0;39 const push = (e: QuoteEvent) => { queue.push(e); wake?.(); };40 for (const rt of this.runtimes) {41 const id = rt.provider.id;42 if (opts.only && !opts.only.includes(id)) continue;43 const cap = rt.provider.capabilities();44 if (!rt.enabled) { push({ type: 'provider', provider: id, status: 'skipped', code: 'disabled', ms: 0 }); continue; }45 if (cap.needsKey && !cap.configured) { push({ type: 'provider', provider: id, status: 'skipped', code: 'not_configured', ms: 0 }); continue; }46 if (!rt.provider.supportsPair(req.from, req.to)) continue;47 if (rt.provider.policy?.scarce && (opts.role === 'leg' || req.from.network === req.to.network)) continue; // keep a small quota for what only it can do48 pending += 1;49 rt.quote(req, ac.signal).then((quotes) => {50 const ms = this.now() - t0; let ok = 0;51 for (const q of quotes) { const bad = checkQuote(q, req, this.now()); if (bad) push({ type: 'provider', provider: id, status: 'error', code: 'invalid_quote', detail: bad, ms }); else { ok += 1; push({ type: 'quote', quote: q, ms }); } }52 if (quotes.length === 0) push({ type: 'provider', provider: id, status: 'no_route', ms });53 void ok;54 }, (e: unknown) => {55 const err = e instanceof ProviderError ? e : new ProviderError(id, 'bad_response', String((e as Error)?.message ?? e));56 push({ type: 'provider', provider: id, status: 'error', code: err.code, detail: err.message.slice(0, 160), ms: this.now() - t0 });57 }).finally(() => { pending -= 1; wake?.(); });58 }59 try {60 for (;;) {61 while (queue.length) yield queue.shift()!;62 if (pending === 0) break;63 await new Promise<void>((r) => { wake = r; }); wake = null;64 }65 while (queue.length) yield queue.shift()!;66 yield { type: 'done', ms: this.now() - t0 };67 } finally { clearTimeout(kill); ac.abort(); }68 }69 async collect(req: SwapQuoteRequest, opts?: { deadlineMs?: number; signal?: AbortSignal; only?: string[]; role?: 'direct' | 'leg' }): Promise<{ quotes: SwapQuote[]; events: QuoteEvent[] }> {70 const quotes: SwapQuote[] = [], events: QuoteEvent[] = [];71 for await (const e of this.stream(req, opts)) { events.push(e); if (e.type === 'quote') quotes.push(e.quote); }72 return { quotes, events };73 }74}75