import { Hex } from 'ox' import * as Db from '../../db/Db.js' import * as FundingDepositAddresses from '../../db/tables/fundingDepositAddresses.js' import * as FundingDepositRequestObservations from '../../db/tables/fundingDepositRequestObservations.js' import * as FundingDeposits from '../../db/tables/fundingDeposits.js' import type * as Metrics from '../../Metrics.js' import * as Value from '../Value.js' import type * as Webhooks from '../Webhooks.js' import * as Deposit from './Deposit.js' import * as DepositWebhook from './DepositWebhook.js' import * as Provider from './Provider.js' import * as Subsidy from './Subsidy.js' const defaultBatchSize = 25 const defaultLeaseMs = 2 * 60_000 const defaultPollIntervalMs = 60_000 const chainProviderFollowUpDelaySeconds = 1 const chainProviderMaximumFollowUpAttempts = 30 const followUpDelaySeconds = { delivery: 1, provider: 5, settlement: 2 } as const satisfies Record< Provider.Webhook.receive.FollowUp['kind'], number > const maximumPages = 100 const maximumFollowUpAttempts = 6 const providerFailureStatuses = ['failure'] as const const providerRefundStatuses = ['refund', 'refunded'] as const /** Verified token transfer returned by an authoritative chain reader. */ export type TransferEvidence = { /** Token quantity in base units. */ amount: string /** Verified sender account, when the chain exposes it. */ sender?: string | undefined /** Verified transaction reference. */ transactionHash: string /** Transfer position within the transaction. */ transferIndex: number } /** Authoritative chain-evidence reader used during deposit reconciliation. */ export type VerifyTransfers = ( parameters: VerifyTransfers.Parameters, ) => Promise /** Parameters passed to an authoritative chain-evidence reader. */ export declare namespace VerifyTransfers { type Parameters = { /** Account that must receive the token transfer. */ recipient: string /** Chain containing the transaction. */ chain: Provider.ChainRef /** Token that must be transferred. */ token: Provider.TokenRef /** Provider-discovered transaction reference to verify. */ transactionHash: string } } /** Creates a reusable-address reconciliation tracker. */ export function createTracker(options: createTracker.Options): createTracker.Tracker { const now = options.now ?? (() => new Date()) const batchSize = options.batchSize ?? defaultBatchSize const leaseMs = options.leaseMs ?? defaultLeaseMs const pollIntervalMs = options.pollIntervalMs ?? defaultPollIntervalMs const record = (outcome: string, providerId: string, trigger: Deposit.DetectionTrigger) => { options.metrics?.count('funding_deposit_reconciliation_count', 1, { outcome, provider: providerId, trigger, }) } return { async reconcile(message) { const database = Db.get(options.db) const startedAt = now() // Legacy poll messages carry a lease version; legacy webhook messages do not. const trigger = (() => { if (message.trigger) return message.trigger if (message.pollLeaseVersion !== undefined) return 'poll' return 'webhook' })() const claimed = await claim(database, message, { leaseUntil: new Date(startedAt.getTime() + leaseMs).toISOString(), now: startedAt.toISOString(), }) if (!claimed) return { type: 'busy' } const provider = options.providers.find((candidate) => candidate.id === claimed.providerId) if (!provider || !Provider.canListDepositAddressRequests(provider)) { await fail(database, claimed, startedAt, pollIntervalMs) record('provider_unavailable', claimed.providerId, trigger) return { type: 'provider_unavailable' } } const heartbeat = maintainLease(database, claimed, { leaseMs, now }) try { await heartbeat.ready() const requests = await listRequests(provider, { address: claimed, requestId: message.requestId, signal: options.signal, }) if (trigger === 'poll') await FundingDepositRequestObservations.observePoll(database, { depositAddressId: claimed.id, observedAt: now().toISOString(), providerRequestIds: requests.requests.map((request) => request.id), }) const webhookUpdateObserved = (() => { const timing = message.webhookTiming if (!timing) return true if (!message.requestId) return false // Relay can send a webhook before its requests API indexes the same update. return requests.requests.some( (request) => request.id === message.requestId && request.updatedAt >= timing.providerUpdatedAt, ) })() heartbeat.assert() let created = 0 const depositWebhookReferences = new Map() const onDeposit: NonNullable = ( _deposit, references, ) => { for (const reference of references) depositWebhookReferences.set( `${reference.subscriptionId}:${reference.eventId}`, reference, ) } let updated = 0 const pendingRequestIds = new Set([ ...requests.pendingRequestIds, ...(message.requestId ? [message.requestId] : []), ]) let followUpKind: Provider.Webhook.receive.FollowUp['kind'] | undefined for (const request of requests.requests) { const result = await reconcileRequest(database, { address: claimed, metrics: options.metrics, onDeposit, provider, request, signal: options.signal, subsidies: options.subsidies, trigger, verifyTransfers: options.verifyTransfers, }) heartbeat.assert() created += result.created updated += result.updated if (result.terminal) pendingRequestIds.delete(request.id) else { pendingRequestIds.add(request.id) followUpKind = result.followUp ?? followUpKind ?? 'provider' } } const unattributed = await reconcileUnattributed(database, { address: claimed, metrics: options.metrics, onDeposit, provider, signal: options.signal, subsidies: options.subsidies, verifyTransfers: options.verifyTransfers, }) heartbeat.assert() updated += unattributed.updatedIds.size if (!unattributed.terminal) followUpKind = unattributed.followUp ?? followUpKind ?? 'provider' const awaitsProvider = await FundingDeposits.hasUnattributed(database, claimed.id) if (message.requestId && message.webhookTiming && !webhookUpdateObserved) pendingRequestIds.add(message.requestId) if (created > 0) options.metrics?.count('funding_deposit_detection_count', created, { provider: claimed.providerId, trigger, }) if (options.dispatchDepositUpdates && depositWebhookReferences.size > 0) { const references = [...depositWebhookReferences.values()] await options.dispatchDepositUpdates(references).then( (count) => options.metrics?.count('funding_deposit_webhook_enqueue_count', count, { outcome: 'succeeded', provider: claimed.providerId, }), (cause) => { options.metrics?.count('funding_deposit_webhook_enqueue_count', references.length, { outcome: 'failed', provider: claimed.providerId, }) throw cause }, ) } const leaseFailure = await heartbeat.stop() if (leaseFailure !== undefined) throw leaseFailure const completedAt = now() const stored = await FundingDepositAddresses.completePoll(database, { id: claimed.id, lastPolledAt: completedAt.toISOString(), nextPollAt: new Date(completedAt.getTime() + pollIntervalMs).toISOString(), pollLeaseVersion: claimed.pollLeaseVersion, providerRequestIds: claimed.providerRequestIds, providerState: reconciliationState(claimed.providerState, { pendingRequestIds: [...pendingRequestIds], requests, ...(message.webhookTiming && message.requestId && webhookUpdateObserved ? { webhookTiming: { ...message.webhookTiming, reconciliationCompletedAt: completedAt.toISOString(), requestId: message.requestId, }, } : {}), }), }) if (!stored) return { type: 'stale' } if (pendingRequestIds.size > 0 || awaitsProvider) { const followUp = nextFollowUp(message, followUpKind ?? 'provider') if (followUp) await options .dispatch( { addressId: claimed.id, followUp: followUp.state, ...(message.requestId ? { requestId: message.requestId } : {}), trigger, type: 'funding:deposit-address:reconcile', ...(message.webhookTiming && !webhookUpdateObserved ? { webhookTiming: message.webhookTiming } : {}), }, { delaySeconds: followUp.delaySeconds }, ) .catch((cause) => { record('follow_up_enqueue_failed', claimed.providerId, trigger) throw cause }) record( followUp ? 'follow_up_enqueued' : 'follow_up_exhausted', claimed.providerId, trigger, ) } if (message.webhookTiming && webhookUpdateObserved) recordWebhookLatency(options.metrics, { completedAt, providerId: claimed.providerId, timing: message.webhookTiming, }) record('completed', claimed.providerId, trigger) options.metrics?.histogram( 'funding_deposit_reconciliation_duration_ms', completedAt.getTime() - startedAt.getTime(), { provider: claimed.providerId, trigger }, ) return { created, type: 'completed', updated } } catch (cause) { const leaseFailure = await heartbeat.stop() await fail(database, claimed, now(), pollIntervalMs) record('failed', claimed.providerId, trigger) throw leaseFailure ?? cause } finally { options.metrics?.flush() } }, async tick() { const database = Db.get(options.db) const startedAt = now() const leaseUntil = new Date(startedAt.getTime() + leaseMs).toISOString() let claimed = 0 let failed = 0 for (const provider of options.providers) { if (!Provider.canListDepositAddressRequests(provider)) continue const records = await FundingDepositAddresses.claimDue(database, { leaseUntil, limit: Math.max(0, batchSize - claimed), now: startedAt.toISOString(), providerId: provider.id, }) for (const address of records) { claimed += 1 options.metrics?.histogram( 'funding_deposit_poll_lag_ms', Math.max(0, startedAt.getTime() - new Date(address.nextPollAt).getTime()), { provider: provider.id }, ) try { await options.dispatch({ addressId: address.id, pollLeaseVersion: address.pollLeaseVersion, trigger: 'poll', type: 'funding:deposit-address:reconcile', }) record('enqueued', provider.id, 'poll') } catch { failed += 1 await fail(database, address, startedAt, pollIntervalMs) record('enqueue_failed', provider.id, 'poll') } } if (claimed >= batchSize) break } options.metrics?.flush() return { claimed, failed } }, } } export declare namespace createTracker { /** Dependencies and bounds for deposit reconciliation. */ type Options = { /** Maximum addresses claimed by one scheduled tick. */ batchSize?: number | undefined /** Authoritative database or per-invocation database factory. */ db: Db.Source /** Durable reconciliation queue transport. */ dispatch: ( message: Provider.Webhook.receive.Dispatchable, options?: createTracker.DispatchOptions, ) => Promise /** Stages and dispatches committed funding deposit webhook updates. */ dispatchDepositUpdates?: | ((references: readonly Webhooks.QueueReference[]) => Promise) | undefined /** Duration of one reconciliation lease. */ leaseMs?: number | undefined /** Optional bounded operational metrics sink. */ metrics?: Metrics.Metrics | undefined /** Clock used for scheduling and tests. */ now?: (() => Date) | undefined /** Delay before a successfully reconciled address is polled again. */ pollIntervalMs?: number | undefined /** Configured funding provider integrations. */ providers: readonly Provider.Provider[] /** Optional cancellation signal for provider reads. */ signal?: AbortSignal | undefined /** Optional pathUSD-funded subsidy settler. */ subsidies?: Subsidy.Settler | undefined /** Authoritative source and Tempo transfer verifier. */ verifyTransfers: VerifyTransfers } /** Delivery options for one reconciliation queue message. */ type DispatchOptions = { /** Delay before the queue makes the message available. */ delaySeconds: number } /** Result of reconciling one queue message. */ type ReconcileResult = | { type: 'busy' | 'provider_unavailable' | 'stale' } | { created: number; type: 'completed'; updated: number } /** Scheduled and queue-driven reusable-address tracker. */ type Tracker = { /** Reconciles one address from provider discovery and verified chain evidence. */ reconcile(message: Provider.Webhook.receive.Queued): Promise /** Claims and enqueues one bounded batch of due reusable addresses. */ tick(): Promise<{ claimed: number; failed: number }> } } type ClaimOptions = { leaseUntil: string now: string } async function claim(db: Db.Db, message: Provider.Webhook.receive.Queued, options: ClaimOptions) { if (message.pollLeaseVersion === undefined) return FundingDepositAddresses.claim(db, { id: message.addressId, leaseUntil: options.leaseUntil, now: options.now, }) const current = await FundingDepositAddresses.get(db, message.addressId) if ( !current?.pollLeaseUntil || current.pollLeaseUntil <= options.now || current.pollLeaseVersion !== message.pollLeaseVersion ) return undefined return current } async function fail( db: Db.Db, address: FundingDepositAddresses.Record, now: Date, pollIntervalMs: number, ) { const delay = Math.min(pollIntervalMs * 2 ** address.pollFailureCount, 15 * 60_000) return FundingDepositAddresses.failPoll(db, { id: address.id, nextPollAt: new Date(now.getTime() + delay).toISOString(), pollLeaseVersion: address.pollLeaseVersion, }) } type MaintainLeaseOptions = { leaseMs: number now: () => Date } function maintainLease( db: Db.Db, address: FundingDepositAddresses.Record, options: MaintainLeaseOptions, ) { let failure: unknown let stopped = false const extend = async () => { if (stopped || failure !== undefined) return const renewed = await FundingDepositAddresses.renewPoll(db, { id: address.id, leaseUntil: new Date(options.now().getTime() + options.leaseMs).toISOString(), pollLeaseVersion: address.pollLeaseVersion, }) if (!renewed) failure = new StaleLeaseError() } const captureFailure = (cause: unknown) => { failure = cause } let renewal = extend().catch(captureFailure) const timer = setInterval( () => { renewal = renewal.then(extend).catch(captureFailure) }, Math.max(1, Math.floor(options.leaseMs / 2)), ) return { assert() { if (failure !== undefined) throw failure }, async ready() { await renewal if (failure !== undefined) throw failure }, async stop() { if (!stopped) { stopped = true clearInterval(timer) } await renewal return failure }, } } type Requests = { highWaterRequestIds: readonly string[] highWaterUpdatedAt?: string | undefined pendingRequestIds: readonly string[] requests: readonly Provider.listDepositAddressRequests.ReturnType['requests'][number][] } type ListRequestsOptions = { address: FundingDepositAddresses.Record requestId?: string | undefined signal?: AbortSignal | undefined } async function listRequests( provider: Provider.Provider & { listDepositAddressRequests: NonNullable }, options: ListRequestsOptions, ): Promise { const highWater = readHighWater(options.address.providerState) const highWaterRequestIds = new Set(highWater.requestIds) const pendingRequestIds = new Set(highWater.pendingRequestIds) const requests: Provider.listDepositAddressRequests.ReturnType['requests'][number][] = [] let continuation: string | undefined for (let page = 0; page < maximumPages; page++) { const result = await provider.listDepositAddressRequests( { address: options.address.address, ...(continuation ? { continuation } : {}) }, options.signal ?? new AbortController().signal, ) requests.push( ...result.requests.filter( (request) => !highWater.updatedAt || request.updatedAt > highWater.updatedAt || (request.updatedAt === highWater.updatedAt && (!highWaterRequestIds.has(request.id) || pendingRequestIds.has(request.id))), ), ) const reached = result.requests.some( (request) => highWater.updatedAt && request.updatedAt < highWater.updatedAt, ) continuation = result.continuation if (!continuation || reached) break if (page === maximumPages - 1) throw new PaginationLimitError() } const observed = new Set(requests.map((request) => request.id)) for (const requestId of unique([ ...highWater.pendingRequestIds, ...(options.requestId ? [options.requestId] : []), ])) { if (observed.has(requestId)) continue const result = await provider.listDepositAddressRequests( { address: options.address.address, requestId }, options.signal ?? new AbortController().signal, ) requests.push(...result.requests) } const addressRequests = requests.filter( (request) => !highWater.pendingRequestIds.includes(request.id), ) const newest = addressRequests.reduce( (value, request) => (!value || request.updatedAt > value ? request.updatedAt : value), highWater.updatedAt, ) return { highWaterRequestIds: newest ? unique([ ...(newest === highWater.updatedAt ? highWater.requestIds : []), ...addressRequests .filter((request) => request.updatedAt === newest) .map((request) => request.id), ]) : [], highWaterUpdatedAt: newest, pendingRequestIds: highWater.pendingRequestIds, requests, } } type HighWater = { pendingRequestIds: readonly string[] requestIds: readonly string[] updatedAt?: string | undefined } function readHighWater(state: FundingDepositAddresses.Record['providerState']): HighWater { if (!state || typeof state !== 'object') return { pendingRequestIds: [], requestIds: [] } const reconciliation = state['reconciliation'] if (!reconciliation || typeof reconciliation !== 'object') return { pendingRequestIds: [], requestIds: [] } const pendingRequestIds = 'pendingRequestIds' in reconciliation && Array.isArray(reconciliation.pendingRequestIds) ? reconciliation.pendingRequestIds.filter( (value: unknown): value is string => typeof value === 'string', ) : [] const requestIds = 'requestIds' in reconciliation && Array.isArray(reconciliation.requestIds) ? reconciliation.requestIds.filter( (value: unknown): value is string => typeof value === 'string', ) : [] return { pendingRequestIds, requestIds, ...('updatedAt' in reconciliation && typeof reconciliation.updatedAt === 'string' ? { updatedAt: reconciliation.updatedAt } : {}), } } type ReconciledWebhookTiming = Provider.Webhook.receive.WebhookTiming & { /** When Tempo completed the webhook-triggered reconciliation. */ reconciliationCompletedAt: string /** Provider request matched to the observed update. */ requestId: string } type ReconciliationStateOptions = { /** Provider requests that still require reconciliation. */ pendingRequestIds: readonly string[] /** Provider requests and high-water state observed in this pass. */ requests: Requests /** Completed webhook timing retained for diagnostics. */ webhookTiming?: ReconciledWebhookTiming | undefined } function reconciliationState( state: FundingDepositAddresses.Record['providerState'], options: ReconciliationStateOptions, ) { const current = state && typeof state === 'object' ? state : {} const reconciliation = (() => { const value = current['reconciliation'] if (value && typeof value === 'object') return value return {} })() return { ...current, reconciliation: { ...reconciliation, pendingRequestIds: options.pendingRequestIds, requestIds: options.requests.highWaterRequestIds, ...(options.requests.highWaterUpdatedAt ? { updatedAt: options.requests.highWaterUpdatedAt } : {}), ...(options.webhookTiming ? { webhookTiming: options.webhookTiming } : {}), }, } } type ObservedSourceTransaction = { deposits: FundingDeposits.Record[] transactionHash: string transfers: readonly TransferEvidence[] } type VerifiedDeposit = { deposit: FundingDeposits.Record required: bigint source: TransferEvidence sourceAmount: bigint } type ReconcileDeliveryOptions = { address: FundingDepositAddresses.Record metrics?: Metrics.Metrics | undefined onDeposit?: reconcileRequest.Options['onDeposit'] | undefined provider?: Provider.Provider | undefined request?: Provider.listDepositAddressRequests.ReturnType['requests'][number] | undefined signal?: AbortSignal | undefined subsidies?: Subsidy.Settler | undefined verified: readonly VerifiedDeposit[] verifyTransfers: VerifyTransfers } async function observeProviderDelivery(options: ReconcileDeliveryOptions) { const { verified } = options const [entry] = verified if ( !entry || verified.length !== 1 || options.address.deliveryStrategy !== 'provider' || options.address.providerOutputToken.tokenKey !== options.address.snapshot.destinationToken.tokenKey || (options.request?.status === 'success' && options.request.destinationTransactionHashes.length > 0) || !options.provider || !Provider.canObserveDepositAddressRequest(options.provider) ) return undefined const result = await options.provider .observeDepositAddressRequest( { address: options.address.address, destinationChain: options.address.snapshot.destinationChain, destinationToken: options.address.providerOutputToken, providerState: entry.deposit.providerState ?? undefined, recipient: options.address.recipient, refundAddress: options.address.refundAddress, ...(options.request ? { request: options.request } : {}), source: entry.source, sourceChain: options.address.snapshot.sourceChain, sourceToken: options.address.snapshot.sourceToken, }, options.signal ?? new AbortController().signal, ) .catch((cause) => { if (options.signal?.aborted) throw cause return undefined }) options.metrics?.count('funding_deposit_delivery_observation_count', 1, { outcome: result?.type ?? 'failed', provider: options.address.providerId, }) return result } async function reconcileUnattributed(db: Db.Db, options: reconcileUnattributed.Options) { const deposits = await FundingDeposits.listUnattributed(db, options.address.id) if (deposits.length === 0) return { terminal: true, updatedIds: new Set() } const transfersByTransaction = new Map( await Promise.all( unique(deposits.map((deposit) => deposit.sourceTransactionHash)).map( async (transactionHash) => [ transactionHash, await options.verifyTransfers({ chain: options.address.snapshot.sourceChain, recipient: options.address.address, token: options.address.snapshot.sourceToken, transactionHash, }), ] as const, ), ), ) let followUp: reconcileRequest.ReturnType['followUp'] let terminal = true const updatedIds = new Set() for (const deposit of deposits) { const source = transfersByTransaction .get(deposit.sourceTransactionHash) ?.find((transfer) => transfer.transferIndex === deposit.sourceTransferIndex) if (!source) { terminal = false continue } const delivery = await reconcileDelivery(db, { ...options, verified: [ { deposit, required: destinationBaseUnits( BigInt(source.amount), options.address.snapshot.sourceToken.decimals, options.address.snapshot.destinationToken.decimals, ), source, sourceAmount: BigInt(source.amount), }, ], }) if (delivery.followUp === 'delivery' || !followUp) followUp = delivery.followUp if (!delivery.terminal) terminal = false for (const id of delivery.updatedIds) updatedIds.add(id) } return { ...(followUp ? { followUp } : {}), terminal, updatedIds, } } declare namespace reconcileUnattributed { type Options = Omit } /** Reconciles one provider request against verified chain evidence. */ export async function reconcileRequest( db: Db.Db, options: reconcileRequest.Options, ): Promise { const trigger = options.trigger ?? 'poll' const observation = await FundingDepositRequestObservations.get(db, { depositAddressId: options.address.id, providerRequestId: options.request.id, }) if (observation) await FundingDeposits.recordRequestObservation(db, { depositAddressId: observation.depositAddressId, ...(observation.pollObservedAt ? { pollObservedAt: observation.pollObservedAt } : {}), providerRequestId: observation.providerRequestId, ...(observation.webhookReceivedAt ? { webhookReceivedAt: observation.webhookReceivedAt } : {}), }) let created = 0 const observed: ObservedSourceTransaction[] = [] // Persist provider observations first so transient chain failures do not hide acknowledged deposits. for (const transactionHash of unique(options.request.sourceTransactionHashes)) { const deposits = await FundingDeposits.withSourceTransaction(db, { depositAddressId: options.address.id, fn: async (tx): Promise => { let records = await FundingDeposits.listByProviderObservation(tx, { depositAddressId: options.address.id, providerRequestId: options.request.id, sourceTransactionHash: transactionHash, }) if (records.length === 0) records = await FundingDeposits.listBySourceTransaction(tx, { depositAddressId: options.address.id, sourceTransactionHash: transactionHash, }) if (records.length > 0) return records created += 1 return [ await Deposit.create(tx, { detectionTrigger: trigger, now: new Date(options.request.createdAt), ...(observation?.pollObservedAt ? { pollObservedAt: observation.pollObservedAt } : {}), providerRequestId: options.request.id, snapshot: { depositAddressId: options.address.id, destinationChain: options.address.snapshot.destinationChain, destinationToken: options.address.snapshot.destinationToken, provider: options.address.snapshot.provider, recipient: options.address.recipient, refundAddress: options.address.refundAddress, sourceChain: options.address.snapshot.sourceChain, sourceToken: options.address.snapshot.sourceToken, sourceTransactionHashes: [transactionHash], }, sourceTransactionHash: transactionHash, ...(observation?.webhookReceivedAt ? { webhookReceivedAt: observation.webhookReceivedAt } : {}), }), ] }, sourceTransactionHash: transactionHash, }) observed.push({ deposits, transactionHash, transfers: [] }) } const sources = await Promise.all( observed.map( async (source): Promise => ({ ...source, transfers: await options.verifyTransfers({ chain: options.address.snapshot.sourceChain, recipient: options.address.address, token: options.address.snapshot.sourceToken, transactionHash: source.transactionHash, }), }), ), ) const sourceEvidenceComplete = sources.length > 0 && sources.every((source) => source.transfers.length > 0) const updatedIds = new Set() const verified: VerifiedDeposit[] = [] for (const source of sources) { const transfers = [...source.transfers].sort( (left, right) => left.transferIndex - right.transferIndex, ) for (const [providerTransferIndex, transfer] of transfers.entries()) { const required = destinationBaseUnits( BigInt(transfer.amount), options.address.snapshot.sourceToken.decimals, options.address.snapshot.destinationToken.decimals, ) const sourceAmount = Value.tokenAmount({ baseUnits: transfer.amount, currency: options.address.snapshot.sourceToken.currency, decimals: options.address.snapshot.sourceToken.decimals, }) const destinationAmountRequired = Value.tokenAmount({ baseUnits: required, currency: options.address.snapshot.destinationToken.currency, decimals: options.address.snapshot.destinationToken.decimals, }) // Bind the placeholder to transfers in deterministic log order when one transaction contains several deposits. let deposit = source.deposits.find((record) => record.providerRequestId === null ? record.sourceTransferIndex === transfer.transferIndex : record.providerTransferIndex === providerTransferIndex, ) if (!deposit) { deposit = await Deposit.create(db, { detectionTrigger: trigger, now: new Date(options.request.createdAt), ...(observation?.pollObservedAt ? { pollObservedAt: observation.pollObservedAt } : {}), providerRequestId: options.request.id, providerTransferIndex, snapshot: { depositAddressId: options.address.id, ...(options.address.snapshot.subsidize ? { destinationAmountRequired } : {}), destinationChain: options.address.snapshot.destinationChain, destinationToken: options.address.snapshot.destinationToken, provider: options.address.snapshot.provider, recipient: options.address.recipient, refundAddress: options.address.refundAddress, ...(transfer.sender ? { sender: transfer.sender } : {}), sourceAmount, sourceChain: options.address.snapshot.sourceChain, sourceToken: options.address.snapshot.sourceToken, sourceTransactionHashes: [transfer.transactionHash], }, sourceTransactionHash: transfer.transactionHash, sourceTransferIndex: transfer.transferIndex, ...(observation?.webhookReceivedAt ? { webhookReceivedAt: observation.webhookReceivedAt } : {}), }) source.deposits.push(deposit) created += 1 } if ( !Deposit.isTerminal(deposit.status) || deposit.providerRequestId === null || !deposit.providerRequestIds.includes(options.request.id) ) { const result = await transitionDeposit( db, { expectedVersion: deposit.version, id: deposit.id, providerRequestId: options.request.id, providerRequestIds: unique([...deposit.providerRequestIds, options.request.id]), providerTransferIndex, snapshot: { ...deposit.snapshot, ...(options.address.snapshot.subsidize ? { destinationAmountRequired } : {}), ...(transfer.sender ? { sender: transfer.sender } : {}), sourceAmount, }, sourceTransferIndex: transfer.transferIndex, status: deposit.status === 'detected' ? 'bridging' : deposit.status, }, options.onDeposit, ) if (result.type === 'applied') { deposit = result.record updatedIds.add(deposit.id) } } verified.push({ deposit, required, source: transfer, sourceAmount: BigInt(transfer.amount) }) } } if (verified.length === 0) return { created, terminal: false, updated: updatedIds.size } if (!sourceEvidenceComplete) { return { created, terminal: false, updated: updatedIds.size } } const recovery = await reconcileProviderRecovery(db, { ...options, verified }) if (recovery) { for (const id of recovery.updatedIds) updatedIds.add(id) return { created, ...(recovery.followUp ? { followUp: recovery.followUp } : {}), terminal: recovery.terminal, updated: updatedIds.size, } } const delivery = await reconcileDelivery(db, { ...options, verified }) for (const id of delivery.updatedIds) updatedIds.add(id) return { created, ...(delivery.followUp ? { followUp: delivery.followUp } : {}), terminal: delivery.terminal, updated: updatedIds.size, } } export declare namespace reconcileRequest { /** Provider request, address, and settlement dependencies. */ type Options = { /** Reusable deposit address attributed by the provider. */ address: FundingDepositAddresses.Record /** Optional bounded operational metrics sink. */ metrics?: Metrics.Metrics | undefined /** Observes the latest committed form of each verified deposit. */ onDeposit?: | ((deposit: FundingDeposits.Record, references: readonly Webhooks.QueueReference[]) => void) | undefined /** Provider integration that discovered the request. */ provider?: Provider.Provider | undefined /** Provider request containing discovered chain transactions. */ request: Provider.listDepositAddressRequests.ReturnType['requests'][number] /** Abort signal forwarded to provider delivery observation. */ signal?: AbortSignal | undefined /** Optional pathUSD-funded subsidy settler. */ subsidies?: Subsidy.Settler | undefined /** Mechanism that requested reconciliation. */ trigger?: Deposit.DetectionTrigger | undefined /** Authoritative source and destination transfer verifier. */ verifyTransfers: VerifyTransfers } /** Material changes produced by one provider request. */ type ReturnType = { /** Number of newly detected source transfers. */ created: number /** Work that needs a bounded delayed follow-up. */ followUp?: 'delivery' | 'settlement' | undefined /** Whether every deposit from this request reached a terminal state. */ terminal: boolean /** Number of existing or new deposits materially changed. */ updated: number } } async function transitionDeposit( db: Db.Db, options: Deposit.transition.Options, onDeposit: reconcileRequest.Options['onDeposit'], ): Promise { const committed = await DepositWebhook.transition(db, options) if (committed.result.type === 'applied') onDeposit?.(committed.result.record, committed.references) return committed.result } async function reconcileProviderRecovery(db: Db.Db, options: ReconcileDeliveryOptions) { const failed = (providerFailureStatuses as readonly string[]).includes( options.request?.status ?? '', ) const refunded = (providerRefundStatuses as readonly string[]).includes( options.request?.status ?? '', ) if (!refunded && !failed) return undefined if (failed) { const destinationEvidence = await Promise.all( unique(options.request?.destinationTransactionHashes ?? []).map((transactionHash) => options.verifyTransfers({ chain: options.address.snapshot.destinationChain, recipient: options.address.recipient, token: options.address.providerOutputToken, transactionHash, }), ), ) if (destinationEvidence.some((transfers) => transfers.length > 0)) return undefined } const updatedIds = new Set() if (options.verified.every((entry) => Deposit.isTerminal(entry.deposit.status))) return { terminal: true, updatedIds } // Relay auto-refunds may include an internal settlement transaction; only matching source-token transfers become public evidence. const providerRefundTransactionHashes = unique(options.request?.refundTransactionHashes ?? []) const refundEvidence = await Promise.all( providerRefundTransactionHashes.map((transactionHash) => options.verifyTransfers({ chain: options.address.snapshot.sourceChain, recipient: options.address.refundAddress, token: options.address.snapshot.sourceToken, transactionHash, }), ), ) const refundTransfers = refundEvidence.flat() const refundOutput = refundTransfers.reduce( (total, transfer) => total + BigInt(transfer.amount), 0n, ) const refundTransactionHashes = unique( refundTransfers.map((transfer) => transfer.transactionHash), ) // Relay can include internal transactions, so its reported amount proves when all matching refund transfers are visible. const expectedRefundOutput = options.request?.refundAmountExpected ? BigInt(options.request.refundAmountExpected) : undefined const refundEvidenceComplete = (() => { if (refundOutput === 0n) return false if (expectedRefundOutput !== undefined) return refundOutput === expectedRefundOutput return refundEvidence.length > 0 && refundEvidence.every((transfers) => transfers.length > 0) })() const totalSource = options.verified.reduce((total, entry) => total + entry.sourceAmount, 0n) let allocated = 0n let cumulativeSource = 0n let terminal = refundEvidenceComplete // Include terminal rows in proportional allocation so retries preserve the original split without reallocating their verified refunds. for (const entry of options.verified) { cumulativeSource += entry.sourceAmount const nextAllocated = totalSource === 0n ? 0n : (refundOutput * cumulativeSource) / totalSource const amount = nextAllocated - allocated allocated = nextAllocated if (Deposit.isTerminal(entry.deposit.status)) continue let deposit = entry.deposit if (deposit.status !== 'refunding') { const result = await transitionDeposit( db, { expectedVersion: deposit.version, id: deposit.id, providerTransactionHashes: unique([ ...deposit.providerTransactionHashes, ...providerRefundTransactionHashes, ]), status: 'refunding', }, options.onDeposit, ) if (result.type === 'applied') { deposit = result.record updatedIds.add(deposit.id) } else if (result.type === 'stale') deposit = result.current } if (!refundEvidenceComplete) continue if (Deposit.isTerminal(deposit.status)) continue if (deposit.status !== 'refunding') { terminal = false continue } const result = await transitionDeposit( db, { expectedVersion: deposit.version, id: deposit.id, snapshot: { ...deposit.snapshot, refundAmount: Value.tokenAmount({ baseUnits: amount, currency: options.address.snapshot.sourceToken.currency, decimals: options.address.snapshot.sourceToken.decimals, }), refundTransactionHashes, }, status: 'refunded', }, options.onDeposit, ) if (result.type === 'applied') updatedIds.add(deposit.id) else if (result.type !== 'stale' || !Deposit.isTerminal(result.current.status)) terminal = false } const outcome = (() => { if (refundEvidenceComplete) return 'verified' // Relay failure can be followed by an automatic refund; durable exhaustion owns escalation to action-required. if (failed) return 'provider_failed' return 'pending' })() options.metrics?.count('funding_deposit_refund_count', 1, { outcome, provider: options.address.providerId, }) return { ...(!refundEvidenceComplete ? { followUp: 'delivery' as const } : {}), terminal, updatedIds, } } async function reconcileDelivery(db: Db.Db, options: ReconcileDeliveryOptions) { let followUp: reconcileRequest.ReturnType['followUp'] let terminal = true const updatedIds = new Set() if (options.verified.every((entry) => Deposit.isTerminal(entry.deposit.status))) return { terminal, updatedIds } const deliveryObservation = await observeProviderDelivery(options) if (deliveryObservation?.type === 'pending') followUp = 'delivery' const providerTransactionHashes = options.request?.destinationTransactionHashes ?? [] const destinationTransactionHashes = unique([ ...providerTransactionHashes, ...(deliveryObservation?.type === 'verified' ? deliveryObservation.destinationTransactionHashes : []), ]) const destinationEvidence = await Promise.all( destinationTransactionHashes.map((transactionHash) => options.verifyTransfers({ chain: options.address.snapshot.destinationChain, recipient: options.address.recipient, token: options.address.providerOutputToken, transactionHash, }), ), ) const destinationTransfers = destinationEvidence.flat() const providerEvidenceComplete = (options.request?.status === 'success' || deliveryObservation?.type === 'verified') && destinationEvidence.length > 0 && destinationEvidence.every((transfers) => transfers.length > 0) const providerOutput = destinationTransfers.reduce( (total, transfer) => total + BigInt(transfer.amount), 0n, ) const totalRequired = options.verified.reduce((total, entry) => total + entry.required, 0n) let allocated = 0n let cumulativeRequired = 0n for (const entry of options.verified) { const { required, sourceAmount } = entry cumulativeRequired += required const nextAllocated = totalRequired === 0n ? 0n : (providerOutput * cumulativeRequired) / totalRequired const delivered = nextAllocated - allocated allocated = nextAllocated const { deposit } = entry if (Deposit.isTerminal(deposit.status)) continue const settlement = await reconcileSubsidy({ address: options.address, delivered, deposit, providerEvidenceComplete, required, sourceAmount, subsidies: options.subsidies, verifyTransfers: options.verifyTransfers, }) if (settlement.outcome) options.metrics?.count('funding_deposit_subsidy_count', 1, { outcome: settlement.outcome, provider: options.address.providerId, }) if (settlement.status === 'bridging' || settlement.status === 'settling') terminal = false if (settlement.status === 'settling') followUp = 'settlement' const destinationAmount = Value.tokenAmount({ baseUnits: delivered + settlement.delivered, currency: options.address.snapshot.destinationToken.currency, decimals: options.address.snapshot.destinationToken.decimals, }) const result = await transitionDeposit( db, { expectedVersion: deposit.version, id: deposit.id, ...(delivered > 0n ? { providerOutputAmount: Value.tokenAmount({ baseUnits: delivered, currency: options.address.providerOutputToken.currency, decimals: options.address.providerOutputToken.decimals, }), } : {}), providerRequestIds: options.request ? unique([...deposit.providerRequestIds, options.request.id]) : deposit.providerRequestIds, ...(deliveryObservation?.providerState ? { providerState: { ...deposit.providerState, ...deliveryObservation.providerState, }, } : {}), providerTransactionHashes: unique([ ...deposit.providerTransactionHashes, ...providerTransactionHashes, ...(deliveryObservation?.providerTransactionHashes ?? []), ]), ...(settlement.settlementTransaction ? { settlementTransaction: settlement.settlementTransaction } : {}), ...(settlement.settlementTransactionHash ? { settlementTransactionHash: settlement.settlementTransactionHash } : {}), snapshot: { ...deposit.snapshot, ...(delivered + settlement.delivered > 0n ? { destinationAmount } : {}), ...(settlement.status === 'completed' && !options.address.snapshot.subsidize ? { destinationAmountRequired: destinationAmount } : {}), destinationTransactionHashes: unique([ ...(deposit.snapshot.destinationTransactionHashes ?? []), ...destinationTransfers.map((evidence) => evidence.transactionHash), ...settlement.destinationTransactionHashes, ]), }, status: settlement.status, ...(settlement.statusReason ? { statusReason: settlement.statusReason } : {}), ...(settlement.subsidyAmount ? { subsidyAmount: settlement.subsidyAmount } : {}), ...(settlement.tempoGasPaid ? { tempoGasPaid: settlement.tempoGasPaid } : {}), }, options.onDeposit, ) if (result.type === 'applied') { updatedIds.add(deposit.id) if (settlement.broadcast && options.subsidies) { await options.subsidies.broadcast(settlement.broadcast).catch((cause) => { options.metrics?.count('funding_deposit_subsidy_count', 1, { outcome: 'broadcast_failed', provider: options.address.providerId, }) throw cause }) options.metrics?.count('funding_deposit_subsidy_count', 1, { outcome: 'broadcast', provider: options.address.providerId, }) } } } return { ...(followUp ? { followUp } : {}), terminal, updatedIds } } type ReconcileSubsidyOptions = { address: FundingDepositAddresses.Record delivered: bigint deposit: FundingDeposits.Record providerEvidenceComplete: boolean required: bigint sourceAmount: bigint subsidies?: Subsidy.Settler | undefined verifyTransfers: VerifyTransfers } type ReconcileSubsidyResult = { broadcast?: Subsidy.Settler.broadcast.Parameters | undefined delivered: bigint destinationTransactionHashes: readonly string[] outcome?: string | undefined settlementTransaction?: string | undefined settlementTransactionHash?: string | undefined status: Deposit.Status statusReason?: Deposit.StatusReason | undefined subsidyAmount?: Deposit.Snapshot['destinationAmountRequired'] | undefined tempoGasPaid?: string | undefined } async function reconcileSubsidy(options: ReconcileSubsidyOptions): Promise { const empty = { delivered: 0n, destinationTransactionHashes: [] } as const if (options.required === 0n) return { ...empty, status: 'action-required', statusReason: { code: 'source_amount_not_supported', message: 'The source amount is smaller than one destination-token base unit.', }, } const deficit = options.required > options.delivered ? options.required - options.delivered : 0n const subsidyAmount = options.deposit.subsidyAmount ?? Value.tokenAmount({ baseUnits: deficit, currency: options.address.snapshot.destinationToken.currency, decimals: options.address.snapshot.destinationToken.decimals, }) if (options.deposit.settlementTransaction && options.deposit.settlementTransactionHash) { if (!options.subsidies) return { ...empty, outcome: 'unavailable', status: 'action-required', statusReason: { code: 'subsidy_failed', message: 'Tempo subsidy settlement is unavailable.', }, subsidyAmount, } const hash = options.deposit.settlementTransactionHash as Hex.Hex const status = await options.subsidies.status({ hash }) if (status.type === 'reverted') return { ...empty, outcome: 'reverted', status: 'action-required', statusReason: { code: 'subsidy_failed', message: 'Tempo could not deliver the required subsidy.', }, subsidyAmount, tempoGasPaid: status.tempoGasPaid, } if (status.type === 'pending') return { ...empty, broadcast: { hash, transaction: options.deposit.settlementTransaction as Hex.Hex, }, outcome: 'pending', status: 'settling', subsidyAmount, } const transfers = await options.verifyTransfers({ chain: options.address.snapshot.destinationChain, recipient: options.address.recipient, token: options.address.snapshot.destinationToken, transactionHash: hash, }) const delivered = transfers.reduce((total, transfer) => total + BigInt(transfer.amount), 0n) return { delivered, destinationTransactionHashes: transfers.map((transfer) => transfer.transactionHash), outcome: options.delivered + delivered >= options.required ? 'completed' : 'underfilled', status: options.delivered + delivered >= options.required ? 'completed' : 'settling', subsidyAmount, tempoGasPaid: status.tempoGasPaid, } } if (options.delivered >= options.required) return { ...empty, status: 'completed' } if (!options.address.snapshot.subsidize) { if (!options.providerEvidenceComplete) return { ...empty, status: 'bridging' } if (options.delivered > 0n) return { ...empty, status: 'completed' } return { ...empty, status: 'action-required', statusReason: { code: 'delivery_failed', message: 'The provider completed without delivering destination funds.', }, } } if (!options.providerEvidenceComplete) return { ...empty, status: 'bridging' } if (!options.subsidies) return { ...empty, outcome: 'unavailable', status: 'action-required', statusReason: { code: 'subsidy_failed', message: 'Tempo subsidy settlement is unavailable.', }, } if ( !Subsidy.supports(options.subsidies.policy, { amount: options.sourceAmount.toString(), destinationToken: options.address.snapshot.destinationToken, sourceToken: options.address.snapshot.sourceToken, }) ) return { ...empty, outcome: 'unsupported', status: 'action-required', statusReason: { code: 'source_amount_not_supported', message: 'The source amount is outside the subsidized deposit limit.', }, } const prepared = await options.subsidies .prepare({ amount: deficit, depositId: options.deposit.id, recipient: options.address.recipient as Hex.Hex, token: options.address.snapshot.destinationToken.address as Hex.Hex, }) .catch((cause) => { if (cause instanceof Subsidy.BalanceUnavailableError) return undefined if (cause instanceof Subsidy.LiquidityUnavailableError) return null throw cause }) if (prepared === null) return { ...empty, outcome: 'liquidity_unavailable', status: 'settling', subsidyAmount, } if (!prepared) return { ...empty, outcome: 'balance_unavailable', status: 'action-required', statusReason: { code: 'subsidy_balance_unavailable', message: 'Tempo lacks enough pathUSD for the required subsidy.', }, subsidyAmount, } return { ...empty, broadcast: prepared, outcome: 'prepared', settlementTransaction: prepared.transaction, settlementTransactionHash: prepared.hash, status: 'settling', subsidyAmount, } } function destinationBaseUnits(amount: bigint, sourceDecimals: number, destinationDecimals: number) { if (destinationDecimals >= sourceDecimals) return amount * 10n ** BigInt(destinationDecimals - sourceDecimals) const divisor = 10n ** BigInt(sourceDecimals - destinationDecimals) // Destination tokens cannot represent finer source precision, so retain only // the amount expressible in destination base units. return amount / divisor } function unique(values: readonly string[]) { return [...new Set(values)] } function nextFollowUp( message: Provider.Webhook.receive.Queued, kind: Provider.Webhook.receive.FollowUp['kind'], ) { const attempt = message.followUp?.kind === kind ? message.followUp.attempt + 1 : 0 const chainProvider = kind === 'provider' && message.trigger === 'chain' const maximumAttempts = chainProvider ? chainProviderMaximumFollowUpAttempts : maximumFollowUpAttempts if (attempt >= maximumAttempts) return undefined return { delaySeconds: chainProvider ? chainProviderFollowUpDelaySeconds : followUpDelaySeconds[kind], state: { attempt, kind }, } } type RecordWebhookLatencyOptions = { /** When Tempo completed the webhook-triggered reconciliation. */ completedAt: Date /** Funding provider that sent the webhook. */ providerId: string /** Authenticated timestamps carried from webhook receipt. */ timing: Provider.Webhook.receive.WebhookTiming } function recordWebhookLatency( metrics: Metrics.Metrics | undefined, options: RecordWebhookLatencyOptions, ) { if (!metrics) return const completedAt = options.completedAt.getTime() const enqueuedAt = Date.parse(options.timing.enqueuedAt) const providerUpdatedAt = Date.parse(options.timing.providerUpdatedAt) const receivedAt = Date.parse(options.timing.receivedAt) const sentAt = Date.parse(options.timing.sentAt) const segments = [ { from: providerUpdatedAt, segment: 'provider_to_sent', to: sentAt }, { from: sentAt, segment: 'sent_to_received', to: receivedAt }, { from: receivedAt, segment: 'received_to_enqueued', to: enqueuedAt }, { from: enqueuedAt, segment: 'enqueued_to_reconciled', to: completedAt }, { from: providerUpdatedAt, segment: 'provider_to_reconciled', to: completedAt }, ] as const for (const segment of segments) metrics.histogram( 'funding_deposit_webhook_latency_ms', Math.max(0, segment.to - segment.from), { provider: options.providerId, segment: segment.segment, status: options.timing.providerStatus, }, ) } /** Provider pagination exceeded the bounded reconciliation limit. */ export class PaginationLimitError extends Error { override name = 'Funding.Reconciliation.PaginationLimitError' } /** The reconciliation lease was claimed by another worker. */ export class StaleLeaseError extends Error { override name = 'Funding.Reconciliation.StaleLeaseError' }