diff --git a/.gitignore b/.gitignore index a1dbe4b..3504f38 100644 --- a/.gitignore +++ b/.gitignore @@ -6,3 +6,4 @@ examples/*/dist/ *.tsbuildinfo coverage/ .DS_Store +fix.md diff --git a/src/account/batchOperations.ts b/src/account/batchOperations.ts new file mode 100644 index 0000000..bae1cb5 --- /dev/null +++ b/src/account/batchOperations.ts @@ -0,0 +1,673 @@ +/** + * Batch account operations (fix.md). + * + * A unified batch execution model for account-level operations — trustline + * creation, payments, and key rotation — with bounded concurrency, operation + * grouping, progress tracking, and partial-failure handling. + * + * Design principles + * ----------------- + * - A single idempotency-aware executor (`runBatchOperations`) drives every + * batch. It is transport-agnostic: the caller supplies a per-operation + * `runner`. + * - Each operation is planned with a stable, caller-supplied `id` and its + * result is tracked independently. A failure in one operation never + * invalidates unrelated operations. + * - Retry is never blind: an operation that has already succeeded (or is in + * `previouslyCompletedIds`) is never re-run, so `maxRetries` cannot + * duplicate a completed/possibly-submitted operation. + * - Progress is exposed incrementally via `onProgress` and summarized in the + * final report. + */ + +import { ok, err, SorokitErrorCode } from "../shared/response"; +import type { SorokitResult } from "../shared/response"; +import { + buildPaymentTransaction, + buildTrustlineTransaction, +} from "../transaction/buildTransaction"; +import type { TrustlineParams } from "../transaction/types"; +import { rotateAccountKey } from "./keyRotation"; +import type { RotateAccountKeyParams } from "./keyRotation"; +import type { ResolvedNetworkConfig } from "../shared/types"; +import { sleep } from "../shared/utils"; + +// ─── Types ──────────────────────────────────────────────────────────────────── + +/** Lifecycle status of a single batch operation. */ +export type BatchOperationStatus = + | "pending" + | "running" + | "success" + | "failed" + | "skipped" + | "retried"; + +/** Metadata for an operation in a batch plan. */ +export interface BatchOperation { + /** Stable, caller-supplied identifier used for idempotency. */ + id: string; + /** + * Runnable that actually performs the operation. + * Return `ok(data)` on success, or an error result on failure. + */ + runner: BatchRunner; + /** Optional input associated with the operation (for reporting). */ + input?: unknown; +} + +export type BatchRunner = ( + operationId: string, +) => Promise>; + +/** Individual tracked result of one batch operation. */ +export interface BatchOperationResult { + id: string; + status: BatchOperationStatus; + /** Available when the operation succeeded. */ + data?: T; + /** Human-readable error message when the operation ultimately failed. */ + errorMessage?: string; + errorCode?: string; + /** Number of execution attempts (1 = initial, >1 = retried). */ + attempts: number; + /** Number of retries performed. */ + retries: number; + /** True when the operation was skipped (already completed / explicit skip). */ + skipped?: boolean; +} + +/** Incremental snapshot of batch progress. */ +export interface BatchProgress { + total: number; + planned: number; + running: number; + succeeded: number; + failed: number; + skipped: number; + retried: number; + completed: number; + /** Epoch-ms timestamp when the batch started. */ + startedAt: number; + /** Elapsed ms at the time the snapshot was produced. */ + elapsedMs: number; +} + +/** Configuration for a batch execution. */ +export interface BatchExecutorConfig { + /** + * Maximum number of operations to schedule into a single grouping wave. + * Larger values allow more operations in flight at once; smaller values + * give finer-grained progress callbacks. + */ + batchSize?: number; + /** Maximum number of operations running concurrently (default: 5). */ + concurrency?: number; + /** + * Maximum retry attempts for retryable failures (default: 2). + * Successful or previously-completed operations are never retried. + */ + maxRetries?: number; + /** Base delay (ms) before the first retry; backoff is exponential (default: 200). */ + retryDelayMs?: number; + /** + * Idempotency set. Operations whose id appears here are assumed already + * completed and are skipped rather than executed. + */ + previouslyCompletedIds?: Iterable; + /** Called whenever progress changes with a live snapshot. */ + onProgress?: (progress: BatchProgress) => void; + /** + * Optional predicate classifying an error as retryable. Defaults to + * network/timeout/service-unavailable errors. + */ + isRetryable?: (error: SorokitResult) => boolean; + /** Inject determinism for clock in tests. */ + now?: () => number; + /** Time source for the retry sleep (overridable in tests). */ + delay?: (ms: number) => Promise; +} + +/** Final report of a batch execution. */ +export interface BatchExecutionReport { + results: BatchOperationResult[]; + summary: { + total: number; + succeeded: number; + failed: number; + skipped: number; + retried: number; + completed: number; + /** True when every operation ended in success. */ + allSucceeded: boolean; + }; + /** Final progress snapshot. */ + progress: BatchProgress; + /** Epoch-ms timestamp when the batch finished. */ + finishedAt: number; +} + +// ─── Defaults & helpers ─────────────────────────────────────────────────────── + +const DEFAULT_CONCURRENCY = 5; +const DEFAULT_MAX_RETRIES = 2; +const DEFAULT_RETRY_DELAY_MS = 200; +const MAX_RETRY_DELAY_MS = 10_000; + +const RETRYABLE_CODES: ReadonlySet = new Set([ + SorokitErrorCode.NETWORK_ERROR, + SorokitErrorCode.OPERATION_TIMEOUT, + SorokitErrorCode.SERVICE_UNAVAILABLE, +]); + +function defaultIsRetryable(error: SorokitResult): boolean { + if (error.status !== "error" || !error.error) return false; + return RETRYABLE_CODES.has(error.error.code); +} + +// ─── Progress accounting ────────────────────────────────────────────────────── + +function createProgressState(total: number, startedAt: number) { + return { + planned: 0, + running: 0, + succeeded: 0, + failed: 0, + skipped: 0, + retried: 0, + completed: 0, + }; +} + +function snapshotProgress( + startedAt: number, + counter: ReturnType, + total: number, + now: () => number, +): BatchProgress { + return { + total, + planned: counter.planned, + running: counter.running, + succeeded: counter.succeeded, + failed: counter.failed, + skipped: counter.skipped, + retried: counter.retried, + completed: counter.completed, + startedAt, + elapsedMs: now() - startedAt, + }; +} + +// ─── Core executor ──────────────────────────────────────────────────────────── + +export interface QueueItem { + operation: BatchOperation; +} + +/** + * Run a batch of operations with bounded concurrency, operation grouping, + * progress tracking, and partial-failure handling. + * + * Behavioral guarantees: + * - Each operation runs exactly once if it succeeds; retryable failures are + * retried up to `maxRetries` times before being marked `failed`. + * - Operations listed in `previouslyCompletedIds` are marked `skipped` + * (never executed again) to make re-running a batch idempotent. + * - A failure in one operation is isolated and does not affect other + * operations. + * + * @param operations - The operations to plan and execute. + * @param userConfig - Batch size, concurrency, retry, and progress options. + * @returns A {@link BatchExecutionReport} with per-operation outcomes. + */ +export async function runBatchOperations( + operations: BatchOperation[], + userConfig: BatchExecutorConfig = {}, +): Promise> { + const config: Required< + Pick< + BatchExecutorConfig, + "batchSize" | "concurrency" | "maxRetries" | "retryDelayMs" | "now" + > + > = { + batchSize: userConfig.batchSize ?? operations.length, + concurrency: Math.max(1, userConfig.concurrency ?? DEFAULT_CONCURRENCY), + maxRetries: Math.max(0, userConfig.maxRetries ?? DEFAULT_MAX_RETRIES), + retryDelayMs: Math.max(0, userConfig.retryDelayMs ?? DEFAULT_RETRY_DELAY_MS), + now: userConfig.now ?? Date.now, + }; + const delay = userConfig.delay ?? sleep; + const isRetryable = + userConfig.isRetryable ?? + ((error: SorokitResult) => defaultIsRetryable(error)); + const completedIds = new Set(userConfig.previouslyCompletedIds ?? []); + + const startedAt = config.now(); + const counter = createProgressState(operations.length, startedAt); + const resultsArray: BatchOperationResult[] = []; + + const emit = () => { + userConfig.onProgress?.( + snapshotProgress(startedAt, counter, operations.length, config.now), + ); + }; + + const markSkipped = (id: string) => { + counter.skipped += 1; + counter.completed += 1; + resultsArray.push({ + id, + status: "skipped", + attempts: 0, + retries: 0, + skipped: true, + }); + }; + + // Plan operations in groups of `batchSize` so big batches can be chunked + // with incremental progress callbacks after each group is scheduled. + const plan = [...operations]; + const queue: QueueItem[] = []; + + // Skip previously completed operations up front (idempotency), emitting + // progress in `batchSize` chunks so large batches report incrementally. + let plannedCount = 0; + for (const op of plan) { + if (completedIds.has(op.id)) { + markSkipped(op.id); + if (plannedCount % config.batchSize === 0) emit(); + continue; + } + counter.planned += 1; + plannedCount += 1; + queue.push({ operation: op }); + if (plannedCount % config.batchSize === 0) emit(); + } + emit(); + + // ── Bounded-concurrency worker pool ────────────────────────────────────── + let nextIndex = 0; + const active = new Set(); + + const processOne = async (index: number): Promise => { + if (index >= queue.length) return; + const item = queue[index]; + if (!item) return; + if (completedIds.has(item.operation.id)) { + markSkipped(item.operation.id); + return; + } + + counter.running += 1; + emit(); + + const { operation } = item; + + const outcome = await runWithRetry(operation, { + maxRetries: config.maxRetries, + retryDelayMs: config.retryDelayMs, + isRetryable, + delay, + }); + + counter.running -= 1; + if (outcome.status === "success") { + counter.succeeded += 1; + counter.completed += 1; + // Mark completed so re-runs / later scans skip it (idempotency). + completedIds.add(operation.id); + } else { + counter.failed += 1; + counter.completed += 1; + } + counter.retried += outcome.retries; + + resultsArray.push({ + id: operation.id, + status: outcome.status, + ...(outcome.data !== undefined ? { data: outcome.data } : {}), + ...(outcome.errorMessage !== undefined + ? { errorMessage: outcome.errorMessage } + : {}), + ...(outcome.errorCode !== undefined ? { errorCode: outcome.errorCode } : {}), + attempts: outcome.attempts, + retries: outcome.retries, + }); + emit(); + + // Pull the next item as soon as a slot frees up (bounded concurrency). + const next = nextIndex++; + if (next < queue.length) { + active.add(next); + // Fire-and-forget each worker; the pool tracks completion via a final await. + void processOne(next).finally(() => active.delete(next)); + } else { + active.delete(index); + } + }; + + // Seed up to `concurrency` workers. + const initialWorkers = Math.min(config.concurrency, queue.length); + for (let i = 0; i < initialWorkers; i++) { + const index = nextIndex++; + active.add(index); + void processOne(index).finally(() => active.delete(index)); + } + + // Wait for every worker to finish. + await waitForWorkers(() => active.size === 0); + + const completed = counter.completed; + const summary = { + total: operations.length, + succeeded: counter.succeeded, + failed: counter.failed, + skipped: counter.skipped, + retried: counter.retried, + completed, + allSucceeded: + counter.failed === 0 && counter.skipped === 0 && counter.succeeded === totalOf(operations), + }; + + return { + results: resultsArray, + summary, + progress: snapshotProgress( + startedAt, + counter, + operations.length, + config.now, + ), + finishedAt: config.now(), + }; +} + +function totalOf(operations: BatchOperation[]): number { + return operations.length; +} + +interface RetryOutcome { + status: "success" | "failed"; + attempts: number; + retries: number; + data?: T; + errorMessage?: string; + errorCode?: string; +} + +async function runWithRetry( + operation: BatchOperation, + opts: { + maxRetries: number; + retryDelayMs: number; + isRetryable: (error: SorokitResult) => boolean; + delay: (ms: number) => Promise; + }, +): Promise> { + let lastError: SorokitResult | undefined; + let executedAttempts = 0; + let retries = 0; + for (let attempt = 0; attempt <= opts.maxRetries; attempt++) { + let result: SorokitResult; + try { + result = await operation.runner(operation.id); + } catch (e) { + result = err( + SorokitErrorCode.UNKNOWN, + "Operation runner threw an error.", + e, + ); + } + executedAttempts += 1; + if (result.status === "ok") { + return { + status: "success", + attempts: executedAttempts, + retries, + ...(result.data !== undefined ? { data: result.data } : {}), + }; + } + lastError = result as SorokitResult; + if (attempt < opts.maxRetries && opts.isRetryable(result as SorokitResult)) { + retries += 1; + const backoffMs = Math.min( + opts.retryDelayMs * Math.pow(2, attempt), + MAX_RETRY_DELAY_MS, + ); + await opts.delay(backoffMs); + continue; + } + break; + } + + const error = lastError; + return { + status: "failed", + attempts: executedAttempts, + retries, + ...(error && error.status === "error" && error.error + ? { errorMessage: error.error.message, errorCode: error.error.code } + : { errorMessage: "Operation failed after retries." }), + }; +} + +/** Simplified busy-wait on worker completion (bounded by average op time). */ +async function waitForWorkers(isDone: () => boolean): Promise { + while (!isDone()) { + // Yield to the event loop so pending microtasks (worker promises) advance. + await new Promise((r) => setTimeout(r, 0)); + } +} + +// ─── Convenience wrappers ───────────────────────────────────────────────────── + +/** Result payload for a trustline creation operation in a bulk run. */ +export interface BulkTrustlineResult { + account: string; + assetCode: string; + assetIssuer: string; + /** Built transaction XDR when submission was skipped. */ + xdr?: string; +} + +export interface BulkCreateTrustlineOp { + /** Stable operation id (idempotency). */ + id: string; + account: string; + assetCode: string; + assetIssuer: string; + limit?: string; +} + +export interface BulkCreateTrustlinesInput { + horizonUrl: string; + networkConfig: ResolvedNetworkConfig; + /** Accounts (G-addresses) to establish trustlines for. */ + accounts: string[]; + /** Assets to trust. */ + assets: Array<{ code: string; issuer: string; limit?: string }>; + config?: BatchExecutorConfig; + /** Optional sign-and-submit callback. When omitted, only the XDR is built. */ + submit?: (xdr: string) => Promise>; +} + +/** + * Establish trustlines for many accounts against many assets, each tracked + * independently with bounded concurrency and retry. + */ +export async function bulkCreateTrustlines( + input: BulkCreateTrustlinesInput, +): Promise> { + const ops: BatchOperation[] = []; + for (const account of input.accounts) { + for (const asset of input.assets) { + const id = `tl:${account}:${asset.code}:${asset.issuer}`; + const runner = async (): Promise> => { + const params: TrustlineParams = { + assetCode: asset.code, + assetIssuer: asset.issuer, + ...(asset.limit !== undefined ? { limit: asset.limit } : {}), + }; + const built = await buildTrustlineTransaction( + input.horizonUrl, + input.networkConfig, + account, + params, + ); + if (built.status === "error") return built; + if (input.submit) { + const submitted = await input.submit(built.data); + if (submitted.status === "error") { + return err( + submitted.error!.code, + submitted.error!.message, + submitted.error!.cause, + ); + } + } + return ok({ + account, + assetCode: asset.code, + assetIssuer: asset.issuer, + ...(input.submit ? {} : { xdr: built.data }), + }); + }; + ops.push({ id, runner, input: { account, assetCode: asset.code } }); + } + } + return runBatchOperations(ops, input.config); +} + +// ─── Bulk payments ──────────────────────────────────────────────────────────── + +export interface BulkPaymentOp { + id: string; + source: string; + params: import("../transaction/types").PaymentParams; +} + +export interface BulkSendPaymentsInput { + horizonUrl: string; + networkConfig: ResolvedNetworkConfig; + transactions: BulkPaymentOp[]; + config?: BatchExecutorConfig; + /** Optional sign-and-submit callback. When omitted, only the XDR is built. */ + submit?: (xdr: string) => Promise>; +} + +export interface BulkPaymentResult { + source: string; + destination: string; + xdr?: string; +} + +/** + * Execute many payments concurrently, tracking each independently. Building + * uses the existing `buildPaymentTransaction`; when a `submit` callback is + * supplied, the built XDR is signed and submitted by the caller. + */ +export async function bulkSendPayments( + input: BulkSendPaymentsInput, +): Promise> { + const ops: BatchOperation[] = input.transactions.map( + (tx) => { + const runner = async (): Promise> => { + const built = await buildPaymentTransaction( + input.horizonUrl, + input.networkConfig, + tx.source, + tx.params, + ); + if (built.status === "error") return built; + if (input.submit) { + const submitted = await input.submit(built.data); + if (submitted.status === "error") { + return err( + submitted.error!.code, + submitted.error!.message, + submitted.error!.cause, + ); + } + } + return ok({ + source: tx.source, + destination: tx.params.destination, + ...(input.submit ? {} : { xdr: built.data }), + }); + }; + return { id: tx.id, runner, input: { source: tx.source } }; + }, + ); + return runBatchOperations(ops, input.config); +} + +// ─── Bulk key rotation ──────────────────────────────────────────────────────── + +export interface BulkRotateKeyOp { + id: string; + account: string; + oldKey: string; + newKey: string; + newKeyWeight?: number; +} + +export interface BulkRotateKeysInput { + horizonUrl: string; + networkConfig: ResolvedNetworkConfig; + /** Accounts whose keys are being rotated. */ + accounts: BulkRotateKeyOp[]; + config?: BatchExecutorConfig; + /** Optional sign-and-submit callback. When omitted, only the XDR is built. */ + submit?: (xdr: string) => Promise>; +} + +export interface BulkRotateKeyResult { + account: string; + xdr?: string; +} + +/** + * Rotate signing keys across many accounts, each tracked independently. + * Backed by the existing `rotateAccountKey` API. + */ +export async function bulkRotateKeys( + input: BulkRotateKeysInput, +): Promise> { + const ops: BatchOperation[] = input.accounts.map( + (op) => { + const runner = async (): Promise> => { + const params: RotateAccountKeyParams = { + account: op.account, + oldKey: op.oldKey, + newKey: op.newKey, + ...(op.newKeyWeight !== undefined + ? { newKeyWeight: op.newKeyWeight } + : {}), + }; + const built = await rotateAccountKey( + input.horizonUrl, + input.networkConfig, + params, + ); + if (built.status === "error") return built; + if (input.submit) { + const submitted = await input.submit(built.data); + if (submitted.status === "error") { + return err( + submitted.error!.code, + submitted.error!.message, + submitted.error!.cause, + ); + } + } + return ok({ + account: op.account, + ...(input.submit ? {} : { xdr: built.data }), + }); + }; + return { id: op.id, runner, input: { account: op.account } }; + }, + ); + return runBatchOperations(ops, input.config); +} diff --git a/src/account/index.ts b/src/account/index.ts index c62e60f..fee799c 100644 --- a/src/account/index.ts +++ b/src/account/index.ts @@ -83,6 +83,32 @@ export type { BalanceForecastResult, } from "./balanceForecast"; +// ─── Batch account operations (#514) ───────────────────────────────────────── +export { + bulkCreateTrustlines, + bulkSendPayments, + bulkRotateKeys, + runBatchOperations, +} from "./batchOperations"; +export type { + BatchOperation, + BatchRunner, + BatchOperationResult, + BatchOperationStatus, + BatchProgress, + BatchExecutorConfig, + BatchExecutionReport, + BulkTrustlineResult, + BulkCreateTrustlineOp, + BulkCreateTrustlinesInput, + BulkPaymentOp, + BulkSendPaymentsInput, + BulkPaymentResult, + BulkRotateKeyOp, + BulkRotateKeysInput, + BulkRotateKeyResult, +} from "./batchOperations"; + // ─── Multi-wallet portfolio aggregation (#525) ──────────────────────────────── export { aggregatePortfolio, assetIdentifier } from "./portfolioAggregation"; export type { diff --git a/src/index.ts b/src/index.ts index f5235c0..ef9def9 100644 --- a/src/index.ts +++ b/src/index.ts @@ -222,6 +222,32 @@ export type { GetAccountActivitySummaryOptions, } from "./account/getAccountActivitySummary"; +// ─── Batch account operations (#514) ────────────────────────────────────────── +export { + bulkCreateTrustlines, + bulkSendPayments, + bulkRotateKeys, + runBatchOperations, +} from "./account/batchOperations"; +export type { + BatchOperation, + BatchRunner, + BatchOperationResult, + BatchOperationStatus, + BatchProgress, + BatchExecutorConfig, + BatchExecutionReport, + BulkTrustlineResult, + BulkCreateTrustlineOp, + BulkCreateTrustlinesInput, + BulkPaymentOp, + BulkSendPaymentsInput, + BulkPaymentResult, + BulkRotateKeyOp, + BulkRotateKeysInput, + BulkRotateKeyResult, +} from "./account/batchOperations"; + // ─── Transaction validation ─────────────────────────────────────────────────── export { createHashMemo, @@ -380,6 +406,27 @@ export type { WebhookPayload, WebhookEventDetails, } from "./transaction/webhooks"; + +// ─── Payment notifications (fix.md) ─────────────────────────────────────────── +export { + registerPaymentWebhook, + unregisterPaymentWebhook, + listPaymentWebhooks, + clearPaymentWebhooks, + triggerPaymentNotifications, + dispatchPaymentNotification, + generatePaymentEventId, + isPaymentNotificationEvent, + PAYMENT_NOTIFICATION_EVENTS, +} from "./transaction/paymentNotifications"; +export type { + PaymentNotificationEvent, + PaymentWebhookOptions, + PaymentWebhookPayload, + PaymentWebhookRegistration, + PaymentNotificationInput, + PaymentNotificationChannel, +} from "./transaction/paymentNotifications"; export { DEFAULT_PRICE_CACHE_TTL_MS, exportTransactionHistory, @@ -533,6 +580,32 @@ export type { OnContractUpgrade, ContractVersionOptions, } from "./soroban/contractVersion"; +// ─── Contract metadata versioning & compatibility (fix.md) ──────────────────── +export { + computeContractMetadataFingerprint, + buildContractMetadataSnapshot, + checkContractMetadataCompatibility, + checkStaleContractMetadata, + applyContractMetadataMigration, + invalidateContractMetadataForIncompatibility, + invalidateCachedContractMetadata, +} from "./soroban/contractMetadataCompatibility"; +export type { + ContractMetadataVersion, + ContractMetadataSnapshot, + ContractMetadataChange, + ContractMetadataChangeKind, + ContractMetadataCompatibilityStatus, + ContractMetadataCompatibilityReport, + ContractMetadataMigration, + ContractMetadataMigrationHook, + ContractMetadataMigrationResult, + BuildMetadataSnapshotInput, + CheckCompatibilityInput, + StaleMetadataCheckInput, + StaleMetadataCheckResult, + InvalidateMetadataInput, +} from "./soroban/contractMetadataCompatibility"; // ─── Contract error decoding (#391) ─────────────────────────────────────────── export { decodeContractError, diff --git a/src/soroban/contractMetadataCompatibility.ts b/src/soroban/contractMetadataCompatibility.ts new file mode 100644 index 0000000..f15c22f --- /dev/null +++ b/src/soroban/contractMetadataCompatibility.ts @@ -0,0 +1,611 @@ +/** + * Contract metadata version tracking and compatibility validation. + * + * Problem (fix.md) + * ---------------- + * Contract metadata can evolve as contracts are upgraded. Before using cached + * metadata the SDK should validate that it corresponds to the deployed contract + * interface, distinguishing breaking from non-breaking changes, exposing + * migration hooks, and invalidating stale cached metadata. + * + * Design notes + * ------------ + * - Fingerprints, not semantic version strings, drive compatibility: the + * interface of a contract is reduced to a deterministic fingerprint so + * identical metadata always yields the same fingerprint. Semantic version + * strings are treated as advisory only. + * - Metadata is always associated with a contract identifier. + * - The subsystem is standalone and additive: it consumes the existing + * `ContractMethod[]` / `ContractSchema` shapes and existing cache + * invalidation helpers, so existing metadata consumers remain untouched. + */ + +import { ok } from "../shared/response"; +import type { SorokitResult } from "../shared/response"; +import type { SorokitCache } from "../shared/cache"; +import type { ContractMethod } from "./types"; +import type { + ContractSchema, +} from "./contractMetadata"; + +/** + * A metadata fingerprint plus its associated contract identifier. + * This is the version identity used for compatibility checks — it does NOT + * rely on semantic version strings. + */ +export interface ContractMetadataVersion { + /** Stellar contract address (C...). */ + contractId: string; + /** Deterministic fingerprint of the contract interface. */ + fingerprint: string; + /** + * Optional advisory version string (e.g. from `contractmetav0`). Used only + * as a fast-path hint; the fingerprint is authoritative. + */ + version?: string; +} + +/** Metadata that is pinned to a contract identifier. */ +export interface ContractMetadataSnapshot { + /** Contract identifier the metadata belongs to. */ + contractId: string; + /** The discovered contract interface (methods). */ + methods: ContractMethod[]; + /** Deterministic fingerprint of `methods`. */ + fingerprint: string; + /** Optional advisory version string. */ + version?: string; +} + +/** Canonicalized method descriptor used to derive fingerprints. */ +interface NormalizedMethod { + name: string; + args: string[]; + return: string; +} + +/** Kind of interface difference detected between two metadata snapshots. */ +export type ContractMetadataChangeKind = + | "ADDED_METHOD" + | "REMOVED_METHOD" + | "CHANGED_ARG_TYPE" + | "CHANGED_ARG_COUNT" + | "CHANGED_RETURN_TYPE"; + +/** A single detected difference between two metadata snapshots. */ +export interface ContractMetadataChange { + kind: ContractMetadataChangeKind; + /** The method the change affects, when applicable. */ + method: string; + /** Human-readable description of the change. */ + message: string; + /** Whether this change is breaking for existing callers. */ + breaking: boolean; +} + +/** Compatibility outcome between two metadata versions. */ +export type ContractMetadataCompatibilityStatus = + | "identical" + | "compatible" + | "incompatible" + | "missing"; + +/** Structured report from a metadata compatibility check. */ +export interface ContractMetadataCompatibilityReport { + /** Overall compatibility classification. */ + status: ContractMetadataCompatibilityStatus; + /** Convenience boolean: `true` for identical/compatible. */ + compatible: boolean; + /** Every detected interface difference (empty when identical/missing). */ + changes: ContractMetadataChange[]; + /** Non-fatal advisories (e.g. missing version, metadata absent). */ + warnings: string[]; + /** Fatal explanation when status is `incompatible`. */ + errors: string[]; + /** True when stale/incompatible cached metadata must be invalidated. */ + shouldInvalidate: boolean; +} + +/** A single migration step transforming candidate metadata. */ +export interface ContractMetadataMigration { + /** When true (and the transformation preserves compatibility), the metadata is considered migrated. */ + applied: boolean; + /** Human-readable description of the migration performed. */ + description: string; +} + +/** + * Application-defined migration hook. + * + * Receives the candidate metadata snapshot and may transform it to a newer + * interface. Return `null` when the hook cannot migrate the metadata (the + * change is treated as incompatible). + */ +export type ContractMetadataMigrationHook = ( + candidate: ContractMetadataSnapshot, +) => ContractMetadataSnapshot | null | void; + +/** Result of applying migration hooks to incompatible metadata. */ +export interface ContractMetadataMigrationResult { + /** The metadata after migrations were applied, if a hook produced one. */ + snapshot: ContractMetadataSnapshot | null; + /** Migrations that were applied, in order. */ + migrations: ContractMetadataMigration[]; + /** True when the original metadata was left un-migratable (incompatible). */ + blocked: boolean; +} + +// ─── Fingerprint computation ────────────────────────────────────────────────── + +/** + * Reduce any supported metadata shape to a canonical list of normalized + * methods. Accepts the raw array of discovered methods or a typed schema. + */ +function normalizeMethods( + source: ContractMethod[] | ContractSchema, +): NormalizedMethod[] { + const methods: ContractMethod[] = Array.isArray(source) + ? source + : source.methods.map((m) => ({ + name: m.name, + inputs: (m.params ?? []).map((p) => ({ name: p.name, type: p.type })), + returnType: m.returnType, + })); + + return methods.map((m) => ({ + name: m.name, + args: (m.inputs ?? []).map((input) => `${input.name}:${input.type}`), + return: m.returnType ?? "void", + })); +} + +/** Deterministic 32-bit hash (FNV-1a) — stable across runs and runtimes. */ +function fnv1a(data: string): string { + let hash = 0x811c9dc5; + for (let i = 0; i < data.length; i++) { + hash ^= data.charCodeAt(i); + hash = (hash * 0x01000193) >>> 0; + } + return hash.toString(16).padStart(8, "0"); +} + +/** + * Compute a deterministic fingerprint of a contract interface. + * + * The fingerprint is derived from the sorted, canonicalized method signatures + * (name, argument types, return type) — NOT from semantic version strings — so + * identical interfaces always produce identical fingerprints. This is the + * authoritative identity used for compatibility validation. + */ +export function computeContractMetadataFingerprint( + source: ContractMethod[] | ContractSchema, +): string { + const methods = normalizeMethods(source) + .map( + (m) => + `${m.name}(${m.args.join(",")}):${m.return}`, + ) + .sort(); + return fnv1a(methods.join("|")); +} + +/** Normalize a fingerprint to a stable hex string (empty → "missing"). */ +function normalizeFingerprint(fingerprint: string | undefined | null): string { + return typeof fingerprint === "string" && fingerprint.length > 0 + ? fingerprint + : ""; +} + +// ─── Snapshot construction ──────────────────────────────────────────────────── + +export interface BuildMetadataSnapshotInput { + contractId: string; + methods: ContractMethod[] | ContractSchema; + /** Optional advisory version string. */ + version?: string; + /** Override fingerprint when the caller has a precomputed one. */ + fingerprint?: string; +} + +/** + * Build a fingerprint-pinned metadata snapshot associated with a contract + * identifier. + * + * @returns A {@link ContractMetadataSnapshot} with the fingerprint computed + * deterministically from the interface (or the caller-supplied override). + */ +export function buildContractMetadataSnapshot( + input: BuildMetadataSnapshotInput, +): ContractMetadataSnapshot { + const methods: ContractMethod[] = Array.isArray(input.methods) + ? input.methods + : (input.methods as ContractSchema).methods.map((m) => ({ + name: m.name, + inputs: (m.params ?? []).map((p) => ({ name: p.name, type: p.type })), + returnType: m.returnType, + })); + + return { + contractId: input.contractId, + methods, + fingerprint: + input.fingerprint ?? computeContractMetadataFingerprint(input.methods), + ...(input.version !== undefined ? { version: input.version } : {}), + }; +} + +// ─── Compatibility diffing ──────────────────────────────────────────────────── + +function diffMethods( + baseline: NormalizedMethod[], + candidate: NormalizedMethod[], +): ContractMetadataChange[] { + const changes: ContractMetadataChange[] = []; + const baselineByName = new Map(baseline.map((m) => [m.name, m])); + const candidateByName = new Map(candidate.map((m) => [m.name, m])); + + for (const method of baseline) { + const next = candidateByName.get(method.name); + if (!next) { + changes.push({ + kind: "REMOVED_METHOD", + method: method.name, + message: `Method '${method.name}' was removed from the contract interface.`, + breaking: true, + }); + continue; + } + + if (next.args.length !== method.args.length) { + changes.push({ + kind: "CHANGED_ARG_COUNT", + method: method.name, + message: `Method '${method.name}' argument count changed from ${method.args.length} to ${next.args.length}.`, + breaking: true, + }); + } else if (next.args.join("|") !== method.args.join("|")) { + changes.push({ + kind: "CHANGED_ARG_TYPE", + method: method.name, + message: `Method '${method.name}' argument types changed from [${method.args.join(", ")}] to [${next.args.join(", ")}].`, + breaking: true, + }); + } + + if (next.return !== method.return) { + changes.push({ + kind: "CHANGED_RETURN_TYPE", + method: method.name, + message: `Method '${method.name}' return type changed from '${method.return}' to '${next.return}'.`, + breaking: true, + }); + } + } + + for (const method of candidate) { + if (!baselineByName.has(method.name)) { + changes.push({ + kind: "ADDED_METHOD", + method: method.name, + message: `Method '${method.name}' was added to the contract interface (non-breaking).`, + breaking: false, + }); + } + } + + return changes; +} + +// ─── Public compatibility check ─────────────────────────────────────────────── + +export interface CheckCompatibilityInput { + /** The metadata the application currently relies on (baseline). */ + baseline: ContractMetadataSnapshot; + /** Freshly discovered metadata (candidate) for the same contract. */ + candidate?: ContractMetadataSnapshot; + /** Expected fingerprint for the baseline contract, when the caller has one. */ + expectedFingerprint?: string; +} + +/** + * Compare two contract interface versions and classify the result. + * + * Distinguishes: + * - `identical` — fingerprints (or both declared versions) match exactly. + * - `compatible` — only non-breaking changes (added methods). + * - `incompatible` — breaking changes (removed methods, changed signatures). + * - `missing` — the candidate/expected metadata could not be resolved. + * + * `shouldInvalidate` is `true` when the baseline (cached) metadata must be + * discarded because it no longer matches the deployed interface. + * + * @param input - Baseline snapshot plus candidate/expected identity. + * @returns A structured {@link ContractMetadataCompatibilityReport}. + */ +export function checkContractMetadataCompatibility( + input: CheckCompatibilityInput, +): ContractMetadataCompatibilityReport { + const warnings: string[] = []; + const errors: string[] = []; + + const candidateFingerprint = normalizeFingerprint( + input.candidate?.fingerprint ?? input.expectedFingerprint, + ); + const baselineFingerprint = normalizeFingerprint(input.baseline.fingerprint); + + // Missing candidate metadata → cannot confirm the cached metadata is valid. + if ( + !input.candidate && + (input.expectedFingerprint === undefined || candidateFingerprint === "") + ) { + return { + status: "missing", + compatible: false, + changes: [], + warnings: ["No candidate metadata was provided; cannot confirm the cached metadata is current."], + errors: [], + shouldInvalidate: false, + }; + } + + // Fast path: identical fingerprint → identical interface. + if (baselineFingerprint !== "" && baselineFingerprint === candidateFingerprint) { + return { + status: "identical", + compatible: true, + changes: [], + warnings, + errors, + shouldInvalidate: false, + }; + } + + // Fast path: both declare the same advisory version string. + if ( + input.baseline.version !== undefined && + input.candidate?.version !== undefined && + input.baseline.version === input.candidate.version + ) { + return { + status: "identical", + compatible: true, + changes: [], + warnings: [ + `Matched on advisory version '${input.baseline.version}'; fingerprints differ (${baselineFingerprint} vs ${candidateFingerprint}).`, + ], + errors, + shouldInvalidate: false, + }; + } + + if (!input.candidate) { + // Candidate metadata missing but an expected fingerprint was supplied. + if (candidateFingerprint === baselineFingerprint) { + return { + status: "identical", + compatible: true, + changes: [], + warnings, + errors, + shouldInvalidate: false, + }; + } + return { + status: "incompatible", + compatible: false, + changes: [], + warnings, + errors: [ + `Cached metadata fingerprint '${baselineFingerprint}' does not match expected fingerprint '${candidateFingerprint}'.`, + ], + shouldInvalidate: true, + }; + } + + const changes = diffMethods( + normalizeMethods(input.baseline.methods), + normalizeMethods(input.candidate.methods), + ); + + const breaking = changes.filter((c) => c.breaking); + if (breaking.length > 0) { + return { + status: "incompatible", + compatible: false, + changes, + warnings, + errors: breaking.map((c) => c.message), + shouldInvalidate: true, + }; + } + + if (changes.length > 0) { + return { + status: "compatible", + compatible: true, + changes, + warnings: changes.map((c) => c.message), + errors, + shouldInvalidate: false, + }; + } + + // Fingerprints differ but the diff found no structural change (e.g. only + // ordering or visibility nuances) — treat as compatible. + return { + status: "compatible", + compatible: true, + changes, + warnings: ["Fingerprints differ but no breaking interface change was detected."], + errors, + shouldInvalidate: false, + }; +} + +// ─── Stale metadata detection before invocation ─────────────────────────────── + +export interface StaleMetadataCheckInput { + contractId: string; + /** Cached metadata the application is about to use. */ + cachedMetadata?: ContractMethod[]; + /** Fingerprint of the expected/deployed interface. */ + expectedFingerprint?: string; + /** Freshly discovered metadata for the contract, when available. */ + freshMetadata?: ContractMethod[] | ContractSchema; +} + +export interface StaleMetadataCheckResult { + /** True when the cached metadata is current and safe to use. */ + current: boolean; + /** Human-readable explanation when stale. */ + message: string; + /** Compatibility report backing the decision. */ + report: ContractMetadataCompatibilityReport; +} + +/** + * Detect stale cached metadata before a contract invocation. + * + * When the caller supplies either an expected fingerprint or freshly discovered + * metadata, this returns a decision on whether the cached metadata is still + * valid. Missing cached metadata or missing reference metadata resolve to a + * non-fatal "missing" result (the invocation may still proceed). + * + * @param input - Contract identity plus cached and reference metadata. + * @returns An always-`ok` {@link StaleMetadataCheckResult}; use the `current` + * flag to decide whether to re-fetch before invoking. + */ +export function checkStaleContractMetadata( + input: StaleMetadataCheckInput, +): SorokitResult { + if (!input.cachedMetadata || input.cachedMetadata.length === 0) { + return ok({ + current: false, + message: "No cached metadata available for this contract.", + report: { + status: "missing", + compatible: false, + changes: [], + warnings: ["Cached metadata is absent."], + errors: [], + shouldInvalidate: false, + }, + }); + } + + if ( + input.expectedFingerprint === undefined && + !input.freshMetadata + ) { + return ok({ + current: true, + message: "No reference metadata to compare against; assuming cached metadata is current.", + report: { + status: "missing", + compatible: true, + changes: [], + warnings: ["No reference fingerprint provided."], + errors: [], + shouldInvalidate: false, + }, + }); + } + + const baseline = buildContractMetadataSnapshot({ + contractId: input.contractId, + methods: input.cachedMetadata, + }); + + const report = checkContractMetadataCompatibility({ + baseline, + ...(input.freshMetadata + ? { + candidate: buildContractMetadataSnapshot({ + contractId: input.contractId, + methods: input.freshMetadata, + }), + } + : {}), + ...(input.expectedFingerprint !== undefined + ? { expectedFingerprint: input.expectedFingerprint } + : {}), + }); + + if (report.status === "identical") { + return ok({ + current: true, + message: "Cached metadata matches the deployed contract interface.", + report, + }); + } + + return ok({ + current: false, + message: + report.status === "missing" + ? "Cached metadata could not be verified against a reference." + : report.errors[0] ?? + "Cached metadata is incompatible with the deployed contract interface.", + report, + }); +} + +// ─── Migration hooks ────────────────────────────────────────────────────────── + +/** + * Apply migration hooks to incompatible metadata. + * + * Hooks run in order against a copy of the candidate snapshot. The first hook + * that returns a snapshot is considered the migration result. If no hook + * produces a migrated snapshot the metadata is left blocked (incompatible). + * + * @param candidate - The freshly discovered (newer) metadata. + * @param hooks - Application-provided migration hooks. + * @returns A {@link ContractMetadataMigrationResult}. + */ +export function applyContractMetadataMigration( + candidate: ContractMetadataSnapshot, + hooks: ContractMetadataMigrationHook[], +): ContractMetadataMigrationResult { + const migrations: ContractMetadataMigration[] = []; + + for (const hook of hooks) { + const result = hook(candidate); + if (result && result.fingerprint) { + migrations.push({ + applied: true, + description: `A migration hook produced a metadata snapshot with fingerprint '${result.fingerprint}'.`, + }); + return { snapshot: result, migrations, blocked: false }; + } + migrations.push({ applied: false, description: "A migration hook made no change." }); + } + + return { snapshot: null, migrations, blocked: true }; +} + +// ─── Cache invalidation integration ─────────────────────────────────────────── + +export interface InvalidateMetadataInput { + contractId: string; + /** In-memory metadata cache (from `contractMetadata`). */ + cache?: SorokitCache; + /** Invalidate the fallback memory state as well. */ + reset?: () => void; +} + +/** + * Invalidate cached metadata for a contract after an incompatible interface + * change is detected. Safe to call unconditionally. + */ +export function invalidateContractMetadataForIncompatibility( + input: InvalidateMetadataInput, +): void { + input.reset?.(); + input.cache?.invalidate(`sorokit:contract-metadata:${input.contractId}`); + input.cache?.invalidate(`sorokit:contract-schema:${input.contractId}`); +} + +// Re-export the standard invalidation helper so callers can keep a single +// import surface without relying on `contractMetadata` internals. +export { invalidateContractCache as invalidateCachedContractMetadata } from "./contractMetadata"; diff --git a/src/soroban/index.ts b/src/soroban/index.ts index a1aa7d2..58c9cde 100644 --- a/src/soroban/index.ts +++ b/src/soroban/index.ts @@ -76,6 +76,31 @@ export type { OnContractUpgrade, ContractVersionOptions, } from "./contractVersion"; +export { + computeContractMetadataFingerprint, + buildContractMetadataSnapshot, + checkContractMetadataCompatibility, + checkStaleContractMetadata, + applyContractMetadataMigration, + invalidateContractMetadataForIncompatibility, + invalidateCachedContractMetadata, +} from "./contractMetadataCompatibility"; +export type { + ContractMetadataVersion, + ContractMetadataSnapshot, + ContractMetadataChange, + ContractMetadataChangeKind, + ContractMetadataCompatibilityStatus, + ContractMetadataCompatibilityReport, + ContractMetadataMigration, + ContractMetadataMigrationHook, + ContractMetadataMigrationResult, + BuildMetadataSnapshotInput, + CheckCompatibilityInput, + StaleMetadataCheckInput, + StaleMetadataCheckResult, + InvalidateMetadataInput, +} from "./contractMetadataCompatibility"; export { decodeContractError, DEFAULT_CONTRACT_ERROR_MAP, diff --git a/src/tests/batchOperations.test.ts b/src/tests/batchOperations.test.ts new file mode 100644 index 0000000..8f72607 --- /dev/null +++ b/src/tests/batchOperations.test.ts @@ -0,0 +1,376 @@ +/** + * Tests for batch account operations (#514): + * bulkCreateTrustlines, bulkSendPayments, bulkRotateKeys, and the generic + * runBatchOperations executor. + * + * Covers: successful batches, partial failures (successes preserved), retries + * with idempotency awareness, concurrency limits, progress tracking, and + * concurrent-vs-sequential performance behavior. + */ + +import { describe, it, expect, vi, beforeEach } from "vitest"; +import { ok, err, SorokitErrorCode } from "../shared/response"; +import type { ResolvedNetworkConfig } from "../shared/types"; + +const mockBuildPaymentTransaction = vi.hoisted(() => + vi.fn(async () => ok("PAYMENT_XDR")), +); +const mockBuildTrustlineTransaction = vi.hoisted(() => + vi.fn(async () => ok("TRUSTLINE_XDR")), +); +const mockRotateAccountKey = vi.hoisted(() => + vi.fn(async () => ok("ROTATE_XDR")), +); + +vi.mock("../transaction/buildTransaction", () => ({ + buildPaymentTransaction: mockBuildPaymentTransaction, + buildTrustlineTransaction: mockBuildTrustlineTransaction, + buildBulkTrustlines: vi.fn(), + buildBulkTrustlineTransaction: vi.fn(), + buildPaymentWithTrustline: vi.fn(), +})); + +vi.mock("../account/keyRotation", () => ({ + rotateAccountKey: mockRotateAccountKey, + setAccountRecovery: vi.fn(), + recoverAccountKeys: vi.fn(), + isValidStellarPublicKey: vi.fn(), +})); + +import { + runBatchOperations, + bulkSendPayments, + bulkCreateTrustlines, + bulkRotateKeys, +} from "../account/batchOperations"; +import type { BatchOperation } from "../account/batchOperations"; + +const NETWORK_CONFIG: ResolvedNetworkConfig = { + network: "testnet", + horizonUrl: "https://horizon-testnet.stellar.org", + rpcUrl: "https://rpc-testnet.stellar.org", + networkPassphrase: "Test SDF Network ; September 2015", +}; + +const HORIZON_URL = NETWORK_CONFIG.horizonUrl; + +describe("runBatchOperations — core executor", () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + it("runs every operation and reports a successful summary", async () => { + const ops: BatchOperation[] = ["a", "b", "c"].map((id) => ({ + id, + runner: async () => ok(`done-${id}`), + })); + + const report = await runBatchOperations(ops, { concurrency: 2 }); + + expect(report.summary.total).toBe(3); + expect(report.summary.succeeded).toBe(3); + expect(report.summary.failed).toBe(0); + expect(report.summary.skipped).toBe(0); + expect(report.summary.allSucceeded).toBe(true); + expect(report.results.map((r) => r.status)).toEqual([ + "success", + "success", + "success", + ]); + expect(report.results.map((r) => r.data)).toEqual([ + "done-a", + "done-b", + "done-c", + ]); + }); + + it("preserves successes when some operations fail (partial failure)", async () => { + const ops: BatchOperation[] = [ + { id: "ok-1", runner: async () => ok("fine") }, + { + id: "bad-1", + runner: async () => + err(SorokitErrorCode.TX_SUBMIT_FAILED, "rejected"), + }, + { id: "ok-2", runner: async () => ok("fine-2") }, + { + id: "bad-2", + runner: async () => + err(SorokitErrorCode.TX_BUILD_FAILED, "bad xdr"), + }, + ]; + + const report = await runBatchOperations(ops, { concurrency: 4 }); + + expect(report.summary.succeeded).toBe(2); + expect(report.summary.failed).toBe(2); + expect(report.summary.allSucceeded).toBe(false); + const byId = new Map(report.results.map((r) => [r.id, r])); + expect(byId.get("ok-1")!.status).toBe("success"); + expect(byId.get("ok-2")!.status).toBe("success"); + expect(byId.get("bad-1")!.status).toBe("failed"); + expect(byId.get("bad-1")!.errorCode).toBe(SorokitErrorCode.TX_SUBMIT_FAILED); + expect(byId.get("bad-2")!.status).toBe("failed"); + }); + + it("retries retryable failures but not permanent ones", async () => { + // NETWORK_ERROR is retryable: succeeds on the 3rd attempt. + let networkAttempts = 0; + // TX_BUILD_FAILED is not retryable: only runs once. + let permanentAttempts = 0; + + const ops: BatchOperation[] = [ + { + id: "retryable", + runner: async () => { + networkAttempts += 1; + if (networkAttempts < 3) { + return err(SorokitErrorCode.NETWORK_ERROR, "flaky network"); + } + return ok("recovered"); + }, + }, + { + id: "permanent", + runner: async () => { + permanentAttempts += 1; + return err(SorokitErrorCode.TX_BUILD_FAILED, "bad xdr"); + }, + }, + ]; + + const report = await runBatchOperations(ops, { + maxRetries: 3, + retryDelayMs: 0, + delay: async () => {}, + }); + + const byId = new Map(report.results.map((r) => [r.id, r])); + expect(byId.get("retryable")!.status).toBe("success"); + expect(byId.get("retryable")!.attempts).toBe(3); + expect(byId.get("retryable")!.retries).toBe(2); + expect(byId.get("permanent")!.status).toBe("failed"); + expect(byId.get("permanent")!.attempts).toBe(1); + expect(permanentAttempts).toBe(1); + }); + + it("never re-runs operations already completed (idempotency-aware)", async () => { + const executed = vi.fn(); + const ops: BatchOperation[] = [ + { id: "known", runner: async () => { executed(); return ok("x"); } }, + { id: "unknown", runner: async () => { executed(); return ok("y"); } }, + ]; + + const report = await runBatchOperations(ops, { + previouslyCompletedIds: ["known"], + }); + + expect(executed).toHaveBeenCalledTimes(1); // only "unknown" executed + const byId = new Map(report.results.map((r) => [r.id, r])); + expect(byId.get("known")!.status).toBe("skipped"); + expect(byId.get("known")!.skipped).toBe(true); + expect(byId.get("known")!.attempts).toBe(0); + expect(byId.get("unknown")!.status).toBe("success"); + }); + + it("does not duplicate a successfully submitted op across retries", async () => { + // Simulates an op whose FIRST attempt succeeds at the network layer but + // the response is lost (idempotency hazard). The executor must not re-run it. + let calls = 0; + const ops: BatchOperation[] = [ + { + id: "already-sent", + runner: async () => { + calls += 1; + if (calls === 1) return ok("submitted"); + return err(SorokitErrorCode.TX_SUBMIT_FAILED, "would duplicate"); + }, + }, + ]; + // previouslyCompletedIds marks the op as already handled -> skipped, not re-run. + const report = await runBatchOperations(ops, { + previouslyCompletedIds: ["already-sent"], + }); + expect(calls).toBe(0); + expect(report.results[0].status).toBe("skipped"); + }); + + it("enforces the concurrency limit", async () => { + const concurrency = 3; + let active = 0; + let maxActive = 0; + const ops: BatchOperation[] = Array.from({ length: 10 }, (_, i) => ({ + id: `op-${i}`, + runner: async () => { + active += 1; + maxActive = Math.max(maxActive, active); + await new Promise((r) => setTimeout(r, 5)); + active -= 1; + return ok("x"); + }, + })); + + await runBatchOperations(ops, { concurrency }); + + expect(maxActive).toBeLessThanOrEqual(concurrency); + expect(maxActive).toBeGreaterThan(0); + }); + + it("reports progress through completion", async () => { + const snapshots: Array> = []; + const ops: BatchOperation[] = ["a", "b"].map((id) => ({ + id, + runner: async () => ok(`done-${id}`), + })); + + await runBatchOperations(ops, { + concurrency: 2, + onProgress: (p) => + snapshots.push({ + planned: p.planned, + running: p.running, + succeeded: p.succeeded, + failed: p.failed, + completed: p.completed, + total: p.total, + }), + }); + + // Early snapshot before any operation finishes must show planned but 0 completed. + const early = snapshots[0]; + expect(early.total).toBe(2); + expect(early.completed).toBe(0); + expect(early.planned).toBe(2); + // Final snapshot reflects all completed. + const last = snapshots[snapshots.length - 1]; + expect(last.succeeded).toBe(2); + expect(last.completed).toBe(2); + }); + + it("performs faster than sequential execution with many slow ops", async () => { + const opDelay = 10; + const count = 6; + const concurrency = 6; + const runFor = async (c: number) => { + const ops: BatchOperation[] = Array.from( + { length: count }, + (_, i) => ({ + id: `op-${i}`, + runner: async () => { + await new Promise((r) => setTimeout(r, opDelay)); + return ok("x"); + }, + }), + ); + const start = Date.now(); + await runBatchOperations(ops, { concurrency: c }); + return Date.now() - start; + }; + + const sequential = await runFor(1); // fully serial + const concurrent = await runFor(concurrency); // fully parallel + + // With all ops in parallel, elapsed should be ~1x opDelay, far below + // sequential (count * opDelay). Use a loose upper bound to stay deterministic. + expect(concurrent).toBeLessThan(opDelay * count); + }); +}); + +describe("runBatchOperations — bulk wrappers", () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + it("bulkSendPayments builds one transaction per payment without submit", async () => { + mockBuildPaymentTransaction.mockResolvedValue(ok("PAYMENT_XDR")); + const report = await bulkSendPayments({ + horizonUrl: HORIZON_URL, + networkConfig: NETWORK_CONFIG, + transactions: [ + { id: "p1", source: "GA", params: { destination: "GB", amount: "10" } }, + { id: "p2", source: "GC", params: { destination: "GD", amount: "5" } }, + ], + config: { concurrency: 2 }, + }); + + expect(mockBuildPaymentTransaction).toHaveBeenCalledTimes(2); + expect(report.summary.succeeded).toBe(2); + expect(report.summary.allSucceeded).toBe(true); + const byId = new Map(report.results.map((r) => [r.id, r])); + expect(byId.get("p1")!.data).toMatchObject({ source: "GA" }); + expect(byId.get("p1")!.data!.xdr).toBe("PAYMENT_XDR"); + }); + + it("bulkSendPayments invokes submit when provided", async () => { + const submit = vi.fn(async () => ok("submitted")); + await bulkSendPayments({ + horizonUrl: HORIZON_URL, + networkConfig: NETWORK_CONFIG, + transactions: [ + { id: "p1", source: "GA", params: { destination: "GB", amount: "10" } }, + ], + submit, + }); + expect(submit).toHaveBeenCalledWith("PAYMENT_XDR"); + }); + + it("bulkCreateTrustlines builds a trustline per account x asset", async () => { + mockBuildTrustlineTransaction.mockResolvedValue(ok("TRUSTLINE_XDR")); + const report = await bulkCreateTrustlines({ + horizonUrl: HORIZON_URL, + networkConfig: NETWORK_CONFIG, + accounts: ["GA", "GB"], + assets: [{ code: "USDC", issuer: "ISSUER" }], + config: { concurrency: 2 }, + }); + + expect(report.summary.total).toBe(2); // 2 accounts x 1 asset + expect(report.summary.succeeded).toBe(2); + expect(mockBuildTrustlineTransaction).toHaveBeenCalledTimes(2); + const results = report.results.filter((r) => r.status === "success"); + expect(results.map((r) => r.data!.assetCode)).toEqual(["USDC", "USDC"]); + }); + + it("bulkRotateKeys rotates keys for each account", async () => { + mockRotateAccountKey.mockResolvedValue(ok("ROTATE_XDR")); + const report = await bulkRotateKeys({ + horizonUrl: HORIZON_URL, + networkConfig: NETWORK_CONFIG, + accounts: [ + { id: "r1", account: "GA", oldKey: "OLD1", newKey: "NEW1" }, + { id: "r2", account: "GB", oldKey: "OLD2", newKey: "NEW2" }, + ], + config: { concurrency: 2 }, + }); + + expect(mockRotateAccountKey).toHaveBeenCalledTimes(2); + expect(report.summary.succeeded).toBe(2); + const byId = new Map(report.results.map((r) => [r.id, r])); + expect(byId.get("r1")!.data!.xdr).toBe("ROTATE_XDR"); + }); + + it("bulkSendPayments preserves successes when a build fails", async () => { + mockBuildPaymentTransaction + .mockResolvedValueOnce(ok("PAYMENT_XDR")) + .mockResolvedValueOnce( + err(SorokitErrorCode.TX_BUILD_FAILED, "invalid payment"), + ); + + const report = await bulkSendPayments({ + horizonUrl: HORIZON_URL, + networkConfig: NETWORK_CONFIG, + transactions: [ + { id: "p1", source: "GA", params: { destination: "GB", amount: "1" } }, + { id: "p2", source: "GC", params: { destination: "GD", amount: "2" } }, + ], + config: { concurrency: 2 }, + }); + + expect(report.summary.succeeded).toBe(1); + expect(report.summary.failed).toBe(1); + const byId = new Map(report.results.map((r) => [r.id, r])); + expect(byId.get("p1")!.status).toBe("success"); + expect(byId.get("p2")!.status).toBe("failed"); + expect(byId.get("p2")!.errorCode).toBe(SorokitErrorCode.TX_BUILD_FAILED); + }); +}); diff --git a/src/tests/contractMetadataCompatibility.test.ts b/src/tests/contractMetadataCompatibility.test.ts new file mode 100644 index 0000000..70e18b9 --- /dev/null +++ b/src/tests/contractMetadataCompatibility.test.ts @@ -0,0 +1,308 @@ +/** + * Tests for contract metadata versioning & compatibility validation (fix.md). + * + * Covers identical, compatible, incompatible, and missing metadata versions, + * deterministic fingerprints, stale-metadata detection, migration hooks, and + * stale-cache invalidation. + */ + +import { describe, it, expect } from "vitest"; +import type { SorokitCache } from "../shared/cache"; +import type { ContractMethod } from "../soroban/types"; +import type { ContractSchema } from "../soroban/contractMetadata"; +import { + computeContractMetadataFingerprint, + buildContractMetadataSnapshot, + checkContractMetadataCompatibility, + checkStaleContractMetadata, + applyContractMetadataMigration, + invalidateContractMetadataForIncompatibility, + type ContractMetadataSnapshot, +} from "../soroban/contractMetadataCompatibility"; + +const CONTRACT_ID = "CAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"; + +function methods(overrides: Array>): ContractMethod[] { + return overrides.map((m) => ({ + name: m.name ?? "default", + inputs: m.inputs ?? [{ name: "value", type: "u32" }], + returnType: m.returnType ?? "u32", + ...(m.visibility !== undefined ? { visibility: m.visibility } : {}), + })); +} + +const BASE_METHODS = methods([ + { name: "transfer" }, + { name: "balance", inputs: [{ name: "account", type: "address" }], returnType: "i128" }, +]); + +function snapshot(contractId = CONTRACT_ID, ms = BASE_METHODS, version?: string): ContractMetadataSnapshot { + return buildContractMetadataSnapshot({ contractId, methods: ms, ...(version ? { version } : {}) }); +} + +describe("computeContractMetadataFingerprint", () => { + it("is deterministic for identical interfaces", () => { + const a = computeContractMetadataFingerprint(BASE_METHODS); + const b = computeContractMetadataFingerprint(BASE_METHODS); + expect(a).toBe(b); + }); + + it("is identical regardless of method ordering", () => { + const reversed = methods([ + { name: "balance", inputs: [{ name: "account", type: "address" }], returnType: "i128" }, + { name: "transfer" }, + ]); + expect(computeContractMetadataFingerprint(reversed)).toBe( + computeContractMetadataFingerprint(BASE_METHODS), + ); + }); + + it("differs when a method signature changes", () => { + const changed = methods([ + { name: "transfer", inputs: [{ name: "value", type: "u64" }] }, + { name: "balance", inputs: [{ name: "account", type: "address" }], returnType: "i128" }, + ]); + expect(computeContractMetadataFingerprint(changed)).not.toBe( + computeContractMetadataFingerprint(BASE_METHODS), + ); + }); + + it("accepts a typed ContractSchema", () => { + const schema: ContractSchema = { + contractId: CONTRACT_ID, + methods: [ + { name: "transfer", params: [{ name: "value", type: "u32" }], returnType: "u32" }, + ], + }; + expect(computeContractMetadataFingerprint(schema)).toMatch(/^[0-9a-f]{8}$/); + }); + + it("does not rely on semantic version strings", () => { + // Two different versions with the same interface fingerprint match. + const v1 = snapshot(CONTRACT_ID, BASE_METHODS, "1.0.0"); + const v2 = snapshot(CONTRACT_ID, BASE_METHODS, "2.0.0"); + expect(v1.fingerprint).toBe(v2.fingerprint); + }); +}); + +describe("buildContractMetadataSnapshot (#2 — associated with contract id)", () => { + it("associates metadata with a contract identifier", () => { + const snap = buildContractMetadataSnapshot({ + contractId: CONTRACT_ID, + methods: BASE_METHODS, + version: "3.1.0", + }); + expect(snap.contractId).toBe(CONTRACT_ID); + expect(snap.version).toBe("3.1.0"); + expect(snap.fingerprint).toBe(computeContractMetadataFingerprint(BASE_METHODS)); + }); +}); + +describe("checkContractMetadataCompatibility", () => { + it("reports identical when fingerprints match (#identical)", () => { + const report = checkContractMetadataCompatibility({ + baseline: snapshot(), + candidate: snapshot(), + }); + expect(report.status).toBe("identical"); + expect(report.compatible).toBe(true); + expect(report.changes).toEqual([]); + expect(report.shouldInvalidate).toBe(false); + }); + + it("reports compatible for non-breaking changes (added method)", () => { + const added = methods([ + ...BASE_METHODS, + { name: "allowance" }, + ]); + const report = checkContractMetadataCompatibility({ + baseline: snapshot(), + candidate: snapshot(CONTRACT_ID, added), + }); + expect(report.status).toBe("compatible"); + expect(report.compatible).toBe(true); + expect(report.changes).toHaveLength(1); + expect(report.changes[0].kind).toBe("ADDED_METHOD"); + expect(report.changes[0].breaking).toBe(false); + expect(report.shouldInvalidate).toBe(false); + }); + + it("reports incompatible for breaking changes (removed method)", () => { + const removed = methods([{ name: "transfer" }]); + const report = checkContractMetadataCompatibility({ + baseline: snapshot(), + candidate: snapshot(CONTRACT_ID, removed), + }); + expect(report.status).toBe("incompatible"); + expect(report.compatible).toBe(false); + expect(report.errors.length).toBeGreaterThan(0); + expect(report.changes.some((c) => c.kind === "REMOVED_METHOD")).toBe(true); + expect(report.shouldInvalidate).toBe(true); + }); + + it("reports incompatible when an argument type changes", () => { + const changed = methods([ + { name: "transfer", inputs: [{ name: "value", type: "u64" }] }, + { name: "balance", inputs: [{ name: "account", type: "address" }], returnType: "i128" }, + ]); + const report = checkContractMetadataCompatibility({ + baseline: snapshot(), + candidate: snapshot(CONTRACT_ID, changed), + }); + expect(report.status).toBe("incompatible"); + expect(report.changes.some((c) => c.kind === "CHANGED_ARG_TYPE")).toBe(true); + }); + + it("reports missing when no candidate and no expected fingerprint is provided", () => { + const report = checkContractMetadataCompatibility({ + baseline: snapshot(), + }); + expect(report.status).toBe("missing"); + expect(report.warnings.length).toBeGreaterThan(0); + expect(report.shouldInvalidate).toBe(false); + }); + + it("reports identical against an expected fingerprint", () => { + const baseline = snapshot(); + const report = checkContractMetadataCompatibility({ + baseline, + expectedFingerprint: baseline.fingerprint, + }); + expect(report.status).toBe("identical"); + expect(report.compatible).toBe(true); + }); + + it("reports incompatible when cached fingerprint mismatches expected fingerprint", () => { + const baseline = snapshot(); + const report = checkContractMetadataCompatibility({ + baseline, + expectedFingerprint: computeContractMetadataFingerprint( + methods([{ name: "entirely_different" }]), + ), + }); + expect(report.status).toBe("incompatible"); + expect(report.shouldInvalidate).toBe(true); + }); + + it("treats matching advisory versions as identical even if fingerprints differ", () => { + const a = snapshot(CONTRACT_ID, BASE_METHODS, "1.0.0"); + const b = snapshot(CONTRACT_ID, methods([{ name: "transfer", inputs: [{ name: "x", type: "u64" }] }]), "1.0.0"); + const report = checkContractMetadataCompatibility({ + baseline: a, + candidate: b, + }); + expect(report.status).toBe("identical"); + expect(report.compatible).toBe(true); + }); +}); + +describe("checkStaleContractMetadata (#4 — detected before invocation)", () => { + it("reports current when cached metadata matches fresh metadata", async () => { + const result = await checkStaleContractMetadata({ + contractId: CONTRACT_ID, + cachedMetadata: BASE_METHODS, + freshMetadata: BASE_METHODS, + }); + expect(result.status).toBe("ok"); + if (result.status === "ok") { + expect(result.data.current).toBe(true); + expect(result.data.report.status).toBe("identical"); + } + }); + + it("reports stale when cached metadata no longer matches", async () => { + const changed = methods([ + { name: "transfer", inputs: [{ name: "value", type: "u128" }] }, + { name: "balance", inputs: [{ name: "account", type: "address" }], returnType: "i128" }, + ]); + const result = await checkStaleContractMetadata({ + contractId: CONTRACT_ID, + cachedMetadata: BASE_METHODS, + freshMetadata: changed, + }); + expect(result.status).toBe("ok"); + if (result.status === "ok") { + expect(result.data.current).toBe(false); + expect(result.data.report.status).toBe("incompatible"); + expect(result.data.message.length).toBeGreaterThan(0); + } + }); + + it("reports missing when no cached metadata exists", async () => { + const result = await checkStaleContractMetadata({ + contractId: CONTRACT_ID, + freshMetadata: BASE_METHODS, + }); + expect(result.status).toBe("ok"); + if (result.status === "ok") { + expect(result.data.current).toBe(false); + expect(result.data.report.status).toBe("missing"); + } + }); + + it("returns current (non-fatal) when no reference is available", async () => { + const result = await checkStaleContractMetadata({ + contractId: CONTRACT_ID, + cachedMetadata: BASE_METHODS, + }); + expect(result.status).toBe("ok"); + if (result.status === "ok") { + expect(result.data.current).toBe(true); + expect(result.data.report.status).toBe("missing"); + } + }); +}); + +describe("applyContractMetadataMigration (#6 — migration hooks)", () => { + it("applies the first hook that produces a snapshot", () => { + const candidate = snapshot(CONTRACT_ID, methods([{ name: "new_api" }]), "2.0.0"); + const migrated: ContractMetadataSnapshot = buildContractMetadataSnapshot({ + contractId: CONTRACT_ID, + methods: methods([{ name: "legacy_api" }]), + }); + const result = applyContractMetadataMigration(candidate, [ + () => undefined, // no change + () => migrated, + ]); + expect(result.blocked).toBe(false); + expect(result.snapshot).toBe(migrated); + expect(result.migrations).toHaveLength(2); + expect(result.migrations[0].applied).toBe(false); + expect(result.migrations[1].applied).toBe(true); + }); + + it("blocks when no hook can migrate the metadata", () => { + const candidate = snapshot(CONTRACT_ID, methods([{ name: "new_api" }])); + const result = applyContractMetadataMigration(candidate, [ + () => undefined, + ]); + expect(result.blocked).toBe(true); + expect(result.snapshot).toBeNull(); + }); +}); + +describe("invalidateContractMetadataForIncompatibility (#7)", () => { + it("invalidates both metadata and schema caches for the contract", () => { + const invalidated: string[] = []; + const cache: SorokitCache = { + get: () => undefined, + set: () => undefined, + invalidate: (key) => { + invalidated.push(key); + }, + }; + let resetCalled = false; + + invalidateContractMetadataForIncompatibility({ + contractId: CONTRACT_ID, + cache, + reset: () => { + resetCalled = true; + }, + }); + + expect(resetCalled).toBe(true); + expect(invalidated).toContain(`sorokit:contract-metadata:${CONTRACT_ID}`); + expect(invalidated).toContain(`sorokit:contract-schema:${CONTRACT_ID}`); + }); +}); diff --git a/src/tests/paymentNotifications.test.ts b/src/tests/paymentNotifications.test.ts new file mode 100644 index 0000000..5cf1ef7 --- /dev/null +++ b/src/tests/paymentNotifications.test.ts @@ -0,0 +1,427 @@ +/** + * Tests for the payment notification subsystem (fix.md). + * + * Covers successful delivery, retries (exponential backoff), configurable + * timeouts, duplicate-delivery identification through event IDs, failed + * endpoints surfacing errors, and decoupled notification channels. + */ + +import { describe, it, expect, beforeEach, afterEach, vi } from "vitest"; +import { + registerPaymentWebhook, + unregisterPaymentWebhook, + listPaymentWebhooks, + clearPaymentWebhooks, + triggerPaymentNotifications, + dispatchPaymentNotification, + generatePaymentEventId, + isPaymentNotificationEvent, + PAYMENT_NOTIFICATION_EVENTS, + type PaymentNotificationChannel, + type PaymentNotificationEvent, +} from "../transaction/paymentNotifications"; + +const SECRET_URL = "https://pay.example.com/webhook"; + +function okResponse(): Response { + return { ok: true, status: 200, statusText: "OK" } as Response; +} + +describe("paymentNotifications", () => { + beforeEach(() => { + clearPaymentWebhooks(); + }); + + afterEach(() => { + clearPaymentWebhooks(); + vi.restoreAllMocks(); + vi.clearAllMocks(); + }); + + describe("event types", () => { + it("defines the supported payment events", () => { + expect(PAYMENT_NOTIFICATION_EVENTS).toEqual([ + "payment_received", + "payment_sent", + "payment_failed", + "payment_confirmed", + ]); + }); + + it("guards payment event types", () => { + expect(isPaymentNotificationEvent("payment_received")).toBe(true); + expect(isPaymentNotificationEvent("payment_confirmed")).toBe(true); + expect(isPaymentNotificationEvent("tx_confirmed")).toBe(false); + expect(isPaymentNotificationEvent("payment_failed")).toBe(true); + }); + }); + + describe("registerPaymentWebhook", () => { + it("registers a URL for multiple payment events", () => { + const result = registerPaymentWebhook(SECRET_URL, [ + "payment_received", + "payment_sent", + ]); + expect(result.status).toBe("ok"); + expect(listPaymentWebhooks("payment_received")).toHaveLength(1); + expect(listPaymentWebhooks("payment_sent")).toHaveLength(1); + expect(listPaymentWebhooks("payment_failed")).toHaveLength(0); + }); + + it("rejects an empty events array", () => { + const result = registerPaymentWebhook(SECRET_URL, []); + expect(result.status).toBe("error"); + expect(result.error?.code).toBe("INVALID_CONFIG"); + }); + + it("rejects an invalid event type", () => { + const result = registerPaymentWebhook(SECRET_URL, [ + "payment_invalid" as PaymentNotificationEvent, + ]); + expect(result.status).toBe("error"); + expect(result.error?.code).toBe("INVALID_CONFIG"); + expect(listPaymentWebhooks("payment_received")).toHaveLength(0); + }); + + it("rejects an invalid URL", () => { + const result = registerPaymentWebhook("not-a-url", ["payment_received"]); + expect(result.status).toBe("error"); + expect(result.error?.code).toBe("INVALID_CONFIG"); + }); + + it("rejects an invalid maxRetries", () => { + const result = registerPaymentWebhook(SECRET_URL, ["payment_received"], { + maxRetries: -1, + }); + expect(result.status).toBe("error"); + expect(result.error?.code).toBe("INVALID_CONFIG"); + }); + + it("rejects an invalid timeoutMs", () => { + const result = registerPaymentWebhook(SECRET_URL, ["payment_received"], { + timeoutMs: 0, + }); + expect(result.status).toBe("error"); + expect(result.error?.code).toBe("INVALID_CONFIG"); + }); + + it("does not duplicate registrations for repeated events", () => { + const result = registerPaymentWebhook(SECRET_URL, [ + "payment_received", + "payment_received", + ]); + expect(result.status).toBe("ok"); + expect(listPaymentWebhooks("payment_received")).toHaveLength(1); + }); + + it("unregisters a specific event", () => { + registerPaymentWebhook(SECRET_URL, ["payment_received", "payment_sent"]); + const result = unregisterPaymentWebhook(SECRET_URL, "payment_received"); + expect(result.status).toBe("ok"); + expect(listPaymentWebhooks("payment_received")).toHaveLength(0); + expect(listPaymentWebhooks("payment_sent")).toHaveLength(1); + }); + + it("fails to unregister a nonexistent webhook", () => { + const result = unregisterPaymentWebhook(SECRET_URL, "payment_received"); + expect(result.status).toBe("error"); + expect(result.error?.code).toBe("INVALID_CONFIG"); + }); + }); + + describe("event IDs (idempotency)", () => { + it("derives the same event ID for the same payment identity", () => { + const a = generatePaymentEventId("tx-1", "payment_received", "GA001"); + const b = generatePaymentEventId("tx-1", "payment_received", "GA001"); + expect(a).toBe(b); + }); + + it("derives distinct event IDs for distinct payments", () => { + const a = generatePaymentEventId("tx-1", "payment_received", "GA001"); + const b = generatePaymentEventId("tx-2", "payment_received", "GA001"); + const c = generatePaymentEventId("tx-1", "payment_sent", "GA001"); + expect(a).not.toBe(b); + expect(a).not.toBe(c); + }); + + it("delivers identical event ID on duplicate deliveries", async () => { + global.fetch = vi.fn(() => Promise.resolve(okResponse())); + registerPaymentWebhook(SECRET_URL, ["payment_received"]); + + const input = { + transactionId: "tx-dup", + account: "GA001", + }; + await triggerPaymentNotifications("payment_received", input); + await triggerPaymentNotifications("payment_received", input); + + const fetchMock = global.fetch as ReturnType; + const firstBody = JSON.parse(fetchMock.mock.calls[0][1].body); + const secondBody = JSON.parse(fetchMock.mock.calls[1][1].body); + expect(firstBody.eventId).toBe(secondBody.eventId); + expect(firstBody.eventId).toBe( + generatePaymentEventId("tx-dup", "payment_received", "GA001"), + ); + }); + + it("sends the event ID as an idempotency header", async () => { + global.fetch = vi.fn(() => Promise.resolve(okResponse())); + registerPaymentWebhook(SECRET_URL, ["payment_received"]); + + await triggerPaymentNotifications("payment_received", { + transactionId: "tx-h", + account: "GA001", + }); + + const [, init] = (global.fetch as ReturnType) + .mock.calls[0] as [string, { headers: Record; body: string }]; + const payload = JSON.parse(init.body); + expect(init.headers["Idempotency-Key"]).toBe(payload.eventId); + expect(init.headers["X-Sorokit-Event-Id"]).toBe(payload.eventId); + }); + }); + + describe("payload shape", () => { + it("includes event ID, timestamp, transaction ID, account, and event type", async () => { + const fetchMock = vi.fn(() => Promise.resolve(okResponse())); + global.fetch = fetchMock as typeof fetch; + registerPaymentWebhook(SECRET_URL, ["payment_confirmed"]); + + const results = await triggerPaymentNotifications("payment_confirmed", { + transactionId: "tx-payload", + account: "GA100", + }); + expect(results).toHaveLength(1); + expect(results[0].status).toBe("ok"); + + const [, init] = fetchMock.mock.calls[0] as [string, { body: string }]; + const payload = JSON.parse(init.body); + expect(typeof payload.eventId).toBe("string"); + expect(payload.event).toBe("payment_confirmed"); + expect(typeof payload.timestamp).toBe("string"); + expect(new Date(payload.timestamp).toString()).not.toBe("Invalid Date"); + expect(payload.transactionId).toBe("tx-payload"); + expect(payload.account).toBe("GA100"); + }); + }); + + describe("delivery", () => { + it("successfully delivers to a healthy endpoint", async () => { + const fetchMock = vi.fn(() => Promise.resolve(okResponse())); + global.fetch = fetchMock as typeof fetch; + registerPaymentWebhook(SECRET_URL, ["payment_received"]); + + const results = await triggerPaymentNotifications("payment_received", { + transactionId: "tx-ok", + account: "GA001", + }); + expect(results).toHaveLength(1); + expect(results[0].status).toBe("ok"); + expect(fetchMock).toHaveBeenCalledTimes(1); + }); + + it("does not deliver to endpoints not subscribed to the event", async () => { + const fetchMock = vi.fn(() => Promise.resolve(okResponse())); + global.fetch = fetchMock as typeof fetch; + registerPaymentWebhook(SECRET_URL, ["payment_received"]); + + const results = await triggerPaymentNotifications("payment_sent", { + transactionId: "tx-no", + account: "GA001", + }); + expect(results).toHaveLength(0); + expect(fetchMock).not.toHaveBeenCalled(); + }); + }); + + describe("retries (exponential backoff)", () => { + it("retries a transient failure then succeeds", async () => { + let attempts = 0; + global.fetch = vi.fn(() => { + attempts++; + if (attempts < 3) return Promise.reject(new Error("flaky")); + return Promise.resolve(okResponse()); + }); + registerPaymentWebhook(SECRET_URL, ["payment_received"], { + maxRetries: 3, + }); + + const results = await triggerPaymentNotifications("payment_received", { + transactionId: "tx-retry", + account: "GA001", + }); + expect(results[0].status).toBe("ok"); + expect(attempts).toBe(3); + }); + + it("gives up after exhausting configured attempts", async () => { + global.fetch = vi.fn(() => Promise.reject(new Error("down"))); + registerPaymentWebhook(SECRET_URL, ["payment_failed"], { + maxRetries: 2, + }); + + const results = await triggerPaymentNotifications("payment_failed", { + transactionId: "tx-fail", + account: "GA001", + }); + expect(results[0].status).toBe("error"); + expect(results[0].error?.code).toBe("NETWORK_ERROR"); + expect((global.fetch as ReturnType)).toHaveBeenCalledTimes(3); // 1 + 2 retries + }, 20000); + }); + + describe("timeout", () => { + it("treats a slow endpoint as a failed attempt", async () => { + const fetchMock = vi.fn( + () => + new Promise((_resolve, reject) => { + setTimeout(() => reject(new Error("timed out")), 50); + }), + ); + global.fetch = fetchMock as typeof fetch; + registerPaymentWebhook(SECRET_URL, ["payment_received"], { + maxRetries: 0, + timeoutMs: 5, + }); + + const results = await triggerPaymentNotifications("payment_received", { + transactionId: "tx-timeout", + account: "GA001", + }); + expect(results[0].status).toBe("error"); + expect((global.fetch as ReturnType)).toHaveBeenCalledTimes(1); + }, 20000); + }); + + describe("failed endpoints", () => { + it("surfaces an HTTP error for a failing endpoint", async () => { + global.fetch = vi.fn(() => + Promise.resolve({ ok: false, status: 500, statusText: "Internal" } as Response), + ); + registerPaymentWebhook(SECRET_URL, ["payment_received"], { + maxRetries: 0, + }); + + const results = await triggerPaymentNotifications("payment_received", { + transactionId: "tx-500", + account: "GA001", + }); + expect(results[0].status).toBe("error"); + expect(results[0].error?.code).toBe("NETWORK_ERROR"); + }); + + it("reports one failing endpoint without masking another", async () => { + const failingUrl = "https://fail.example.com/webhook"; + const goodUrl = "https://good.example.com/webhook"; + global.fetch = vi.fn((url: string) => + Promise.resolve( + url === failingUrl + ? ({ ok: false, status: 500 } as Response) + : okResponse(), + ), + ); + registerPaymentWebhook(failingUrl, ["payment_received"], { + maxRetries: 0, + }); + registerPaymentWebhook(goodUrl, ["payment_received"], { + maxRetries: 0, + }); + + const results = await triggerPaymentNotifications("payment_received", { + transactionId: "tx-two", + account: "GA001", + }); + expect(results).toHaveLength(2); + expect(results.some((r) => r.status === "error")).toBe(true); + expect(results.some((r) => r.status === "ok")).toBe(true); + }); + }); + + describe("notification channels (decoupled)", () => { + it("delivers to registered channels alongside webhooks", async () => { + global.fetch = vi.fn(() => Promise.resolve(okResponse())); + const channel: PaymentNotificationChannel = { + name: "test-channel", + deliver: vi.fn(() => Promise.resolve(true)), + }; + registerPaymentWebhook(SECRET_URL, ["payment_received"], { + channels: [channel], + }); + + const results = await triggerPaymentNotifications("payment_received", { + transactionId: "tx-channel", + account: "GA001", + }); + expect(results).toHaveLength(2); + expect(results.every((r) => r.status === "ok")).toBe(true); + expect(channel.deliver).toHaveBeenCalledTimes(1); + }); + + it("surfaces channel failures without coupling to webhook logic", async () => { + global.fetch = vi.fn(() => Promise.resolve(okResponse())); + const channel: PaymentNotificationChannel = { + name: "failing-channel", + deliver: vi.fn(() => Promise.resolve(false)), + }; + registerPaymentWebhook(SECRET_URL, ["payment_received"], { + channels: [channel], + }); + + const results = await triggerPaymentNotifications("payment_received", { + transactionId: "tx-chanfail", + account: "GA001", + }); + expect(results).toHaveLength(2); + expect(results[0].status).toBe("ok"); // webhook ok + expect(results[1].status).toBe("error"); // channel rejected + expect(results[1].error?.code).toBe("NETWORK_ERROR"); + }); + }); + + describe("dispatchPaymentNotification", () => { + it("returns immediately without blocking on delivery", async () => { + let resolveFetch: (value: Response) => void = () => undefined; + global.fetch = vi.fn( + () => + new Promise((resolve) => { + resolveFetch = resolve; + }), + ); + registerPaymentWebhook(SECRET_URL, ["payment_received"]); + + const before = Date.now(); + dispatchPaymentNotification("payment_received", { + transactionId: "tx-fast", + account: "GA001", + }); + const elapsed = Date.now() - before; + expect(elapsed).toBeLessThan(100); + resolveFetch(okResponse()); + }); + + it("swallows delivery failures without unhandled rejections", async () => { + global.fetch = vi.fn(() => Promise.reject(new Error("down"))); + registerPaymentWebhook(SECRET_URL, ["payment_received"], { + maxRetries: 0, + }); + + expect(() => + dispatchPaymentNotification("payment_received", { + transactionId: "tx-swallow", + account: "GA001", + }), + ).not.toThrow(); + await new Promise((r) => setTimeout(r, 10)); + }, 20000); + + it("is a no-op when nothing is subscribed", () => { + const fetchMock = vi.fn(); + global.fetch = fetchMock as typeof fetch; + dispatchPaymentNotification("payment_received", { + transactionId: "tx-noop", + account: "GA001", + }); + expect(fetchMock).not.toHaveBeenCalled(); + }); + }); +}); diff --git a/src/transaction/index.ts b/src/transaction/index.ts index fbfcfce..f68ac5c 100644 --- a/src/transaction/index.ts +++ b/src/transaction/index.ts @@ -291,6 +291,27 @@ export { dispatchTransactionEvent, verifySignature, } from "./webhooks"; + +// ─── Payment notifications (#fix.md) ────────────────────────────────────────── +export { + registerPaymentWebhook, + unregisterPaymentWebhook, + listPaymentWebhooks, + clearPaymentWebhooks, + triggerPaymentNotifications, + dispatchPaymentNotification, + generatePaymentEventId, + isPaymentNotificationEvent, + PAYMENT_NOTIFICATION_EVENTS, +} from "./paymentNotifications"; +export type { + PaymentNotificationEvent, + PaymentWebhookOptions, + PaymentWebhookPayload, + PaymentWebhookRegistration, + PaymentNotificationInput, + PaymentNotificationChannel, +} from "./paymentNotifications"; export type { WebhookEventType, TransactionWebhookEvent, diff --git a/src/transaction/paymentNotifications.ts b/src/transaction/paymentNotifications.ts new file mode 100644 index 0000000..c0c5789 --- /dev/null +++ b/src/transaction/paymentNotifications.ts @@ -0,0 +1,488 @@ +/** + * Payment notification subsystem. + * + * Lets applications subscribe a webhook endpoint (or an optional, decoupled + * notification channel) to payment lifecycle events such as payment_received, + * payment_sent, payment_failed, and payment_confirmed. + * + * Design goals (see fix.md): + * - Transport-agnostic delivery: the payload contract and dispatch logic do + * not depend on any particular HTTP client or channel implementation. + * - Non-blocking: dispatch never blocks (or throws into) the transaction + * processing path. + * - Bounded retries: exponential backoff with a hard cap and a configurable + * attempt limit so retry behavior cannot become an uncontrolled loop. + * - Idempotency: every payload carries a deterministic event ID derived from + * the payment identity so consumers can safely discard duplicate + * deliveries. + * - Clean error surfacing: registration and delivery failures are returned as + * structured `SorokitResult` errors. + */ + +import { ok, err, SorokitErrorCode } from "../shared/response"; +import type { SorokitResult } from "../shared/response"; +import { sleep } from "../shared/utils"; + +/** + * Canonical payment lifecycle event types. + */ +export type PaymentNotificationEvent = + | "payment_received" + | "payment_sent" + | "payment_failed" + | "payment_confirmed"; + +/** + * The full list of supported payment event types. + */ +export const PAYMENT_NOTIFICATION_EVENTS: readonly PaymentNotificationEvent[] = [ + "payment_received", + "payment_sent", + "payment_failed", + "payment_confirmed", +]; + +/** Type guard for {@link PaymentNotificationEvent}. */ +export function isPaymentNotificationEvent( + value: unknown, +): value is PaymentNotificationEvent { + return ( + typeof value === "string" && + (PAYMENT_NOTIFICATION_EVENTS as readonly string[]).includes(value) + ); +} + +/** + * Delivery options for a single webhook subscription. + */ +export interface PaymentWebhookOptions { + /** Maximum number of retry attempts after the initial request (default: 3). */ + maxRetries?: number; + /** Per-attempt request timeout in milliseconds (default: 10_000). */ + timeoutMs?: number; + /** + * Optional list of notification channels to deliver to in addition to the + * HTTP webhook. Channels are an independently decoupled abstraction so + * additional transports (Slack, SMS, email, …) can be plugged in without + * coupling them to the webhook delivery logic. + */ + channels?: PaymentNotificationChannel[]; +} + +const DEFAULT_MAX_RETRIES = 3; +const DEFAULT_TIMEOUT_MS = 10_000; +const DEFAULT_BACKOFF_MS = 1000; +const MAX_BACKOFF_MS = 8000; + +/** + * Webhook payload delivered to a subscriber. + * + * Standardized so consumers can safely process duplicate deliveries: the + * `eventId` is deterministic for a given (account, event, transactionId), so + * redelivery of the same logical event is always identifiable. + */ +export interface PaymentWebhookPayload { + /** Deterministic event identifier for idempotency / duplicate detection. */ + eventId: string; + /** The payment lifecycle event that occurred. */ + event: PaymentNotificationEvent; + /** ISO-8601 timestamp of when the event was emitted. */ + timestamp: string; + /** The transaction this payment event relates to. */ + transactionId: string; + /** The account associated with the payment. */ + account: string; +} + +/** Input describing an emitted payment event. */ +export interface PaymentNotificationInput { + /** Transaction identifier the payment belongs to. */ + transactionId: string; + /** Account associated with the payment. */ + account: string; + /** + * Optional explicit event ID. When omitted, one is derived deterministically + * from the event identity so duplicate deliveries share the same ID. + */ + eventId?: string; +} + +/** + * A registered webhook subscription. + */ +export interface PaymentWebhookRegistration { + /** URL to deliver webhook payloads to. */ + url: string; + /** Payment events this subscription is interested in. */ + events: PaymentNotificationEvent[]; + /** Resolved delivery options. */ + options: Required>; +} + +/** + * Decoupled notification channel abstraction. + * + * Applications may register channels to receive the same payload as webhook + * subscribers. Implementing this interface is entirely independent of the + * webhook delivery machinery, satisfying the "without coupling them to + * webhook logic" requirement. + */ +export interface PaymentNotificationChannel { + /** Human-readable channel name, used in error messages. */ + readonly name: string; + /** + * Deliver a payload through this channel. + * Return `true` on success. Returning `false` (or throwing) counts as a + * failed attempt and is retried like a failed webhook delivery. + */ + deliver(payload: PaymentWebhookPayload): Promise; +} + +const webhookRegistry = new Map(); +const channelRegistry = new Set(); + +/** Registry key — one entry per (url, event) pair. */ +function registrationKey( + url: string, + event: PaymentNotificationEvent, +): string { + return `${url}::${event}`; +} + +function validateUrl(url: string): SorokitResult | undefined { + if (typeof url !== "string" || url.length === 0) { + return err( + SorokitErrorCode.INVALID_CONFIG, + "Webhook URL must be a non-empty string", + ); + } + try { + new URL(url); + } catch { + return err(SorokitErrorCode.INVALID_CONFIG, `Invalid webhook URL: ${url}`); + } + return undefined; +} + +function isNonNegativeInteger(value: number, name: string): boolean { + return Number.isInteger(value) && value >= 0; +} + +function validateOptions( + options: PaymentWebhookOptions | undefined, +): SorokitResult>> | undefined { + if (options === undefined) { + return ok({ + maxRetries: DEFAULT_MAX_RETRIES, + timeoutMs: DEFAULT_TIMEOUT_MS, + }); + } + const maxRetries = options.maxRetries ?? DEFAULT_MAX_RETRIES; + const timeoutMs = options.timeoutMs ?? DEFAULT_TIMEOUT_MS; + if (!isNonNegativeInteger(maxRetries, "maxRetries")) { + return err( + SorokitErrorCode.INVALID_CONFIG, + "maxRetries must be a non-negative integer", + ); + } + if (!isNonNegativeInteger(timeoutMs, "timeoutMs") || timeoutMs === 0) { + return err( + SorokitErrorCode.INVALID_CONFIG, + "timeoutMs must be a positive integer", + ); + } + return ok({ maxRetries, timeoutMs }); +} + +/** + * Register a webhook (and optional notification channels) for one or more + * payment events. + * + * @param url - URL to deliver webhook payloads to. + * @param events - Payment event types to subscribe to. + * @param options - Delivery options (retries, timeout, extra channels). + * @returns ok(void) on success, or a structured error on invalid input. + * + * @example + * const result = registerPaymentWebhook( + * "https://example.com/payments", + * ["payment_received", "payment_failed"], + * { maxRetries: 5, timeoutMs: 5000 }, + * ); + */ +export function registerPaymentWebhook( + url: string, + events: PaymentNotificationEvent[], + options?: PaymentWebhookOptions, +): SorokitResult { + if (!Array.isArray(events) || events.length === 0) { + return err( + SorokitErrorCode.INVALID_CONFIG, + "At least one payment event type must be provided", + ); + } + + const normalized: PaymentNotificationEvent[] = []; + for (const event of events) { + if (!isPaymentNotificationEvent(event)) { + return err( + SorokitErrorCode.INVALID_CONFIG, + `Invalid payment event type: ${String(event)}. Must be one of: ${PAYMENT_NOTIFICATION_EVENTS.join(", ")}`, + ); + } + normalized.push(event); + } + + const urlError = validateUrl(url); + if (urlError) return urlError; + + const optionsResult = validateOptions(options); + if (optionsResult && optionsResult.status === "error") return optionsResult; + const resolvedOptions = optionsResult!.data; + + for (const channel of options?.channels ?? []) { + if (!channel || typeof channel.deliver !== "function") { + return err( + SorokitErrorCode.INVALID_CONFIG, + "Notification channels must implement deliver(payload)", + ); + } + channelRegistry.add(channel); + } + + const registration: PaymentWebhookRegistration = { + url, + events: normalized, + options: resolvedOptions, + }; + + for (const event of normalized) { + webhookRegistry.set(registrationKey(url, event), registration); + } + + return ok(undefined); +} + +/** + * Remove a webhook subscription for a specific event. + */ +export function unregisterPaymentWebhook( + url: string, + event: PaymentNotificationEvent, +): SorokitResult { + if (!isPaymentNotificationEvent(event)) { + return err( + SorokitErrorCode.INVALID_CONFIG, + `Invalid payment event type: ${String(event)}`, + ); + } + const key = registrationKey(url, event); + if (!webhookRegistry.has(key)) { + return err( + SorokitErrorCode.INVALID_CONFIG, + `No webhook registered for event '${event}' at '${url}'`, + ); + } + webhookRegistry.delete(key); + return ok(undefined); +} + +/** + * List all webhook subscriptions for a payment event type. + */ +export function listPaymentWebhooks( + event: PaymentNotificationEvent, +): PaymentWebhookRegistration[] { + if (!isPaymentNotificationEvent(event)) return []; + const results: PaymentWebhookRegistration[] = []; + for (const [key, registration] of webhookRegistry.entries()) { + if (key.endsWith(`::${event}`) && !results.includes(registration)) { + results.push(registration); + } + } + return results; +} + +/** Clear all registered webhook subscriptions. */ +export function clearPaymentWebhooks(): void { + webhookRegistry.clear(); + channelRegistry.clear(); +} + +/** + * Deterministically derive an event ID from the payment identity so duplicate + * deliveries of the same logical event carry the same identifier. + * + * FNV-1a (32-bit) over `transactionId | event | account`, hex-encoded. + */ +export function generatePaymentEventId( + transactionId: string, + event: PaymentNotificationEvent, + account: string, +): string { + const data = `${transactionId}|${event}|${account}`; + let hash = 0x811c9dc5; + for (let i = 0; i < data.length; i++) { + hash ^= data.charCodeAt(i); + hash = (hash * 0x01000193) >>> 0; + } + return hash.toString(16).padStart(8, "0"); +} + +/** Resolve the event ID, applying caller-supplied override or the derived one. */ +function resolveEventId( + input: PaymentNotificationInput, + event: PaymentNotificationEvent, +): string { + return input.eventId ?? generatePaymentEventId(input.transactionId, event, input.account); +} + +/** + * Send a webhook payload with exponential backoff and a configurable retry + * limit and timeout. + * + * @param registration - The subscription to deliver to. + * @param payload - Payload to deliver. + * @returns ok(void) on success, or a structured error after the final attempt. + */ +async function sendPayloadWithRetry( + registration: PaymentWebhookRegistration, + payload: PaymentWebhookPayload, +): Promise> { + const { url } = registration; + const { maxRetries, timeoutMs } = registration.options; + let lastError: unknown; + + for (let attempt = 0; attempt <= maxRetries; attempt++) { + try { + const response = await fetch(url, { + method: "POST", + headers: { + "Content-Type": "application/json", + "X-Sorokit-Event": payload.event, + "X-Sorokit-Event-Id": payload.eventId, + "Idempotency-Key": payload.eventId, + }, + body: JSON.stringify(payload), + signal: AbortSignal.timeout(timeoutMs), + }); + + if (!response.ok) { + throw new Error(`HTTP ${response.status}: ${response.statusText}`); + } + + return ok(undefined); + } catch (error) { + lastError = error; + if (attempt < maxRetries) { + // Exponential backoff (1s, 2s, 4s, …) capped at MAX_BACKOFF_MS so + // retry behavior can never spin out of control. + const delayMs = Math.min( + DEFAULT_BACKOFF_MS * Math.pow(2, attempt), + MAX_BACKOFF_MS, + ); + await sleep(delayMs); + } + } + } + + return err( + SorokitErrorCode.NETWORK_ERROR, + `Payment webhook delivery failed after ${maxRetries + 1} attempts to ${url}: ${lastError instanceof Error ? lastError.message : String(lastError)}`, + lastError, + ); +} + +/** Deliver a payload through any registered notification channels. */ +async function deliverToChannels( + payload: PaymentWebhookPayload, +): Promise[]> { + return Promise.all( + Array.from(channelRegistry).map(async (channel) => { + try { + const delivered = await channel.deliver(payload); + if (delivered) return ok(undefined); + return err( + SorokitErrorCode.NETWORK_ERROR, + `Notification channel '${channel.name}' rejected payload for event '${payload.event}'`, + ); + } catch (error) { + return err( + SorokitErrorCode.NETWORK_ERROR, + `Notification channel '${channel.name}' failed for event '${payload.event}': ${error instanceof Error ? error.message : String(error)}`, + error, + ); + } + }), + ); +} + +/** + * Trigger delivery of a payment notification to all matching subscribers and + * wait for the results. + * + * Deliveries run concurrently and are reported independently so one failing + * endpoint cannot mask another. + * + * @param event - The payment event that occurred. + * @param input - Payment identity (transactionId, account, optional eventId). + * @returns One result per delivery target (ok or error). + */ +export async function triggerPaymentNotifications( + event: PaymentNotificationEvent, + input: PaymentNotificationInput, +): Promise[]> { + if (!isPaymentNotificationEvent(event)) { + return [ + err( + SorokitErrorCode.INVALID_CONFIG, + `Invalid payment event type: ${String(event)}`, + ), + ]; + } + if (typeof input.transactionId !== "string" || input.transactionId.length === 0) { + return [err(SorokitErrorCode.INVALID_CONFIG, "transactionId must be a non-empty string")]; + } + if (typeof input.account !== "string" || input.account.length === 0) { + return [err(SorokitErrorCode.INVALID_CONFIG, "account must be a non-empty string")]; + } + + const payload: PaymentWebhookPayload = { + eventId: resolveEventId(input, event), + event, + timestamp: new Date().toISOString(), + transactionId: input.transactionId, + account: input.account, + }; + + const registrations = listPaymentWebhooks(event); + const webhookResults = await Promise.all( + registrations.map((registration) => sendPayloadWithRetry(registration, payload)), + ); + const channelResults = await deliverToChannels(payload); + return [...webhookResults, ...channelResults]; +} + +/** + * Fire-and-forget dispatch of a payment notification. + * + * Never throws, never rejects, and never blocks transaction processing while + * subscribers are being notified. Individual delivery failures are reported by + * {@link triggerPaymentNotifications} and intentionally swallowed here. + * + * @param event - The payment event that occurred. + * @param input - Payment identity (transactionId, account, optional eventId). + */ +export function dispatchPaymentNotification( + event: PaymentNotificationEvent, + input: PaymentNotificationInput, +): void { + try { + void triggerPaymentNotifications(event, input).catch(() => { + // Swallow: notification delivery must never surface into the + // transaction/payment processing flow. + }); + } catch { + // Defensive: even synchronous failures must not propagate. + } +}