From 9787bb9cd47378b7438b22bce1c3b7cd3e0d5524 Mon Sep 17 00:00:00 2001 From: Thomas Shelby Date: Wed, 29 Jul 2026 14:17:03 +0000 Subject: [PATCH 1/4] Add transaction effect aggregator with receipt integration --- src/effectAggregator.ts | 87 +++++++++++++++++++++++++ src/index.ts | 5 ++ src/receipt.ts | 36 +++++++++-- src/types.ts | 20 ++++++ test/effectAggregator.test.ts | 118 ++++++++++++++++++++++++++++++++++ test/receipt.test.ts | 33 +++++++++- 6 files changed, 293 insertions(+), 6 deletions(-) create mode 100644 src/effectAggregator.ts create mode 100644 test/effectAggregator.test.ts diff --git a/src/effectAggregator.ts b/src/effectAggregator.ts new file mode 100644 index 0000000..82f09d1 --- /dev/null +++ b/src/effectAggregator.ts @@ -0,0 +1,87 @@ +/** + * Transaction operation effect aggregator. + * + * Fetches all Horizon effects for a submitted transaction and consolidates + * them into a per-account net asset balance delta summary, so callers get a + * single "what changed" view instead of raw per-operation effect records. + */ + +import type { Horizon } from "@stellar/stellar-sdk"; +import type { AccountEffectSummary, AssetDelta } from "./types.js"; +import { collectAll } from "./horizonPaginator.js"; + +/** Effect types that represent a net balance change for an account. */ +const CREDIT_EFFECT = "account_credited"; +const DEBIT_EFFECT = "account_debited"; + +/** + * Convert a Horizon decimal amount string (always 7 fractional digits) to a + * stroop bigint, without floating-point rounding. + */ +function decimalToStroops(amount: string): bigint { + const [intPart, fracPart = ""] = amount.split("."); + const frac = fracPart.padEnd(7, "0").slice(0, 7); + return BigInt(intPart || "0") * 10_000_000n + BigInt(frac || "0"); +} + +function effectAssetKey(effect: Record): string { + if (effect.asset_type === "native") return "native"; + const code = effect.asset_code as string | undefined; + const issuer = effect.asset_issuer as string | undefined; + if (code && issuer) return `${code}:${issuer}`; + return String(effect.asset_type ?? "unknown"); +} + +/** + * Aggregate all effects of a transaction into per-account net asset deltas. + * + * Only `account_credited`/`account_debited` effects are counted; intermediate + * DEX effects (offer creation/removal, trades) are ignored since they net out + * to the same credited/debited effects on the accounts actually affected. + * + * @param server - Horizon server instance. + * @param txHash - Hash of the transaction to aggregate effects for. + * @returns One summary per affected account, sorted by account ID. + */ +export async function aggregateEffects( + server: Horizon.Server, + txHash: string, +): Promise { + const initialPage = await server.effects().forTransaction(txHash).call(); + const effects = await collectAll(initialPage); + + const byAccount = new Map>(); + + for (const raw of effects) { + const effect = raw as unknown as Record; + const type = effect.type as string | undefined; + if (type !== CREDIT_EFFECT && type !== DEBIT_EFFECT) continue; + + const account = effect.account as string | undefined; + const amount = effect.amount as string | undefined; + if (!account || amount === undefined) continue; + + const asset = effectAssetKey(effect); + const stroops = decimalToStroops(amount); + const signed = type === DEBIT_EFFECT ? -stroops : stroops; + + let assetDeltas = byAccount.get(account); + if (!assetDeltas) { + assetDeltas = new Map(); + byAccount.set(account, assetDeltas); + } + assetDeltas.set(asset, (assetDeltas.get(asset) ?? 0n) + signed); + } + + const summaries: AccountEffectSummary[] = []; + for (const accountId of Array.from(byAccount.keys()).sort()) { + const assetDeltas: AssetDelta[] = Array.from(byAccount.get(accountId)!.entries()) + .filter(([, delta]) => delta !== 0n) + .map(([asset, delta]) => ({ asset, delta })); + if (assetDeltas.length > 0) { + summaries.push({ accountId, assetDeltas }); + } + } + + return summaries; +} diff --git a/src/index.ts b/src/index.ts index 38d8fe1..20e501b 100644 --- a/src/index.ts +++ b/src/index.ts @@ -573,8 +573,13 @@ export type { PaymentReceipt, PaymentReceiptJSON, InvoiceFetcher, + ReceiptConfig, } from "./receipt.js"; +// Transaction operation effect aggregator +export { aggregateEffects } from "./effectAggregator.js"; +export type { AccountEffectSummary, AssetDelta } from "./types.js"; + // Merkle proof functionality export { generateMerkleProof, verifyMerkleProof } from "./merkle.js"; export type { MerkleProof } from "./merkle.js"; diff --git a/src/receipt.ts b/src/receipt.ts index deaaa39..4cc6aa2 100644 --- a/src/receipt.ts +++ b/src/receipt.ts @@ -1,6 +1,8 @@ import { createHash } from "crypto"; -import type { Invoice, Payment } from "./types.js"; +import type { Horizon } from "@stellar/stellar-sdk"; +import type { AccountEffectSummary, Invoice, Payment } from "./types.js"; import { PayerAddressRequiredError } from "./errors.js"; +import { aggregateEffects } from "./effectAggregator.js"; /** A verified payment receipt compiled from on-chain invoice data. */ export interface PaymentReceipt { @@ -18,10 +20,22 @@ export interface PaymentReceipt { generatedAt: number; /** Ledger timestamp or sequence used in the proof hash calculation. */ ledgerTimestamp: number; + /** Net per-account asset balance changes for the underlying transaction, when requested via `ReceiptConfig.includeEffects`. */ + effectSummary?: AccountEffectSummary[]; /** Convert receipt to a JSON-serializable object (with bigints represented as strings). */ toJSON(): PaymentReceiptJSON; } +/** Options controlling optional receipt enrichment. */ +export interface ReceiptConfig { + /** When true (and `server`/`txHash` are supplied), attach an effect summary to the receipt. */ + includeEffects?: boolean; + /** Horizon server used to fetch effects when `includeEffects` is set. */ + server?: Horizon.Server; + /** Hash of the transaction to aggregate effects for. */ + txHash?: string; +} + /** JSON-serializable representation of a PaymentReceipt. */ export interface PaymentReceiptJSON { invoiceId: string; @@ -37,6 +51,7 @@ export interface PaymentReceiptJSON { proofHash: string; generatedAt: number; ledgerTimestamp: number; + effectSummary?: AccountEffectSummary[]; } /** Interface for any client capable of fetching an invoice by ID. */ @@ -100,22 +115,31 @@ export function compilePaymentReceipt( * @param source - Either a client with `getInvoice` or an `Invoice` object. * @param invoiceIdOrPayer - The invoice ID (if passing a client) or payer address (if passing an Invoice). * @param payerAddress - The payer address (if passing a client). + * @param config - Optional enrichment config; set `includeEffects` (with `server`/`txHash`) to attach an effect summary. * @returns Promise resolving to the PaymentReceipt. */ export async function generatePaymentReceipt( source: InvoiceFetcher | Invoice, invoiceIdOrPayer: string, - payerAddress?: string + payerAddress?: string, + config?: ReceiptConfig ): Promise { + let receipt: PaymentReceipt; if ("getInvoice" in source && typeof source.getInvoice === "function") { if (!payerAddress) { throw new PayerAddressRequiredError(); } const invoice = await source.getInvoice(invoiceIdOrPayer); - return compilePaymentReceipt(invoice, payerAddress); + receipt = compilePaymentReceipt(invoice, payerAddress); + } else { + receipt = compilePaymentReceipt(source as Invoice, invoiceIdOrPayer); + } + + if (config?.includeEffects && config.server && config.txHash) { + receipt.effectSummary = await aggregateEffects(config.server, config.txHash); } - return compilePaymentReceipt(source as Invoice, invoiceIdOrPayer); + return receipt; } /** @@ -152,6 +176,7 @@ export function deserializePaymentReceipt(json: string): PaymentReceipt { proofHash: data.proofHash, generatedAt: data.generatedAt, ledgerTimestamp: data.ledgerTimestamp, + effectSummary: data.effectSummary, }); } @@ -163,6 +188,7 @@ function _buildReceiptObject(data: { proofHash: string; generatedAt: number; ledgerTimestamp: number; + effectSummary?: AccountEffectSummary[]; }): PaymentReceipt { return { invoiceId: data.invoiceId, @@ -172,6 +198,7 @@ function _buildReceiptObject(data: { proofHash: data.proofHash, generatedAt: data.generatedAt, ledgerTimestamp: data.ledgerTimestamp, + effectSummary: data.effectSummary, toJSON(): PaymentReceiptJSON { return { invoiceId: this.invoiceId, @@ -184,6 +211,7 @@ function _buildReceiptObject(data: { proofHash: this.proofHash, generatedAt: this.generatedAt, ledgerTimestamp: this.ledgerTimestamp, + effectSummary: this.effectSummary, }; }, }; diff --git a/src/types.ts b/src/types.ts index fff3f03..29f7ce4 100644 --- a/src/types.ts +++ b/src/types.ts @@ -1647,3 +1647,23 @@ export interface CursorStore { /** Delete a saved cursor. */ delete(key: string): Promise; } + +// --------------------------------------------------------------------------- +// Effect Aggregator Types +// --------------------------------------------------------------------------- + +/** Net balance change for a single asset within an account's effect summary. */ +export interface AssetDelta { + /** Asset identifier: "native" or "CODE:ISSUER". */ + asset: string; + /** Net change in stroops; positive = credited, negative = debited. */ + delta: bigint; +} + +/** Net effect summary for one account, aggregated across a transaction's effects. */ +export interface AccountEffectSummary { + /** Stellar account ID affected. */ + accountId: string; + /** Net asset balance changes for this account. */ + assetDeltas: AssetDelta[]; +} diff --git a/test/effectAggregator.test.ts b/test/effectAggregator.test.ts new file mode 100644 index 0000000..a0d284b --- /dev/null +++ b/test/effectAggregator.test.ts @@ -0,0 +1,118 @@ +import { describe, it, expect, vi } from "vitest"; +import { aggregateEffects } from "../src/effectAggregator.js"; +import type { CollectionPage } from "../src/types.js"; + +function makePage(records: T[], nextPage: CollectionPage | null = null): CollectionPage { + return { records, next: vi.fn().mockResolvedValue(nextPage) }; +} + +function makeServer(effects: unknown[], pageSize = effects.length) { + const pages: CollectionPage[] = []; + for (let i = 0; i < effects.length; i += pageSize) { + pages.push(makePage(effects.slice(i, i + pageSize))); + } + if (pages.length === 0) pages.push(makePage([])); + for (let i = 0; i < pages.length - 1; i++) { + (pages[i].next as ReturnType).mockResolvedValue(pages[i + 1]); + } + + return { + effects: () => ({ + forTransaction: () => ({ + call: () => Promise.resolve(pages[0]), + }), + }), + } as any; +} + +describe("aggregateEffects", () => { + it("aggregates a simple credit/debit pair into net deltas", async () => { + const server = makeServer([ + { type: "account_debited", account: "GPAYER", asset_type: "native", amount: "10.0000000" }, + { type: "account_credited", account: "GRECIPIENT", asset_type: "native", amount: "10.0000000" }, + ]); + + const summaries = await aggregateEffects(server, "txhash1"); + + expect(summaries).toEqual([ + { accountId: "GPAYER", assetDeltas: [{ asset: "native", delta: -100_000_000n }] }, + { accountId: "GRECIPIENT", assetDeltas: [{ asset: "native", delta: 100_000_000n }] }, + ]); + }); + + it("sums deltas across multiple operations for the same account and asset", async () => { + const server = makeServer([ + { type: "account_debited", account: "GPAYER", asset_type: "native", amount: "6.0000000" }, + { type: "account_debited", account: "GPAYER", asset_type: "native", amount: "4.0000000" }, + { type: "account_credited", account: "GRECIPIENT_A", asset_type: "native", amount: "6.0000000" }, + { type: "account_credited", account: "GRECIPIENT_B", asset_type: "native", amount: "4.0000000" }, + ]); + + const summaries = await aggregateEffects(server, "txhash2"); + + expect(summaries).toEqual([ + { accountId: "GPAYER", assetDeltas: [{ asset: "native", delta: -100_000_000n }] }, + { accountId: "GRECIPIENT_A", assetDeltas: [{ asset: "native", delta: 60_000_000n }] }, + { accountId: "GRECIPIENT_B", assetDeltas: [{ asset: "native", delta: 40_000_000n }] }, + ]); + }); + + it("handles a two-recipient payment with path conversion across two credited assets", async () => { + const server = makeServer([ + { type: "account_debited", account: "GPAYER", asset_type: "native", amount: "20.0000000" }, + { type: "trade", account: "GPAYER", offer_id: "1" }, + { + type: "account_credited", + account: "GRECIPIENT_A", + asset_type: "credit_alphanum4", + asset_code: "USDC", + asset_issuer: "GISSUER", + amount: "5.0000000", + }, + { + type: "account_credited", + account: "GRECIPIENT_B", + asset_type: "credit_alphanum4", + asset_code: "USDC", + asset_issuer: "GISSUER", + amount: "5.0000000", + }, + ], 2); + + const summaries = await aggregateEffects(server, "txhash3"); + + expect(summaries).toEqual([ + { accountId: "GPAYER", assetDeltas: [{ asset: "native", delta: -200_000_000n }] }, + { accountId: "GRECIPIENT_A", assetDeltas: [{ asset: "USDC:GISSUER", delta: 50_000_000n }] }, + { accountId: "GRECIPIENT_B", assetDeltas: [{ asset: "USDC:GISSUER", delta: 50_000_000n }] }, + ]); + }); + + it("omits accounts whose deltas net to zero", async () => { + const server = makeServer([ + { type: "account_credited", account: "GROUNDTRIP", asset_type: "native", amount: "5.0000000" }, + { type: "account_debited", account: "GROUNDTRIP", asset_type: "native", amount: "5.0000000" }, + ]); + + const summaries = await aggregateEffects(server, "txhash4"); + + expect(summaries).toEqual([]); + }); + + it("walks multiple pages via the horizon paginator", async () => { + const server = makeServer( + [ + { type: "account_debited", account: "GPAYER", asset_type: "native", amount: "1.0000000" }, + { type: "account_credited", account: "GRECIPIENT", asset_type: "native", amount: "1.0000000" }, + ], + 1, + ); + + const summaries = await aggregateEffects(server, "txhash5"); + + expect(summaries).toEqual([ + { accountId: "GPAYER", assetDeltas: [{ asset: "native", delta: -10_000_000n }] }, + { accountId: "GRECIPIENT", assetDeltas: [{ asset: "native", delta: 10_000_000n }] }, + ]); + }); +}); diff --git a/test/receipt.test.ts b/test/receipt.test.ts index b02a86f..89f4340 100644 --- a/test/receipt.test.ts +++ b/test/receipt.test.ts @@ -1,4 +1,4 @@ -import { describe, it, expect } from "vitest"; +import { describe, it, expect, vi } from "vitest"; import { createHash } from "crypto"; import { compilePaymentReceipt, @@ -6,7 +6,7 @@ import { serializePaymentReceipt, deserializePaymentReceipt, } from "../src/receipt.js"; -import type { Invoice } from "../src/types.js"; +import type { Invoice, CollectionPage } from "../src/types.js"; const mockInvoice: Invoice = { id: "inv_99", @@ -107,4 +107,33 @@ describe("generatePaymentReceipt", () => { expect(deserialized.payments).toHaveLength(2); expect(deserialized.payments[0]!.amount).toBe(40000000n); }); + + it("attaches an effect summary when includeEffects is set with server/txHash", async () => { + const page: CollectionPage = { + records: [ + { type: "account_debited", account: "GPAYER_A", asset_type: "native", amount: "40.0000000" }, + { type: "account_credited", account: "GRECIPIENT123", asset_type: "native", amount: "40.0000000" }, + ], + next: vi.fn().mockResolvedValue(null), + }; + const server = { + effects: () => ({ forTransaction: () => ({ call: () => Promise.resolve(page) }) }), + } as any; + + const receipt = await generatePaymentReceipt(mockInvoice, "GPAYER_A", undefined, { + includeEffects: true, + server, + txHash: "txhash", + }); + + expect(receipt.effectSummary).toEqual([ + { accountId: "GPAYER_A", assetDeltas: [{ asset: "native", delta: -400_000_000n }] }, + { accountId: "GRECIPIENT123", assetDeltas: [{ asset: "native", delta: 400_000_000n }] }, + ]); + }); + + it("omits effectSummary when includeEffects is not set", () => { + const receipt = compilePaymentReceipt(mockInvoice, "GPAYER_A"); + expect(receipt.effectSummary).toBeUndefined(); + }); }); From e39e3e2e53f2e9cb2396c5f07c45241d83775f1a Mon Sep 17 00:00:00 2001 From: Thomas Shelby Date: Wed, 29 Jul 2026 14:18:51 +0000 Subject: [PATCH 2/4] Add multi-asset invoice line item normalizer --- src/errors.ts | 20 +++++++ src/index.ts | 11 ++++ src/lineItemNormalizer.ts | 86 ++++++++++++++++++++++++++++ src/priceOracle.ts | 72 ++++++++++++++++++++++++ src/types.ts | 42 ++++++++++++++ test/lineItemNormalizer.test.ts | 99 +++++++++++++++++++++++++++++++++ 6 files changed, 330 insertions(+) create mode 100644 src/lineItemNormalizer.ts create mode 100644 src/priceOracle.ts create mode 100644 test/lineItemNormalizer.test.ts diff --git a/src/errors.ts b/src/errors.ts index 47f1a5c..a196177 100644 --- a/src/errors.ts +++ b/src/errors.ts @@ -708,6 +708,22 @@ export class Sep41AdapterError extends StellarSplitError { } } +/** Thrown when a line item's asset has no oracle price available for normalisation. */ +export class UnsupportedLineItemAssetError extends StellarSplitError { + readonly asset: string; + + constructor(asset: string) { + super( + `No oracle price available for line item asset: ${asset}`, + "UNSUPPORTED_LINE_ITEM_ASSET", + { asset } + ); + this.name = "UnsupportedLineItemAssetError"; + this.asset = asset; + Object.setPrototypeOf(this, new.target.prototype); + } +} + /** Thrown when tranche status check fails. */ export class TrancheProgressError extends StellarSplitError { constructor(message: string) { @@ -1087,6 +1103,10 @@ export function isInsufficientSignaturesError(err: unknown): err is Insufficient return err instanceof InsufficientSignaturesError; } +export function isUnsupportedLineItemAssetError(err: unknown): err is UnsupportedLineItemAssetError { + return err instanceof UnsupportedLineItemAssetError; +} + export function isCloneChainTooDeepError(err: unknown): err is CloneChainTooDeepError { return err instanceof CloneChainTooDeepError; } diff --git a/src/index.ts b/src/index.ts index 20e501b..308922b 100644 --- a/src/index.ts +++ b/src/index.ts @@ -580,6 +580,17 @@ export type { export { aggregateEffects } from "./effectAggregator.js"; export type { AccountEffectSummary, AssetDelta } from "./types.js"; +// Multi-asset invoice line item normalizer +export { normalizeLineItems } from "./lineItemNormalizer.js"; +export { ContractPriceOracle } from "./priceOracle.js"; +export type { PriceOracle } from "./priceOracle.js"; +export type { + InvoiceLineItem, + NormalizedLineItem, + NormalizedInvoiceTotal, +} from "./types.js"; +export { UnsupportedLineItemAssetError, isUnsupportedLineItemAssetError } from "./errors.js"; + // Merkle proof functionality export { generateMerkleProof, verifyMerkleProof } from "./merkle.js"; export type { MerkleProof } from "./merkle.js"; diff --git a/src/lineItemNormalizer.ts b/src/lineItemNormalizer.ts new file mode 100644 index 0000000..2601e3f --- /dev/null +++ b/src/lineItemNormalizer.ts @@ -0,0 +1,86 @@ +/** + * Multi-asset invoice line item normaliser. + * + * Resolves every line item to the invoice's settlement asset using a + * `PriceOracle`, so an invoice mixing e.g. USDC and XLM line items can be + * presented and persisted as a single normalised total. + */ + +import type { InvoiceLineItem, NormalizedInvoiceTotal, NormalizedLineItem } from "./types.js"; +import type { PriceOracle } from "./priceOracle.js"; +import { UnsupportedLineItemAssetError } from "./errors.js"; + +/** Fixed-point scale used for conversion rates (1e18 = 1.0). */ +const RATE_SCALE = 1_000_000_000_000_000_000n; + +/** + * Normalise a set of line items to a single settlement asset. + * + * Same-asset items pass through without an oracle call. Cross-asset items + * are converted using `oracle.getRate`, called at most once per unique + * (asset, settlementAsset) pair. Zero-amount items (discounts, waivers) never + * trigger an oracle call. + * + * @param items - Line items, each denominated in its own asset. + * @param settlementAsset - Asset identifier to normalise all amounts to. + * @param oracle - Price oracle used for cross-asset conversion. + * @throws {UnsupportedLineItemAssetError} When a line item's asset has no oracle price available. + */ +export async function normalizeLineItems( + items: InvoiceLineItem[], + settlementAsset: string, + oracle: PriceOracle +): Promise { + const rateCache = new Map(); + const normalizedItems: NormalizedLineItem[] = []; + let total = 0n; + + for (const item of items) { + const originalAmount = item.total ?? item.unitPrice * BigInt(item.quantity); + + if (originalAmount === 0n) { + normalizedItems.push({ + description: item.description, + originalAmount, + originalAsset: item.asset, + convertedAmount: 0n, + conversionRate: RATE_SCALE, + }); + continue; + } + + if (item.asset === settlementAsset) { + normalizedItems.push({ + description: item.description, + originalAmount, + originalAsset: item.asset, + convertedAmount: originalAmount, + conversionRate: RATE_SCALE, + }); + total += originalAmount; + continue; + } + + let rate = rateCache.get(item.asset); + if (rate === undefined) { + const fetched = await oracle.getRate(item.asset, settlementAsset); + if (fetched === undefined) { + throw new UnsupportedLineItemAssetError(item.asset); + } + rate = fetched; + rateCache.set(item.asset, rate); + } + + const convertedAmount = (originalAmount * rate) / RATE_SCALE; + normalizedItems.push({ + description: item.description, + originalAmount, + originalAsset: item.asset, + convertedAmount, + conversionRate: rate, + }); + total += convertedAmount; + } + + return { settlementAsset, total, items: normalizedItems }; +} diff --git a/src/priceOracle.ts b/src/priceOracle.ts new file mode 100644 index 0000000..1a73de9 --- /dev/null +++ b/src/priceOracle.ts @@ -0,0 +1,72 @@ +/** + * Cross-asset price oracle for settlement calculations (invoice normalisation, + * multi-asset line items). Unlike `currencyConverter.ts` (display-only, never + * feeds into real amounts), rates returned here are used to compute amounts + * that settle on-chain, so callers should treat failures as fatal rather than + * falling back to a stale/cached display value. + */ + +import { + Contract, + rpc as SorobanRpc, + TransactionBuilder, + BASE_FEE, + nativeToScVal, + scValToNative, +} from "@stellar/stellar-sdk"; +import { OraclePriceError, NoReturnValueError } from "./errors.js"; + +/** Resolves a conversion rate between two on-chain assets. */ +export interface PriceOracle { + /** + * Fixed-point rate (1e18 = 1.0) to convert 1 unit of `fromAsset` into + * `toAsset`, or `undefined` when no price is available for that pair. + */ + getRate(fromAsset: string, toAsset: string): Promise; +} + +/** Soroban contract-backed price oracle. */ +export class ContractPriceOracle implements PriceOracle { + constructor( + private readonly server: SorobanRpc.Server, + private readonly oracleAddress: string, + private readonly networkPassphrase: string + ) {} + + async getRate(fromAsset: string, toAsset: string): Promise { + const contract = new Contract(this.oracleAddress); + const operation = contract.call( + "get_price", + nativeToScVal(fromAsset, { type: "symbol" }), + nativeToScVal(toAsset, { type: "symbol" }) + ); + + const sourceAccount = { + accountId: () => this.oracleAddress, + sequenceNumber: () => "0", + incrementSequenceNumber: () => {}, + } as any; + + const tx = new TransactionBuilder(sourceAccount, { + fee: BASE_FEE, + networkPassphrase: this.networkPassphrase, + }) + .addOperation(operation) + .setTimeout(30) + .build(); + + const simResult = await this.server.simulateTransaction(tx); + if (SorobanRpc.Api.isSimulationError(simResult)) { + if (/no price|not found|unsupported/i.test(simResult.error)) { + return undefined; + } + throw new OraclePriceError(`Oracle simulation failed: ${simResult.error}`); + } + + const returnVal = (simResult as SorobanRpc.Api.SimulateTransactionSuccessResponse).result + ?.retval; + if (!returnVal) throw new NoReturnValueError("oracle get_price"); + + return BigInt(scValToNative(returnVal)); + } +} diff --git a/src/types.ts b/src/types.ts index 29f7ce4..a7ee658 100644 --- a/src/types.ts +++ b/src/types.ts @@ -1410,6 +1410,48 @@ export interface InvoiceMetadata { attachmentCIDs: string[]; } +// --------------------------------------------------------------------------- +// Multi-Asset Line Item Normalizer Types +// --------------------------------------------------------------------------- + +/** A line item denominated in its own asset, prior to settlement normalisation. */ +export interface InvoiceLineItem { + /** Description of the item or service. */ + description: string; + /** Quantity of items. */ + quantity: number; + /** Unit price in stroops, denominated in `asset`. */ + unitPrice: bigint; + /** Optional total override (defaults to quantity * unitPrice), denominated in `asset`. */ + total?: bigint; + /** Asset identifier this line item is priced in: "native" or "CODE:ISSUER" or a contract address. */ + asset: string; +} + +/** A line item after conversion to the invoice's settlement asset. */ +export interface NormalizedLineItem { + /** Description of the item or service. */ + description: string; + /** Original amount in stroops, denominated in `originalAsset`. */ + originalAmount: bigint; + /** Asset identifier the line item was originally denominated in. */ + originalAsset: string; + /** Amount in stroops after conversion to the settlement asset. */ + convertedAmount: bigint; + /** Fixed-point rate (1e18 = 1.0) used for the conversion; 1e18 when no conversion was needed. */ + conversionRate: bigint; +} + +/** Aggregate result of normalising an invoice's line items to a single settlement asset. */ +export interface NormalizedInvoiceTotal { + /** Asset identifier all amounts were normalised to. */ + settlementAsset: string; + /** Sum of all `convertedAmount` values, in stroops. */ + total: bigint; + /** Per-item normalised amounts, in the same order as the input. */ + items: NormalizedLineItem[]; +} + /** Configuration for IPFS backend. */ export interface IPFSConfig { /** Backend type: 'gateway' for HTTP gateway or 'kubo' for Kubo RPC API. */ diff --git a/test/lineItemNormalizer.test.ts b/test/lineItemNormalizer.test.ts new file mode 100644 index 0000000..cc73841 --- /dev/null +++ b/test/lineItemNormalizer.test.ts @@ -0,0 +1,99 @@ +import { describe, it, expect, vi } from "vitest"; +import { normalizeLineItems } from "../src/lineItemNormalizer.js"; +import { UnsupportedLineItemAssetError } from "../src/errors.js"; +import type { InvoiceLineItem } from "../src/types.js"; +import type { PriceOracle } from "../src/priceOracle.js"; + +const RATE_SCALE = 1_000_000_000_000_000_000n; + +function mockOracle(rates: Record): PriceOracle { + return { + getRate: vi.fn(async (fromAsset: string) => rates[fromAsset]), + }; +} + +describe("normalizeLineItems", () => { + it("passes same-asset items through without calling the oracle", async () => { + const items: InvoiceLineItem[] = [ + { description: "Service fee", quantity: 2, unitPrice: 5_000_000n, asset: "USDC" }, + ]; + const oracle = mockOracle({}); + + const result = await normalizeLineItems(items, "USDC", oracle); + + expect(oracle.getRate).not.toHaveBeenCalled(); + expect(result.total).toBe(10_000_000n); + expect(result.items[0]).toEqual({ + description: "Service fee", + originalAmount: 10_000_000n, + originalAsset: "USDC", + convertedAmount: 10_000_000n, + conversionRate: RATE_SCALE, + }); + }); + + it("calls the oracle once per unique cross-asset pair", async () => { + const items: InvoiceLineItem[] = [ + { description: "Network cost", quantity: 1, unitPrice: 10_000_000n, asset: "native" }, + { description: "More network cost", quantity: 1, unitPrice: 5_000_000n, asset: "native" }, + { description: "Platform credit", quantity: 1, unitPrice: 1_000_000n, asset: "CREDIT:GISSUER" }, + ]; + const oracle = mockOracle({ + native: 2n * RATE_SCALE, // 1 native = 2 USDC + "CREDIT:GISSUER": RATE_SCALE / 2n, // 1 credit = 0.5 USDC + }); + + const result = await normalizeLineItems(items, "USDC", oracle); + + expect(oracle.getRate).toHaveBeenCalledTimes(2); + expect(oracle.getRate).toHaveBeenCalledWith("native", "USDC"); + expect(oracle.getRate).toHaveBeenCalledWith("CREDIT:GISSUER", "USDC"); + expect(result.items[0]!.convertedAmount).toBe(20_000_000n); + expect(result.items[1]!.convertedAmount).toBe(10_000_000n); + expect(result.items[2]!.convertedAmount).toBe(500_000n); + expect(result.total).toBe(30_500_000n); + }); + + it("passes zero-amount items through without calling the oracle", async () => { + const items: InvoiceLineItem[] = [ + { description: "Discount", quantity: 1, unitPrice: 0n, asset: "native" }, + { description: "Waiver", quantity: 5, unitPrice: 0n, asset: "CREDIT:GISSUER" }, + ]; + const oracle = mockOracle({}); + + const result = await normalizeLineItems(items, "USDC", oracle); + + expect(oracle.getRate).not.toHaveBeenCalled(); + expect(result.total).toBe(0n); + expect(result.items.every((i) => i.convertedAmount === 0n)).toBe(true); + }); + + it("throws UnsupportedLineItemAssetError when no oracle price is available", async () => { + const items: InvoiceLineItem[] = [ + { description: "Mystery token", quantity: 1, unitPrice: 1_000_000n, asset: "MYSTERY:GISSUER" }, + ]; + const oracle = mockOracle({ "MYSTERY:GISSUER": undefined }); + + await expect(normalizeLineItems(items, "USDC", oracle)).rejects.toThrow( + UnsupportedLineItemAssetError + ); + }); + + it("uses the total override instead of quantity * unitPrice when provided", async () => { + const items: InvoiceLineItem[] = [ + { + description: "Bulk disccount", + quantity: 10, + unitPrice: 1_000_000n, + total: 8_000_000n, + asset: "USDC", + }, + ]; + const oracle = mockOracle({}); + + const result = await normalizeLineItems(items, "USDC", oracle); + + expect(result.items[0]!.originalAmount).toBe(8_000_000n); + expect(result.total).toBe(8_000_000n); + }); +}); From 4c90948a31c901f5270760918f19b510bf1f2382 Mon Sep 17 00:00:00 2001 From: Thomas Shelby Date: Wed, 29 Jul 2026 14:21:03 +0000 Subject: [PATCH 3/4] Add persistent contract invocation retry queue --- src/contractRetryQueue.ts | 129 ++++++++++++++++++++++++++++++++ src/errors.ts | 20 +++++ src/index.ts | 6 ++ src/persistentQueue.ts | 7 ++ src/types.ts | 24 ++++++ test/contractRetryQueue.test.ts | 115 ++++++++++++++++++++++++++++ 6 files changed, 301 insertions(+) create mode 100644 src/contractRetryQueue.ts create mode 100644 test/contractRetryQueue.test.ts diff --git a/src/contractRetryQueue.ts b/src/contractRetryQueue.ts new file mode 100644 index 0000000..ef754d7 --- /dev/null +++ b/src/contractRetryQueue.ts @@ -0,0 +1,129 @@ +/** + * Persistent retry queue for Soroban contract invocations. + * + * Wraps a caller-supplied executor (which builds the initial unsigned + * transaction and signs/submits a resource-assembled one) with re-simulation, + * exponential backoff, and durable tracking so transient ledger congestion or + * fee/resource mis-estimation doesn't surface as an immediate failure to the + * caller. + * + * Note: `RetryEngine` (retryEngine.ts) intentionally never retries + * contract-classified errors, so this queue implements its own backoff loop + * rather than reusing that engine. + */ + +import { rpc as SorobanRpc, type Transaction } from "@stellar/stellar-sdk"; +import type { ContractInvocation, ContractResult } from "./types.js"; +import { ContractRetryExhaustedError } from "./errors.js"; +import { PersistentTxQueue } from "./persistentQueue.js"; +import { TypedEventEmitter, type Unsubscribe } from "./events/TypedEventEmitter.js"; + +/** Builds and submits the transaction for a contract invocation. */ +export interface ContractInvocationExecutor { + /** Build the unsigned transaction for this invocation (fee/sequence handled by the caller). */ + buildTransaction(invocation: ContractInvocation): Promise; + /** Sign and submit a simulated-and-assembled transaction, returning the result once confirmed. */ + submit(tx: Transaction): Promise; +} + +export interface ContractRetryQueueConfig { + /** Initial backoff delay in milliseconds. Default: 500. */ + baseDelayMs?: number; + /** Maximum backoff delay in milliseconds. Default: 30 000. */ + maxDelayMs?: number; + /** Maximum number of attempts before giving up. Default: 5. */ + maxAttempts?: number; +} + +interface ContractRetryEvents { + contractRetryAttempted: { invocation: ContractInvocation; attempt: number; delay: number; error: unknown }; + contractRetryExhausted: { invocation: ContractInvocation; attempts: number; error: unknown }; +} + +const DEFAULT_CONFIG: Required = { + baseDelayMs: 500, + maxDelayMs: 30_000, + maxAttempts: 5, +}; + +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +export class ContractRetryQueue { + private readonly _config: Required; + private readonly _emitter = new TypedEventEmitter(); + private readonly _persistentQueue = new PersistentTxQueue(); + private _seq = 0; + + constructor( + private readonly server: SorobanRpc.Server, + private readonly executor: ContractInvocationExecutor, + config: ContractRetryQueueConfig = {} + ) { + this._config = { ...DEFAULT_CONFIG, ...config }; + } + + /** Subscribe to `contractRetryAttempted` / `contractRetryExhausted` events. */ + on( + event: K, + handler: (payload: ContractRetryEvents[K]) => void + ): Unsubscribe { + return this._emitter.on(event, handler); + } + + /** + * Enqueue a contract invocation. Resolves once the invocation succeeds, + * re-simulating and resubmitting with exponential backoff on failure. + * + * @throws {ContractRetryExhaustedError} After `maxAttempts` failed attempts. + */ + async enqueue(invocation: ContractInvocation): Promise { + const id = `contract-retry-${++this._seq}-${invocation.contractId}`; + await this._persistentQueue.enqueue({ id, payload: invocation, enqueuedAt: Date.now() }); + + try { + return await this._runWithRetry(invocation); + } finally { + await this._persistentQueue.remove(id); + } + } + + private async _runWithRetry(invocation: ContractInvocation): Promise { + let lastError: unknown; + + for (let attempt = 1; attempt <= this._config.maxAttempts; attempt++) { + try { + return await this._attempt(invocation); + } catch (error) { + lastError = error; + if (attempt >= this._config.maxAttempts) break; + + const delay = Math.min( + this._config.baseDelayMs * 2 ** (attempt - 1), + this._config.maxDelayMs + ); + this._emitter.emit("contractRetryAttempted", { invocation, attempt, delay, error }); + await sleep(delay); + } + } + + this._emitter.emit("contractRetryExhausted", { + invocation, + attempts: this._config.maxAttempts, + error: lastError, + }); + throw new ContractRetryExhaustedError(this._config.maxAttempts, lastError); + } + + private async _attempt(invocation: ContractInvocation): Promise { + const tx = await this.executor.buildTransaction(invocation); + const simResult = await this.server.simulateTransaction(tx); + if (SorobanRpc.Api.isSimulationError(simResult)) { + throw new Error(`Simulation failed: ${simResult.error}`); + } + + const prepared = SorobanRpc.assembleTransaction(tx, simResult).build(); + return this.executor.submit(prepared); + } +} diff --git a/src/errors.ts b/src/errors.ts index a196177..d680af0 100644 --- a/src/errors.ts +++ b/src/errors.ts @@ -708,6 +708,22 @@ export class Sep41AdapterError extends StellarSplitError { } } +/** Thrown when a queued contract invocation exhausts its retry attempts. */ +export class ContractRetryExhaustedError extends StellarSplitError { + readonly attempts: number; + + constructor(attempts: number, lastError: unknown) { + super( + `Contract invocation retry exhausted after ${attempts} attempts`, + "CONTRACT_RETRY_EXHAUSTED", + { attempts, lastError: lastError instanceof Error ? lastError.message : String(lastError) } + ); + this.name = "ContractRetryExhaustedError"; + this.attempts = attempts; + Object.setPrototypeOf(this, new.target.prototype); + } +} + /** Thrown when a line item's asset has no oracle price available for normalisation. */ export class UnsupportedLineItemAssetError extends StellarSplitError { readonly asset: string; @@ -1107,6 +1123,10 @@ export function isUnsupportedLineItemAssetError(err: unknown): err is Unsupporte return err instanceof UnsupportedLineItemAssetError; } +export function isContractRetryExhaustedError(err: unknown): err is ContractRetryExhaustedError { + return err instanceof ContractRetryExhaustedError; +} + export function isCloneChainTooDeepError(err: unknown): err is CloneChainTooDeepError { return err instanceof CloneChainTooDeepError; } diff --git a/src/index.ts b/src/index.ts index 308922b..2873ae1 100644 --- a/src/index.ts +++ b/src/index.ts @@ -591,6 +591,12 @@ export type { } from "./types.js"; export { UnsupportedLineItemAssetError, isUnsupportedLineItemAssetError } from "./errors.js"; +// Contract invocation retry queue +export { ContractRetryQueue } from "./contractRetryQueue.js"; +export type { ContractInvocationExecutor, ContractRetryQueueConfig } from "./contractRetryQueue.js"; +export type { ContractInvocation, ContractResult } from "./types.js"; +export { ContractRetryExhaustedError, isContractRetryExhaustedError } from "./errors.js"; + // Merkle proof functionality export { generateMerkleProof, verifyMerkleProof } from "./merkle.js"; export type { MerkleProof } from "./merkle.js"; diff --git a/src/persistentQueue.ts b/src/persistentQueue.ts index 6c70494..78aa13c 100644 --- a/src/persistentQueue.ts +++ b/src/persistentQueue.ts @@ -111,6 +111,13 @@ export class PersistentTxQueue { return this.memory.slice(); } + /** Remove a single item from the queue by id, without processing it. */ + async remove(id: string): Promise { + this.memory = this.memory.filter((i) => i.id !== id); + const db = await this.getDb(); + if (db) await idbDelete(db, id); + } + /** Clear all items from memory and IndexedDB. */ async clear(): Promise { this.memory = []; diff --git a/src/types.ts b/src/types.ts index a7ee658..1386054 100644 --- a/src/types.ts +++ b/src/types.ts @@ -1452,6 +1452,30 @@ export interface NormalizedInvoiceTotal { items: NormalizedLineItem[]; } +// --------------------------------------------------------------------------- +// Contract Retry Queue Types +// --------------------------------------------------------------------------- + +/** A single Soroban contract invocation to be submitted (with retry on failure). */ +export interface ContractInvocation { + /** Contract address being invoked. */ + contractId: string; + /** Contract method name. */ + method: string; + /** Method arguments, in the order the contract expects them. */ + args: unknown[]; + /** Stellar address the invocation is submitted on behalf of. */ + source: string; +} + +/** Result of a successfully submitted contract invocation. */ +export interface ContractResult { + /** Hash of the submitted transaction. */ + txHash: string; + /** Decoded return value from the contract call, if any. */ + returnValue?: unknown; +} + /** Configuration for IPFS backend. */ export interface IPFSConfig { /** Backend type: 'gateway' for HTTP gateway or 'kubo' for Kubo RPC API. */ diff --git a/test/contractRetryQueue.test.ts b/test/contractRetryQueue.test.ts new file mode 100644 index 0000000..49039a0 --- /dev/null +++ b/test/contractRetryQueue.test.ts @@ -0,0 +1,115 @@ +import { describe, it, expect, vi, beforeEach } from "vitest"; +import { IDBFactory } from "fake-indexeddb"; + +const { assembleTransactionMock, isSimulationErrorMock } = vi.hoisted(() => ({ + assembleTransactionMock: vi.fn(), + isSimulationErrorMock: vi.fn(), +})); + +vi.mock("@stellar/stellar-sdk", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + rpc: { + ...actual.rpc, + assembleTransaction: assembleTransactionMock, + Api: { + ...actual.rpc.Api, + isSimulationError: isSimulationErrorMock, + }, + }, + }; +}); + +const { ContractRetryQueue } = await import("../src/contractRetryQueue.js"); +const { ContractRetryExhaustedError } = await import("../src/errors.js"); +import type { ContractInvocation } from "../src/types.js"; +import type { ContractInvocationExecutor } from "../src/contractRetryQueue.js"; + +const invocation: ContractInvocation = { + contractId: "CCONTRACT", + method: "release", + args: [], + source: "GSOURCE", +}; + +function makeExecutor(submit: ContractInvocationExecutor["submit"]): ContractInvocationExecutor { + return { + buildTransaction: vi.fn().mockResolvedValue({ tx: true }), + submit, + }; +} + +beforeEach(() => { + (globalThis as Record).indexedDB = new IDBFactory(); + assembleTransactionMock.mockReset(); + isSimulationErrorMock.mockReset(); + isSimulationErrorMock.mockReturnValue(false); + assembleTransactionMock.mockImplementation(() => ({ build: () => ({ prepared: true }) })); +}); + +describe("ContractRetryQueue", () => { + it("resolves on the first successful attempt without emitting retry events", async () => { + const server = { simulateTransaction: vi.fn().mockResolvedValue({ result: {} }) } as any; + const submit = vi.fn().mockResolvedValue({ txHash: "tx1" }); + const executor = makeExecutor(submit); + const queue = new ContractRetryQueue(server, executor, { baseDelayMs: 5, maxDelayMs: 20, maxAttempts: 3 }); + + const attempted = vi.fn(); + queue.on("contractRetryAttempted", attempted); + + const result = await queue.enqueue(invocation); + + expect(result).toEqual({ txHash: "tx1" }); + expect(server.simulateTransaction).toHaveBeenCalledTimes(1); + expect(submit).toHaveBeenCalledTimes(1); + expect(attempted).not.toHaveBeenCalled(); + }); + + it("re-simulates and resubmits on failure, succeeding within maxAttempts", async () => { + const server = { simulateTransaction: vi.fn().mockResolvedValue({ result: {} }) } as any; + const submit = vi + .fn() + .mockRejectedValueOnce(new Error("resource limits exceeded")) + .mockResolvedValueOnce({ txHash: "tx2" }); + const executor = makeExecutor(submit); + const queue = new ContractRetryQueue(server, executor, { baseDelayMs: 5, maxDelayMs: 20, maxAttempts: 3 }); + + const attempted: unknown[] = []; + queue.on("contractRetryAttempted", (payload) => attempted.push(payload)); + + const result = await queue.enqueue(invocation); + + expect(result).toEqual({ txHash: "tx2" }); + expect(server.simulateTransaction).toHaveBeenCalledTimes(2); + expect(submit).toHaveBeenCalledTimes(2); + expect(attempted).toHaveLength(1); + expect(attempted[0]).toMatchObject({ attempt: 1, delay: 5 }); + }); + + it("throws ContractRetryExhaustedError and emits contractRetryExhausted after maxAttempts", async () => { + const server = { simulateTransaction: vi.fn().mockResolvedValue({ result: {} }) } as any; + const submit = vi.fn().mockRejectedValue(new Error("always fails")); + const executor = makeExecutor(submit); + const queue = new ContractRetryQueue(server, executor, { baseDelayMs: 5, maxDelayMs: 20, maxAttempts: 3 }); + + const exhausted = vi.fn(); + queue.on("contractRetryExhausted", exhausted); + + await expect(queue.enqueue(invocation)).rejects.toThrow(ContractRetryExhaustedError); + expect(submit).toHaveBeenCalledTimes(3); + expect(exhausted).toHaveBeenCalledTimes(1); + expect(exhausted.mock.calls[0]![0]).toMatchObject({ attempts: 3 }); + }); + + it("throws when simulation fails", async () => { + const server = { simulateTransaction: vi.fn().mockResolvedValue({ error: "sim error" }) } as any; + isSimulationErrorMock.mockReturnValue(true); + const submit = vi.fn(); + const executor = makeExecutor(submit); + const queue = new ContractRetryQueue(server, executor, { baseDelayMs: 5, maxDelayMs: 20, maxAttempts: 1 }); + + await expect(queue.enqueue(invocation)).rejects.toThrow(ContractRetryExhaustedError); + expect(submit).not.toHaveBeenCalled(); + }); +}); From 5c8adc2eb6cec906f6bd88721fd93d91ba2cc369 Mon Sep 17 00:00:00 2001 From: Thomas Shelby Date: Wed, 29 Jul 2026 14:22:27 +0000 Subject: [PATCH 4/4] Add invoice batch processor with concurrency limiter --- src/index.ts | 8 ++ src/invoiceBatchProcessor.ts | 137 +++++++++++++++++++++++++++++ test/invoiceBatchProcessor.test.ts | 134 ++++++++++++++++++++++++++++ 3 files changed, 279 insertions(+) create mode 100644 src/invoiceBatchProcessor.ts create mode 100644 test/invoiceBatchProcessor.test.ts diff --git a/src/index.ts b/src/index.ts index 2873ae1..b20ae15 100644 --- a/src/index.ts +++ b/src/index.ts @@ -597,6 +597,14 @@ export type { ContractInvocationExecutor, ContractRetryQueueConfig } from "./con export type { ContractInvocation, ContractResult } from "./types.js"; export { ContractRetryExhaustedError, isContractRetryExhaustedError } from "./errors.js"; +// Invoice batch processor with concurrency limiter +export { InvoiceBatchProcessor } from "./invoiceBatchProcessor.js"; +export type { + BatchInvoiceResult, + InvoiceBatchConfig, + InvoicePaymentSubmitter, +} from "./invoiceBatchProcessor.js"; + // Merkle proof functionality export { generateMerkleProof, verifyMerkleProof } from "./merkle.js"; export type { MerkleProof } from "./merkle.js"; diff --git a/src/invoiceBatchProcessor.ts b/src/invoiceBatchProcessor.ts new file mode 100644 index 0000000..65f1162 --- /dev/null +++ b/src/invoiceBatchProcessor.ts @@ -0,0 +1,137 @@ +/** + * Concurrency-limited batch invoice payment processor. + * + * Submits a set of invoice payments with a configurable concurrency ceiling + * so month-end batch billing doesn't overwhelm Horizon's rate limits, and + * reports each result as it settles rather than waiting for the whole batch. + */ + +import { TypedEventEmitter } from "./events/TypedEventEmitter.js"; + +/** Result of submitting a single invoice's payment within a batch. */ +export interface BatchInvoiceResult { + invoiceId: string; + status: "success" | "failed"; + txHash?: string; + error?: string; +} + +/** Configuration for a batch run. */ +export interface InvoiceBatchConfig { + /** Stellar address submitting each payment. */ + payer: string; + /** Amount (in stroops) to pay per invoice ID. */ + amounts: Record; + /** Maximum number of submissions in flight at once. Default: 3. */ + maxConcurrent?: number; + /** Fallback pause duration (ms) on a rate-limit error when no explicit `retryAfterMs` is available. Default: 2000. */ + rateLimitPauseMs?: number; +} + +interface InvoiceBatchEvents { + batchInvoiceSettled: BatchInvoiceResult; + batchInvoiceFailed: BatchInvoiceResult; +} + +/** Minimal client surface required to submit a batched invoice payment. */ +export interface InvoicePaymentSubmitter { + submitPayment(params: { + invoiceId: string; + payer: string; + amount: bigint; + }): Promise<{ txHash: string }>; +} + +const DEFAULT_MAX_CONCURRENT = 3; +const DEFAULT_RATE_LIMIT_PAUSE_MS = 2_000; + +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +function isRateLimitError(error: unknown): boolean { + return error instanceof Error && /429|rate.?limit|too many requests/i.test(error.message); +} + +function retryAfterMs(error: unknown, fallbackMs: number): number { + const withRetryAfter = error as { retryAfterMs?: number } | undefined; + return typeof withRetryAfter?.retryAfterMs === "number" ? withRetryAfter.retryAfterMs : fallbackMs; +} + +export class InvoiceBatchProcessor { + readonly events = new TypedEventEmitter(); + + constructor(private readonly client: InvoicePaymentSubmitter) {} + + /** + * Submit payments for `invoiceIds`, yielding a {@link BatchInvoiceResult} as + * each one settles. No more than `config.maxConcurrent` submissions are + * in-flight at once. On a 429 from any submission, all further dispatch + * pauses until the rate limit's backoff window elapses. + */ + async *process( + invoiceIds: string[], + config: InvoiceBatchConfig + ): AsyncIterableIterator { + const maxConcurrent = config.maxConcurrent ?? DEFAULT_MAX_CONCURRENT; + const rateLimitPauseMs = config.rateLimitPauseMs ?? DEFAULT_RATE_LIMIT_PAUSE_MS; + + let cursor = 0; + let pausedUntil = 0; + let slotSeq = 0; + const inFlight = new Map>(); + + const runOne = async (invoiceId: string): Promise => { + const wait = pausedUntil - Date.now(); + if (wait > 0) await sleep(wait); + + const amount = config.amounts[invoiceId]; + if (amount === undefined) { + const result: BatchInvoiceResult = { + invoiceId, + status: "failed", + error: `No amount specified for invoice ${invoiceId}`, + }; + this.events.emit("batchInvoiceFailed", result); + return result; + } + + try { + const { txHash } = await this.client.submitPayment({ invoiceId, payer: config.payer, amount }); + const result: BatchInvoiceResult = { invoiceId, status: "success", txHash }; + this.events.emit("batchInvoiceSettled", result); + return result; + } catch (error) { + if (isRateLimitError(error)) { + pausedUntil = Date.now() + retryAfterMs(error, rateLimitPauseMs); + } + const result: BatchInvoiceResult = { + invoiceId, + status: "failed", + error: error instanceof Error ? error.message : String(error), + }; + this.events.emit("batchInvoiceFailed", result); + return result; + } + }; + + const launch = (): void => { + if (cursor >= invoiceIds.length) return; + const invoiceId = invoiceIds[cursor++]!; + const slot = slotSeq++; + inFlight.set( + slot, + runOne(invoiceId).then((result) => ({ slot, result })) + ); + }; + + for (let i = 0; i < maxConcurrent; i++) launch(); + + while (inFlight.size > 0) { + const { slot, result } = await Promise.race(inFlight.values()); + inFlight.delete(slot); + launch(); + yield result; + } + } +} diff --git a/test/invoiceBatchProcessor.test.ts b/test/invoiceBatchProcessor.test.ts new file mode 100644 index 0000000..a85f5b1 --- /dev/null +++ b/test/invoiceBatchProcessor.test.ts @@ -0,0 +1,134 @@ +import { describe, it, expect, vi } from "vitest"; +import { InvoiceBatchProcessor } from "../src/invoiceBatchProcessor.js"; +import type { InvoicePaymentSubmitter } from "../src/invoiceBatchProcessor.js"; + +async function drain(iter: AsyncIterableIterator): Promise { + const results: T[] = []; + for await (const item of iter) results.push(item); + return results; +} + +describe("InvoiceBatchProcessor", () => { + it("submits every invoice and reports success results", async () => { + const submitPayment = vi.fn().mockImplementation(async ({ invoiceId }: { invoiceId: string }) => ({ + txHash: `tx-${invoiceId}`, + })); + const processor = new InvoiceBatchProcessor({ submitPayment } as InvoicePaymentSubmitter); + + const results = await drain( + processor.process(["inv1", "inv2", "inv3"], { + payer: "GPAYER", + amounts: { inv1: 1n, inv2: 2n, inv3: 3n }, + }) + ); + + expect(results).toHaveLength(3); + expect(new Set(results.map((r) => r.invoiceId))).toEqual(new Set(["inv1", "inv2", "inv3"])); + expect(results.every((r) => r.status === "success")).toBe(true); + expect(submitPayment).toHaveBeenCalledTimes(3); + }); + + it("never runs more than maxConcurrent submissions at once", async () => { + let active = 0; + let maxActive = 0; + const submitPayment = vi.fn().mockImplementation(async () => { + active++; + maxActive = Math.max(maxActive, active); + await new Promise((r) => setTimeout(r, 10)); + active--; + return { txHash: "tx" }; + }); + const processor = new InvoiceBatchProcessor({ submitPayment } as InvoicePaymentSubmitter); + + await drain( + processor.process(["a", "b", "c", "d", "e"], { + payer: "GPAYER", + amounts: { a: 1n, b: 1n, c: 1n, d: 1n, e: 1n }, + maxConcurrent: 2, + }) + ); + + expect(maxActive).toBeLessThanOrEqual(2); + }); + + it("reports failed invoices individually without aborting the rest of the batch", async () => { + const submitPayment = vi.fn().mockImplementation(async ({ invoiceId }: { invoiceId: string }) => { + if (invoiceId === "bad") throw new Error("insufficient funds"); + return { txHash: `tx-${invoiceId}` }; + }); + const processor = new InvoiceBatchProcessor({ submitPayment } as InvoicePaymentSubmitter); + + const results = await drain( + processor.process(["good1", "bad", "good2"], { + payer: "GPAYER", + amounts: { good1: 1n, bad: 1n, good2: 1n }, + }) + ); + + expect(results).toHaveLength(3); + const bad = results.find((r) => r.invoiceId === "bad")!; + expect(bad.status).toBe("failed"); + expect(bad.error).toContain("insufficient funds"); + expect(results.filter((r) => r.status === "success")).toHaveLength(2); + }); + + it("pauses in-flight dispatch on a 429 and resumes after the backoff window", async () => { + let calls = 0; + const submitPayment = vi.fn().mockImplementation(async ({ invoiceId }: { invoiceId: string }) => { + calls++; + if (invoiceId === "throttled") { + const err = new Error("429 too many requests") as Error & { retryAfterMs?: number }; + err.retryAfterMs = 20; + throw err; + } + return { txHash: `tx-${invoiceId}` }; + }); + const processor = new InvoiceBatchProcessor({ submitPayment } as InvoicePaymentSubmitter); + + const start = Date.now(); + const results = await drain( + processor.process(["throttled", "after1", "after2"], { + payer: "GPAYER", + amounts: { throttled: 1n, after1: 1n, after2: 1n }, + maxConcurrent: 1, + }) + ); + const elapsed = Date.now() - start; + + expect(calls).toBe(3); + expect(results.find((r) => r.invoiceId === "throttled")!.status).toBe("failed"); + expect(results.filter((r) => r.status === "success")).toHaveLength(2); + expect(elapsed).toBeGreaterThanOrEqual(20); + }); + + it("emits batchInvoiceSettled and batchInvoiceFailed events", async () => { + const submitPayment = vi.fn().mockImplementation(async ({ invoiceId }: { invoiceId: string }) => { + if (invoiceId === "bad") throw new Error("failed"); + return { txHash: "tx" }; + }); + const processor = new InvoiceBatchProcessor({ submitPayment } as InvoicePaymentSubmitter); + + const settled = vi.fn(); + const failed = vi.fn(); + processor.events.on("batchInvoiceSettled", settled); + processor.events.on("batchInvoiceFailed", failed); + + await drain(processor.process(["good", "bad"], { payer: "GPAYER", amounts: { good: 1n, bad: 1n } })); + + expect(settled).toHaveBeenCalledTimes(1); + expect(failed).toHaveBeenCalledTimes(1); + }); + + it("fails an invoice with no configured amount without calling submitPayment", async () => { + const submitPayment = vi.fn().mockResolvedValue({ txHash: "tx" }); + const processor = new InvoiceBatchProcessor({ submitPayment } as InvoicePaymentSubmitter); + + const results = await drain( + processor.process(["missing"], { payer: "GPAYER", amounts: {} }) + ); + + expect(results[0]!.status).toBe("failed"); + expect(submitPayment).not.toHaveBeenCalled(); + }); +}); +