diff --git a/src/client.ts b/src/client.ts index 75c9e2b..586a2e0 100644 --- a/src/client.ts +++ b/src/client.ts @@ -233,6 +233,7 @@ import { PriorityQueue } from "./priorityQueue.js"; import type { RequestPriority } from "./priorityQueue.js"; import { IdempotencyManager } from "./idempotency.js"; import type { IdempotencyConfig } from "./idempotency.js"; +import { RollbackCoordinator } from "./splitRollbackCoordinator.js"; import { validateInvoicePayload } from "./payloadGuard.js"; import { validateSplitRatiosOrThrow } from "./validators/splitRatioValidator.js"; import type { SplitConfig } from "./types.js"; @@ -489,6 +490,12 @@ export interface StellarSplitClientConfig { * for inspecting in-flight transaction envelopes. Defaults to false. */ debug?: boolean; + /** + * Optional fiat-to-asset price oracle adapter (see `PriceOracleAdapter` in + * types.ts). Used by `convertFiatToAsset` in currencyConverter.ts to + * resolve display conversions. Defaults to no oracle configured. + */ + priceOracle?: import("./types.js").PriceOracleAdapter; } /** Network configuration. */ @@ -622,6 +629,7 @@ export class StellarSplitClient extends TypedEventEmitter { private _retryOptions: RetryOptions | null = null; private _horizonReader: HorizonFallbackReader | null = null; private _idempotency: IdempotencyManager | null = null; + private _rollbackCoordinator: RollbackCoordinator | null = null; private _pool: ConnectionPool | null = null; /** * Effective pool size chosen at construction (or 0 when pooling is off). @@ -2119,6 +2127,19 @@ export class StellarSplitClient extends TypedEventEmitter { const result = await this._submitWaterfallTx(params.payer, operations); this._cache?.invalidate(params.invoiceId); + + // Record a rollback checkpoint for the submitted legs. The on-chain + // submission is atomic (all-or-nothing), so every funded step succeeded + // together; downstream app-layer failures (e.g. webhook delivery) are + // reconciled by callers via RollbackCoordinator.markLegFailed. + const coordinator = this.getRollbackCoordinator(); + coordinator.begin( + result.txHash, + params.invoiceId, + fundedSteps.map((step) => ({ recipient: step.recipient, amount: step.amount })), + ); + fundedSteps.forEach((_, index) => coordinator.markLegSuccess(result.txHash, index)); + return { txHash: result.txHash }; } @@ -2390,6 +2411,22 @@ export class StellarSplitClient extends TypedEventEmitter { return this._advancedCircuitBreaker; } + /** + * The rollback coordinator tracking split-payment leg checkpoints created + * by `submitPayment`'s waterfall path. Lazily instantiated on first use. + */ + getRollbackCoordinator(): RollbackCoordinator { + if (!this._rollbackCoordinator) { + this._rollbackCoordinator = new RollbackCoordinator(this._idempotency ?? undefined); + } + return this._rollbackCoordinator; + } + + /** The configured fiat-to-asset price oracle adapter, or null if none was provided. */ + get priceOracle(): import("./types.js").PriceOracleAdapter | null { + return this.config.priceOracle ?? null; + } + /** * The optimistic UI cache, or null when `optimisticCache` was not passed * to the constructor. Use `client.optimisticCache?.onRollback(...)` to diff --git a/src/currencyConverter.ts b/src/currencyConverter.ts index 1f9acfe..a646dae 100644 --- a/src/currencyConverter.ts +++ b/src/currencyConverter.ts @@ -115,4 +115,43 @@ export async function convertAmount( fromToken, toDisplayCurrency, }; +} + +/** Result of converting a fiat-denominated invoice amount to an on-chain asset. */ +export interface FiatConversion { + /** The requested fiat amount. */ + fiatAmount: number; + /** The fiat currency code the amount was denominated in (e.g. "USD"). */ + fiatCurrency: string; + /** The equivalent amount in the target asset. */ + assetAmount: number; + /** The asset code the amount was converted to (e.g. "XLM"). */ + assetCode: string; + /** The base/quote rate used for the conversion (units of `fiatCurrency` per 1 `assetCode`). */ + rate: number; +} + +/** + * Convert a fiat-denominated amount into its equivalent in `assetCode` using + * a pluggable {@link PriceOracleAdapter} (see priceOracle.ts for the default + * CoinGecko-backed implementation). Display-only, like the rest of this module. + */ +export async function convertFiatToAsset( + fiatAmount: number, + fiatCurrency: string, + assetCode: string, + oracle: PriceOracleAdapter, +): Promise { + const rate = await oracle.getPrice(assetCode, fiatCurrency); + if (!(rate > 0)) { + throw new OraclePriceError(`Invalid oracle rate for ${assetCode}/${fiatCurrency}: ${rate}`); + } + + return { + fiatAmount, + fiatCurrency, + assetAmount: fiatAmount / rate, + assetCode, + rate, + }; } \ No newline at end of file diff --git a/src/index.ts b/src/index.ts index 309a85c..276c565 100644 --- a/src/index.ts +++ b/src/index.ts @@ -336,6 +336,13 @@ export type { SequenceCacheConfig } from "./sequenceCache.js"; export { PathRouter } from "./pathRouter.js"; export type { PathResult, PathHop, PathRequest, PathRouterConfig } from "./pathRouter.js"; +export { PathQueryBuilder } from "./pathQueryBuilder.js"; +export type { + StrictSendQueryParams, + StrictReceiveQueryParams, + PathQueryBuilderConfig, +} from "./pathQueryBuilder.js"; +export type { PathQuery, PathQueryResult, StrictSendPathQuery, StrictReceivePathQuery } from "./types.js"; export { OfferTracker } from "./offerTracker.js"; export type { OfferTrackerConfig, OfferTrackerEventMap } from "./offerTracker.js"; @@ -778,6 +785,24 @@ export type { TranchedInvoice, TrancheStatus, } from "./trancheProgress.js"; + +// Invoice payment progress tracking +export { PaymentProgressTracker } from "./paymentProgressTracker.js"; +export type { + PaymentProgressEventMap, + PaymentProgressTrackerOptions, +} from "./paymentProgressTracker.js"; +export type { InvoicePaymentProgress, RecipientPaymentState } from "./types.js"; + +// Fiat-to-asset price oracle adapter +export { CoinGeckoPriceOracle } from "./priceOracle.js"; +export type { CoinGeckoPriceOracleOptions } from "./priceOracle.js"; +export { RateCache } from "./rateCache.js"; +export type { RateCacheConfig } from "./rateCache.js"; +export type { PriceOracleAdapter } from "./types.js"; +export { convertFiatToAsset } from "./currencyConverter.js"; +export type { FiatConversion, ConvertedAmount } from "./currencyConverter.js"; + export { Sep41Adapter, createSep41Adapter } from "./sep41Adapter.js"; export type { Sep41TokenCapabilities } from "./sep41Adapter.js"; @@ -900,6 +925,11 @@ export type { export { IdempotencyManager } from "./idempotency.js"; export type { IdempotencyConfig } from "./idempotency.js"; +export { RollbackCoordinator } from "./splitRollbackCoordinator.js"; +export type { SplitRollbackEventMap } from "./splitRollbackCoordinator.js"; +export type { SplitRollbackRecord } from "./snapshot.js"; +export type { SplitLeg, SplitLegState, SplitResult, SplitRollbackCheckpoint } from "./types.js"; + export { validateInvoicePayload, PayloadSizeError, diff --git a/src/pathQueryBuilder.ts b/src/pathQueryBuilder.ts new file mode 100644 index 0000000..f3c310d --- /dev/null +++ b/src/pathQueryBuilder.ts @@ -0,0 +1,171 @@ +/** + * Builds and executes Horizon strict-send / strict-receive path queries. + * + * Encapsulates query assembly and validation so callers (and tests) can work + * with path queries independently of any particular routing strategy. + * {@link PathRouter} (pathRouter.ts) uses this builder internally. + */ + +import { Asset, Horizon } from "@stellar/stellar-sdk"; +import { SimpleCache } from "./cache.js"; +import { InvalidPathQueryError } from "./errors.js"; +import { isValidStellarAddress } from "./utils.js"; +import type { PathQuery, PathQueryResult } from "./types.js"; + +/** Parameters for {@link PathQueryBuilder.forStrictSend}. */ +export interface StrictSendQueryParams { + /** Asset the sender will supply. */ + sourceAsset: Asset; + /** Amount the sender will supply, in the source asset's base unit. */ + sourceAmount: bigint; + /** A specific destination account to query reachable assets for. */ + destinationAccount?: string; + /** A fixed list of acceptable destination assets. */ + destinationAssets?: Asset[]; +} + +/** Parameters for {@link PathQueryBuilder.forStrictReceive}. */ +export interface StrictReceiveQueryParams { + /** Asset the recipient should receive. */ + destinationAsset: Asset; + /** Amount the recipient should receive, in the destination asset's base unit. */ + destinationAmount: bigint; + /** A specific source account to query spendable assets for. */ + sourceAccount?: string; + /** A fixed list of acceptable source assets. */ + sourceAssets?: Asset[]; +} + +/** Configuration for {@link PathQueryBuilder}. */ +export interface PathQueryBuilderConfig { + /** Cache TTL in milliseconds. Default: 10_000 (10s). */ + ttlMs?: number; + /** Maximum number of cached query results. Default: 5_000. */ + maxEntries?: number; +} + +/** + * Assembles validated `PathQuery` objects and executes them against Horizon, + * caching results by query parameters within a configurable TTL. + */ +export class PathQueryBuilder { + private readonly server: Horizon.Server; + private readonly cache: SimpleCache; + + constructor(server: Horizon.Server, config: PathQueryBuilderConfig = {}) { + this.server = server; + this.cache = new SimpleCache({ + enabled: true, + ttlMs: config.ttlMs ?? 10_000, + maxEntries: config.maxEntries ?? 5_000, + }); + } + + /** + * Build a strict-send query: `sourceAmount` is fixed, the destination + * amount is estimated. Requires either `destinationAccount` or a non-empty + * `destinationAssets` list. + */ + forStrictSend(params: StrictSendQueryParams): PathQuery { + if (!(params.sourceAmount > 0n)) { + throw new InvalidPathQueryError("sourceAmount must be greater than 0", { + sourceAmount: params.sourceAmount.toString(), + }); + } + if (params.destinationAccount === undefined && (!params.destinationAssets || params.destinationAssets.length === 0)) { + throw new InvalidPathQueryError("Either destinationAccount or destinationAssets is required"); + } + if (params.destinationAccount !== undefined && !isValidStellarAddress(params.destinationAccount)) { + throw new InvalidPathQueryError(`Invalid destination account: ${params.destinationAccount}`, { + destinationAccount: params.destinationAccount, + }); + } + + return { + kind: "strictSend", + sourceAsset: params.sourceAsset, + sourceAmount: params.sourceAmount, + destination: params.destinationAccount ?? params.destinationAssets!, + }; + } + + /** + * Build a strict-receive query: `destinationAmount` is fixed, the source + * amount is estimated. Requires either `sourceAccount` or a non-empty + * `sourceAssets` list. + */ + forStrictReceive(params: StrictReceiveQueryParams): PathQuery { + if (!(params.destinationAmount > 0n)) { + throw new InvalidPathQueryError("destinationAmount must be greater than 0", { + destinationAmount: params.destinationAmount.toString(), + }); + } + if (params.sourceAccount === undefined && (!params.sourceAssets || params.sourceAssets.length === 0)) { + throw new InvalidPathQueryError("Either sourceAccount or sourceAssets is required"); + } + if (params.sourceAccount !== undefined && !isValidStellarAddress(params.sourceAccount)) { + throw new InvalidPathQueryError(`Invalid source account: ${params.sourceAccount}`, { + sourceAccount: params.sourceAccount, + }); + } + + return { + kind: "strictReceive", + source: params.sourceAccount ?? params.sourceAssets!, + destinationAsset: params.destinationAsset, + destinationAmount: params.destinationAmount, + }; + } + + /** + * Submit `query` to Horizon and return every matching path, sorted by + * best rate (highest destination amount for strict-send, lowest source + * amount for strict-receive). Cached by query parameters within the TTL. + */ + async execute(query: PathQuery): Promise { + const cacheKey = this.cacheKey(query); + const cached = this.cache.get(cacheKey); + if (cached) return cached; + + const records = + query.kind === "strictSend" + ? await this.server.strictSendPaths(query.sourceAsset, query.sourceAmount.toString(), query.destination).call() + : await this.server.strictReceivePaths(query.source, query.destinationAsset, query.destinationAmount.toString()).call(); + + const results: PathQueryResult[] = records.records.map((record) => ({ + path: record.path, + sourceAmount: BigInt(record.source_amount), + destinationAmount: BigInt(record.destination_amount), + })); + + results.sort((a, b) => { + if (query.kind === "strictSend") { + return a.destinationAmount === b.destinationAmount ? 0 : a.destinationAmount > b.destinationAmount ? -1 : 1; + } + return a.sourceAmount === b.sourceAmount ? 0 : a.sourceAmount < b.sourceAmount ? -1 : 1; + }); + + this.cache.set(cacheKey, results); + return results; + } + + /** Clear all cached query results. */ + clearCache(): void { + this.cache.clear(); + } + + private cacheKey(query: PathQuery): string { + if (query.kind === "strictSend") { + const dest = Array.isArray(query.destination) + ? query.destination.map(assetKey).join(",") + : query.destination; + return `strictSend:${assetKey(query.sourceAsset)}:${query.sourceAmount}:${dest}`; + } + const src = Array.isArray(query.source) ? query.source.map(assetKey).join(",") : query.source; + return `strictReceive:${src}:${assetKey(query.destinationAsset)}:${query.destinationAmount}`; + } +} + +function assetKey(asset: Asset): string { + return asset.isNative() ? "native" : `${asset.getCode()}:${asset.getIssuer()}`; +} diff --git a/src/pathRouter.ts b/src/pathRouter.ts index 2bdbc8a..dfea58f 100644 --- a/src/pathRouter.ts +++ b/src/pathRouter.ts @@ -5,13 +5,14 @@ * Horizon's path-payment endpoints to find the best conversion route and * returns the optimal path for each split leg. * - * Integrates with {@link SimpleCache} to avoid redundant Horizon calls for - * identical source/destination pairs within the cache TTL window. + * Delegates query assembly, validation, and caching to {@link PathQueryBuilder} + * to avoid redundant Horizon calls for identical source/destination pairs + * within the cache TTL window. */ import { Asset, Horizon, Operation } from "@stellar/stellar-sdk"; -import { SimpleCache } from "./cache.js"; import { PathNotFoundError, PathRouterError } from "./errors.js"; +import { PathQueryBuilder } from "./pathQueryBuilder.js"; // --------------------------------------------------------------------------- // Types @@ -70,7 +71,7 @@ export interface PathRouterConfig { */ export class PathRouter { private readonly server: Horizon.Server; - private readonly cache: SimpleCache; + private readonly queryBuilder: PathQueryBuilder; /** * @param horizonUrl - Horizon server URL. @@ -78,8 +79,7 @@ export class PathRouter { */ constructor(horizonUrl: string, config: PathRouterConfig = {}) { this.server = new Horizon.Server(horizonUrl); - this.cache = new SimpleCache({ - enabled: true, + this.queryBuilder = new PathQueryBuilder(this.server, { ttlMs: config.ttlMs ?? 15_000, maxEntries: config.maxEntries ?? 5_000, }); @@ -97,45 +97,24 @@ export class PathRouter { * the destination amount is estimated. */ async findStrictSendPath(req: PathRequest): Promise { - const cacheKey = this.cacheKey("send", req); - const cached = this.cache.get(cacheKey); - if (cached) return cached; + const sourceAssetType = assetType(req.sourceAsset); + const destAssetType = assetType(req.destinationAsset); try { - const sourceAssetType = req.sourceAsset.isNative() - ? "native" - : `${req.sourceAsset.getCode()}:${req.sourceAsset.getIssuer()}`; - const destAssetType = req.destinationAsset.isNative() - ? "native" - : `${req.destinationAsset.getCode()}:${req.destinationAsset.getIssuer()}`; - - const records = await this.server - .strictSendPaths( - req.sourceAsset, - req.sourceAmount.toString(), - [req.destinationAsset], - ) - .call(); - - if (records.records.length === 0) { - throw new PathNotFoundError( - sourceAssetType, - destAssetType, - req.sourceAmount, - ); + const query = this.queryBuilder.forStrictSend({ + sourceAsset: req.sourceAsset, + sourceAmount: req.sourceAmount, + destinationAssets: [req.destinationAsset], + }); + const results = await this.queryBuilder.execute(query); + + if (results.length === 0) { + throw new PathNotFoundError(sourceAssetType, destAssetType, req.sourceAmount); } - // First record is the best path (highest destination amount) - const best = records.records[0]!; - - const result: PathResult = { - path: best.path, - destinationAmount: BigInt(best.destination_amount), - sourceAmount: BigInt(best.source_amount), - }; - - this.cache.set(cacheKey, result); - return result; + // Results are sorted best-first (highest destination amount). + const best = results[0]!; + return { path: best.path, destinationAmount: best.destinationAmount, sourceAmount: best.sourceAmount }; } catch (err) { if (err instanceof PathNotFoundError) throw err; throw new PathRouterError( @@ -156,51 +135,24 @@ export class PathRouter { destAmount: bigint, destinationAsset: Asset, ): Promise { - const cacheKey = this.cacheKeyReceive(sourceAsset, destAmount, destinationAsset); - const cached = this.cache.get(cacheKey); - if (cached) return cached; + const sourceAssetType = assetType(sourceAsset); + const destAssetType = assetType(destinationAsset); try { - const sourceAssetType = sourceAsset.isNative() - ? "native" - : `${sourceAsset.getCode()}:${sourceAsset.getIssuer()}`; - const destAssetType = destinationAsset.isNative() - ? "native" - : `${destinationAsset.getCode()}:${destinationAsset.getIssuer()}`; - - const srcStr = sourceAsset.isNative() - ? "native" - : `${sourceAsset.getCode()}:${sourceAsset.getIssuer()}`; - const dstStr = destinationAsset.isNative() - ? "native" - : `${destinationAsset.getCode()}:${destinationAsset.getIssuer()}`; - - const records = await this.server - .strictReceivePaths( - srcStr, - destinationAsset, - destAmount.toString(), - ) - .call(); - - if (records.records.length === 0) { - throw new PathNotFoundError( - sourceAssetType, - destAssetType, - destAmount, - ); + const query = this.queryBuilder.forStrictReceive({ + sourceAssets: [sourceAsset], + destinationAsset, + destinationAmount: destAmount, + }); + const results = await this.queryBuilder.execute(query); + + if (results.length === 0) { + throw new PathNotFoundError(sourceAssetType, destAssetType, destAmount); } - const best = records.records[0]!; - - const result: PathResult = { - path: best.path, - destinationAmount: BigInt(best.destination_amount), - sourceAmount: BigInt(best.source_amount), - }; - - this.cache.set(cacheKey, result); - return result; + // Results are sorted best-first (lowest source amount). + const best = results[0]!; + return { path: best.path, destinationAmount: best.destinationAmount, sourceAmount: best.sourceAmount }; } catch (err) { if (err instanceof PathNotFoundError) throw err; throw new PathRouterError( @@ -262,34 +214,10 @@ export class PathRouter { * Clear all cached paths. */ clearCache(): void { - this.cache.clear(); - } - - // ------------------------------------------------------------------------- - // Internal helpers - // ------------------------------------------------------------------------- - - private cacheKey(kind: "send" | "receive", req: PathRequest): string { - const src = req.sourceAsset.isNative() - ? "native" - : `${req.sourceAsset.getCode()}:${req.sourceAsset.getIssuer()}`; - const dst = req.destinationAsset.isNative() - ? "native" - : `${req.destinationAsset.getCode()}:${req.destinationAsset.getIssuer()}`; - return `path:${kind}:${src}:${dst}:${req.sourceAmount.toString()}`; + this.queryBuilder.clearCache(); } +} - private cacheKeyReceive( - sourceAsset: Asset, - destAmount: bigint, - destinationAsset: Asset, - ): string { - const src = sourceAsset.isNative() - ? "native" - : `${sourceAsset.getCode()}:${sourceAsset.getIssuer()}`; - const dst = destinationAsset.isNative() - ? "native" - : `${destinationAsset.getCode()}:${destinationAsset.getIssuer()}`; - return `path:receive:${src}:${dst}:${destAmount.toString()}`; - } +function assetType(asset: Asset): string { + return asset.isNative() ? "native" : `${asset.getCode()}:${asset.getIssuer()}`; } diff --git a/src/paymentProgressTracker.ts b/src/paymentProgressTracker.ts new file mode 100644 index 0000000..619d680 --- /dev/null +++ b/src/paymentProgressTracker.ts @@ -0,0 +1,153 @@ +/** + * Aggregates per-recipient payment status into a single progress view for + * long-running, multi-recipient invoices (multiple tranches or async + * recipient claims). + * + * One `InvoiceStatusPoller` (see poller.ts) is run per recipient leg, keyed + * by a composite `${invoiceId}:${accountId}` id so per-recipient pollers for + * the same invoice don't coalesce into a single poller. Progress is recomputed + * and emitted every time any recipient's status changes. + */ + +import { InvoiceStatusPoller } from "./poller.js"; +import { TypedEventEmitter } from "./events/TypedEventEmitter.js"; +import { getTrancheProgress } from "./trancheProgress.js"; +import type { + Invoice, + InvoicePaymentProgress, + RecipientPaymentState, +} from "./types.js"; + +/** Statuses treated as final/settled when computing percentComplete. */ +const SETTLED_STATUSES: ReadonlySet = new Set([ + "Released", + "Refunded", + "Cancelled", +]); + +/** Event map emitted by {@link PaymentProgressTracker}. */ +export interface PaymentProgressEventMap { + invoiceProgressUpdated: InvoicePaymentProgress; +} + +/** Options controlling {@link PaymentProgressTracker} behaviour. */ +export interface PaymentProgressTrackerOptions { + /** Poll interval per recipient leg, in milliseconds. Default: 5000. */ + pollIntervalMs?: number; + /** Fetch the current status of a single recipient's payment leg. */ + fetchRecipientStatus?: (invoiceId: string, accountId: string) => Promise; +} + +interface TrackedSubscription { + pollers: InvoiceStatusPoller[]; + unsubscribes: Array<() => void>; + states: Map; +} + +/** + * Subscribes callers to aggregated payment progress for an invoice, + * polling every recipient leg independently and reporting a single + * `InvoicePaymentProgress` snapshot on every change. + */ +export class PaymentProgressTracker extends TypedEventEmitter { + private readonly subscriptions = new Map(); + private readonly pollIntervalMs: number; + private readonly fetchRecipientStatus: (invoiceId: string, accountId: string) => Promise; + private readonly fetchInvoice: (invoiceId: string) => Promise; + + constructor( + fetchInvoice: (invoiceId: string) => Promise, + options: PaymentProgressTrackerOptions = {}, + ) { + super(); + this.fetchInvoice = fetchInvoice; + this.pollIntervalMs = options.pollIntervalMs ?? 5_000; + this.fetchRecipientStatus = options.fetchRecipientStatus ?? (async () => "Pending"); + } + + /** + * Start polling every recipient leg of `invoiceId` and invoke `onProgress` + * with an aggregated `InvoicePaymentProgress` snapshot on each change (and + * once immediately with the initial state). No-op if already subscribed. + */ + async subscribe( + invoiceId: string, + onProgress: (progress: InvoicePaymentProgress) => void, + ): Promise { + if (this.subscriptions.has(invoiceId)) return; + + const invoice = await this.fetchInvoice(invoiceId); + const trancheProgress = await getTrancheProgress(invoice).catch(() => undefined); + + const states = new Map(); + for (const recipient of invoice.recipients) { + states.set(recipient.address, { + accountId: recipient.address, + status: "Pending", + settledAmount: 0n, + lastUpdatedAt: Date.now(), + }); + } + + const subscription: TrackedSubscription = { pollers: [], unsubscribes: [], states }; + this.subscriptions.set(invoiceId, subscription); + + const emitProgress = () => { + const progress = this.computeProgress(invoiceId, states, trancheProgress); + this.emit("invoiceProgressUpdated", progress); + onProgress(progress); + }; + + for (const recipient of invoice.recipients) { + const compositeId = `${invoiceId}:${recipient.address}`; + const poller = new InvoiceStatusPoller( + { invoiceId: compositeId, pollIntervalMs: this.pollIntervalMs }, + () => this.fetchRecipientStatus(invoiceId, recipient.address), + ); + + const off = poller.on("invoiceStatusChanged", (event) => { + const state = states.get(recipient.address); + if (!state) return; + state.status = event.current; + state.lastUpdatedAt = event.timestamp; + if (SETTLED_STATUSES.has(event.current)) { + state.settledAmount = recipient.amount; + } + emitProgress(); + }); + + subscription.unsubscribes.push(off); + subscription.pollers.push(poller); + poller.start(); + } + + emitProgress(); + } + + /** Stop all pollers for `invoiceId` and remove its subscription. */ + unsubscribe(invoiceId: string): void { + const subscription = this.subscriptions.get(invoiceId); + if (!subscription) return; + for (const poller of subscription.pollers) poller.stop(); + for (const off of subscription.unsubscribes) off(); + this.subscriptions.delete(invoiceId); + } + + private computeProgress( + invoiceId: string, + states: Map, + trancheProgress: InvoicePaymentProgress["trancheProgress"], + ): InvoicePaymentProgress { + const recipientStates = Array.from(states.values()); + const settledLegs = recipientStates.filter((s) => SETTLED_STATUSES.has(s.status)).length; + const totalLegs = recipientStates.length; + const percentComplete = totalLegs === 0 ? 0 : (settledLegs / totalLegs) * 100; + + return { + invoiceId, + percentComplete, + recipientStates, + ...(trancheProgress ? { trancheProgress } : {}), + }; + } +} diff --git a/src/splitRollbackCoordinator.ts b/src/splitRollbackCoordinator.ts new file mode 100644 index 0000000..e5b8762 --- /dev/null +++ b/src/splitRollbackCoordinator.ts @@ -0,0 +1,123 @@ +/** + * Coordinates rollback/reconciliation of multi-recipient split payments. + * + * A split payment groups several `pay` operations for one invoice into a + * single transaction envelope (see `submitPayment`'s waterfall path). The + * on-chain submission is atomic, but application-layer acknowledgement of + * each leg (webhooks, database writes) can still partially fail after the + * transaction lands. `RollbackCoordinator` records the intended legs of a + * split as a checkpoint, lets callers mark each leg's outcome, and exposes + * `initiateRollback` to identify legs that still need corrective action. + */ + +import { IdempotencyManager } from "./idempotency.js"; +import { snapshotSplitRollback, type SplitRollbackRecord } from "./snapshot.js"; +import { UnknownSplitError } from "./errors.js"; +import { TypedEventEmitter } from "./events/TypedEventEmitter.js"; +import type { SplitLeg, SplitLegState, SplitRollbackCheckpoint } from "./types.js"; + +/** Event payloads emitted by {@link RollbackCoordinator}. */ +export interface SplitRollbackEventMap { + splitRollbackInitiated: { splitId: string; incomplete: SplitLeg[] }; + splitRollbackCompleted: { splitId: string; record: SplitRollbackRecord }; +} + +/** + * Tracks per-leg acknowledgement state for split payments and coordinates + * rollback when some legs are left in an unknown state. + */ +export class RollbackCoordinator extends TypedEventEmitter { + private readonly checkpoints = new Map(); + private readonly rollbackRecords = new Map(); + private readonly idempotency: IdempotencyManager; + + constructor(idempotency: IdempotencyManager = new IdempotencyManager()) { + super(); + this.idempotency = idempotency; + } + + /** + * Create a persistent rollback checkpoint recording all intended split + * legs. Must be called before `markLegSuccess`/`markLegFailed`. + */ + begin( + splitId: string, + invoiceId: string, + legs: Array<{ recipient: string; amount: bigint }>, + ): SplitRollbackCheckpoint { + const checkpoint: SplitRollbackCheckpoint = { + splitId, + invoiceId, + createdAt: Date.now(), + legs: legs.map((leg) => ({ ...leg, state: "pending" as SplitLegState })), + }; + this.checkpoints.set(splitId, checkpoint); + return checkpoint; + } + + /** Mark the leg at `legIndex` as successfully acknowledged. */ + markLegSuccess(splitId: string, legIndex: number): void { + this.setLegState(splitId, legIndex, "succeeded"); + } + + /** Mark the leg at `legIndex` as confirmed failed. */ + markLegFailed(splitId: string, legIndex: number): void { + this.setLegState(splitId, legIndex, "failed"); + } + + /** Legs that are neither confirmed succeeded nor confirmed failed. */ + getIncomplete(splitId: string): SplitLeg[] { + return this.getCheckpoint(splitId).legs.filter((leg) => leg.state === "pending"); + } + + /** + * Initiate a rollback for `splitId`: emits `splitRollbackInitiated` with + * the current incomplete legs, persists a rollback record via + * {@link snapshotSplitRollback}, then emits `splitRollbackCompleted`. + * + * Idempotent per `splitId` — calling this more than once returns the + * originally stored record without re-emitting events. + */ + initiateRollback(splitId: string): SplitRollbackRecord { + const checkpoint = this.getCheckpoint(splitId); + const key = this.idempotency.generateKey(splitId, "rollback"); + + const existing = this.rollbackRecords.get(splitId); + if (existing && this.idempotency.isDuplicate(key)) { + return existing; + } + + const incomplete = this.getIncomplete(splitId); + this.emit("splitRollbackInitiated", { splitId, incomplete }); + + const record = snapshotSplitRollback(checkpoint); + this.rollbackRecords.set(splitId, record); + this.idempotency.tryClaim(key, { txHash: record.snapshotId }); + + this.emit("splitRollbackCompleted", { splitId, record }); + return record; + } + + /** The persisted rollback record for `splitId`, if a rollback was initiated. */ + getRollbackRecord(splitId: string): SplitRollbackRecord | undefined { + return this.rollbackRecords.get(splitId); + } + + /** The current checkpoint for `splitId`, if one exists. */ + getCheckpointFor(splitId: string): SplitRollbackCheckpoint | undefined { + return this.checkpoints.get(splitId); + } + + private getCheckpoint(splitId: string): SplitRollbackCheckpoint { + const checkpoint = this.checkpoints.get(splitId); + if (!checkpoint) throw new UnknownSplitError(splitId); + return checkpoint; + } + + private setLegState(splitId: string, legIndex: number, state: SplitLegState): void { + const checkpoint = this.getCheckpoint(splitId); + const leg = checkpoint.legs[legIndex]; + if (!leg) throw new UnknownSplitError(`${splitId}[${legIndex}]`); + leg.state = state; + } +} diff --git a/src/types.ts b/src/types.ts index b439457..f668010 100644 --- a/src/types.ts +++ b/src/types.ts @@ -125,6 +125,47 @@ export interface Recipient { import { StellarSplitError } from "./errors.js"; +// --------------------------------------------------------------------------- +// Split Rollback Coordinator Types +// --------------------------------------------------------------------------- + +/** Lifecycle state of a single leg tracked by the rollback coordinator. */ +export type SplitLegState = "pending" | "succeeded" | "failed"; + +/** A single recipient leg within a multi-recipient split payment. */ +export interface SplitLeg { + /** Stellar address of this leg's recipient. */ + recipient: string; + /** Amount owed to this recipient, in stroops. */ + amount: bigint; + /** Current reconciliation state of this leg. */ + state: SplitLegState; +} + +/** Result of submitting a multi-recipient split payment. */ +export interface SplitResult { + /** Identifier grouping the legs of this split (typically the tx hash). */ + splitId: string; + /** Invoice this split payment was submitted for. */ + invoiceId: string; + /** Transaction hash of the on-chain submission. */ + txHash: string; + /** Per-recipient legs included in the split. */ + legs: SplitLeg[]; +} + +/** A persistent checkpoint recording the intended legs of a split payment. */ +export interface SplitRollbackCheckpoint { + /** Identifier grouping the legs of this split. */ + splitId: string; + /** Invoice this split payment was submitted for. */ + invoiceId: string; + /** Unix epoch ms when the checkpoint was created. */ + createdAt: number; + /** Per-recipient legs, in submission order. */ + legs: SplitLeg[]; +} + // --------------------------------------------------------------------------- // AMM Calculator Types // ---------------------------------------------------------------------------