feat(sdk): add payment workflow utilities

src/approvalWorkflowSequencer.ts

@@ -0,0 +1,95 @@
+import { ApprovalTimeoutError } from "./errors.js";
+import { emitSdkEvent } from "./events.js";
+import type { ApprovalSessionResult, MultiSigPolicy } from "./types.js";
+
+export type NotificationAdapter = (signerPublicKey: string, txXdr: string) => void | Promise<void>;
+export type SignatureApplier = (
+  txXdr: string,
+  signatures: ReadonlyMap<string, string>,
+) => string;
+
+export interface ApprovalWorkflowOptions {
+  notifySigner?: NotificationAdapter;
+  applySignatures?: SignatureApplier;
+}
+
+export class ApprovalSession {
+  private readonly signatures = new Map<string, string>();
+  private readonly signerWeights = new Map<string, number>();
+  private readonly expiresAt: number;
+  private completed = false;
+  private timer: ReturnType<typeof setTimeout>;
+
+  constructor(
+    private readonly txXdr: string,
+    private readonly policy: MultiSigPolicy,
+    private readonly applySignatures: SignatureApplier,
+  ) {
+    for (const signer of policy.signers) {
+      this.signerWeights.set(signer.publicKey, signer.weight);
+    }
+    this.expiresAt = Date.now() + policy.timeoutMs;
+    this.timer = setTimeout(() => undefined, policy.timeoutMs);
+  }
+
+  submitSignature(signerPublicKey: string, signatureBase64: string): ApprovalSessionResult {
+    this.assertActive();
+    if (!this.signerWeights.has(signerPublicKey)) {
+      throw new Error(`Signer is not authorized: ${signerPublicKey}`);
+    }
+
+    this.signatures.set(signerPublicKey, signatureBase64);
+    emitSdkEvent("approvalReceived", { signerPublicKey });
+
+    if (this.weight >= this.policy.threshold) {
+      this.completed = true;
+      clearTimeout(this.timer);
+      emitSdkEvent("approvalWorkflowComplete", { signerCount: this.signatures.size });
+    }
+
+    return { complete: this.completed, weight: this.weight };
+  }
+
+  getSignedXdr(): string {
+    this.assertActive();
+    if (!this.completed) {
+      throw new Error("Approval threshold has not been reached");
+    }
+    return this.applySignatures(this.txXdr, this.signatures);
+  }
+
+  private get weight(): number {
+    let total = 0;
+    for (const publicKey of this.signatures.keys()) {
+      total += this.signerWeights.get(publicKey) ?? 0;
+    }
+    return total;
+  }
+
+  private assertActive(): void {
+    if (this.completed) return;
+    if (Date.now() > this.expiresAt) {
+      clearTimeout(this.timer);
+      throw new ApprovalTimeoutError(this.policy.timeoutMs);
+    }
+  }
+}
+
+export class ApprovalWorkflowSequencer {
+  constructor(private readonly options: ApprovalWorkflowOptions = {}) {}
+
+  initiate(txXdr: string, policy: MultiSigPolicy): ApprovalSession {
+    const session = new ApprovalSession(
+      txXdr,
+      policy,
+      this.options.applySignatures ?? ((xdr) => xdr),
+    );
+
+    for (const signer of policy.signers) {
+      emitSdkEvent("approvalRequested", { signerPublicKey: signer.publicKey });
+      void this.options.notifySigner?.(signer.publicKey, txXdr);
+    }
+
+    return session;
+  }
+}

src/cursorTracker.ts

@@ -8,6 +8,8 @@
 
 import type { CursorStore } from "./types.js";
 
+export type CursorPersistence = CursorStore;
+
 /**
  * In-memory cursor store suitable for session-scoped pagination.
  * Cursors are lost when the process exits.
@@ -35,6 +37,7 @@ export class InMemoryCursorStore implements CursorStore {
 
 /** Singleton in-memory store shared across the module. */
 let defaultStore: CursorStore = new InMemoryCursorStore();
+const syncCursorStore = new Map<string, string>();
 
 /**
  * Override the default cursor store.
@@ -44,13 +47,47 @@ export function setDefaultCursorStore(store: CursorStore): void {
   defaultStore = store;
 }
 
+/** Alias retained for public API compatibility. */
+export function configureCursorStore(store: CursorStore): void {
+  setDefaultCursorStore(store);
+}
+
 /**
  * Get the current default cursor store.
  */
 export function getDefaultCursorStore(): CursorStore {
   return defaultStore;
 }
 
+/** Persist a cursor in the default in-memory stream cursor cache. */
+export function setCursor(key: string, cursor: string): void {
+  syncCursorStore.set(key, cursor);
+  void defaultStore.save(key, cursor);
+}
+
+/** Read a cursor from the default in-memory stream cursor cache. */
+export function getCursor(key: string): string | null {
+  return syncCursorStore.get(key) ?? null;
+}
+
+/** Remove a persisted cursor from the default stream cursor cache. */
+export function removeCursor(key: string): void {
+  syncCursorStore.delete(key);
+  void defaultStore.delete(key);
+}
+
+/** Persist a cursor from a snapshot-like object containing a cursor field. */
+export function setCursorFromSnapshot(key: string, snapshot: { cursor?: string | number | null }): void {
+  if (snapshot.cursor !== undefined && snapshot.cursor !== null) {
+    setCursor(key, String(snapshot.cursor));
+  }
+}
+
+/** Clear in-memory cursors for tests. */
+export function _resetCursorTrackerForTesting(): void {
+  syncCursorStore.clear();
+}
+
 /**
  * Build a namespaced cursor key from a base name and namespace.
  */

src/events.ts

@@ -1,4 +1,6 @@
 import { rpc as SorobanRpc } from "@stellar/stellar-sdk";
+import { TypedEventEmitter } from "./events/TypedEventEmitter.js";
+import type { FinalityStatus } from "./types.js";
 
 /** Type of contract event. */
 export type ContractEventType = "created" | "payment" | "released" | "refunded";
@@ -17,6 +19,21 @@ export interface ContractEvent {
   timestamp: number;
 }
 
+export interface SDKEventMap extends Record<string, unknown> {
+  streamStallDetected: { streamId: string };
+  streamAutoReset: { streamId: string };
+  invoiceFinalized: { txHash: string; finality: FinalityStatus };
+  approvalRequested: { signerPublicKey: string };
+  approvalReceived: { signerPublicKey: string };
+  approvalWorkflowComplete: { signerCount: number };
+}
+
+export const sdkEvents = new TypedEventEmitter<SDKEventMap>();
+
+export function emitSdkEvent<K extends keyof SDKEventMap>(event: K, payload: SDKEventMap[K]): void {
+  sdkEvents.emit(event, payload);
+}
+
 /**
  * Replay historical contract events in a ledger range.
  *

src/finalityChecker.ts

@@ -0,0 +1,77 @@
+import { FinalityTimeoutError } from "./errors.js";
+import { emitSdkEvent } from "./events.js";
+import type { FinalityCheckConfig, FinalityStatus } from "./types.js";
+
+interface CallBuilder<T> {
+  call(): Promise<T>;
+}
+
+interface TransactionRecordLike {
+  ledger: number | string;
+  successful?: boolean;
+  result_successful?: boolean;
+}
+
+interface LedgerRecordLike {
+  sequence?: number | string;
+}
+
+export interface FinalityServerLike {
+  transactions(): {
+    transaction(hash: string): CallBuilder<TransactionRecordLike>;
+  };
+  ledgers(): {
+    order(direction: "desc"): {
+      limit(limit: number): CallBuilder<{ records: LedgerRecordLike[] }>;
+    };
+  };
+}
+
+const sleep = (ms: number): Promise<void> => new Promise((resolve) => setTimeout(resolve, ms));
+
+export class FinalityChecker {
+  private readonly server: FinalityServerLike;
+  private readonly config: Required<FinalityCheckConfig>;
+
+  constructor(server: FinalityServerLike, config: FinalityCheckConfig = {}) {
+    this.server = server;
+    this.config = {
+      minConfirmations: config.minConfirmations ?? 2,
+      pollIntervalMs: config.pollIntervalMs ?? 1_000,
+      maxWaitMs: config.maxWaitMs ?? 30_000,
+    };
+  }
+
+  async check(txHash: string): Promise<FinalityStatus> {
+    const startedAt = Date.now();
+
+    while (true) {
+      const status = await this.readStatus(txHash);
+      if (status.finalized) {
+        emitSdkEvent("invoiceFinalized", { txHash, finality: status });
+        return status;
+      }
+
+      if (Date.now() - startedAt >= this.config.maxWaitMs) {
+        throw new FinalityTimeoutError(txHash, this.config.maxWaitMs);
+      }
+
+      await sleep(this.config.pollIntervalMs);
+    }
+  }
+
+  private async readStatus(txHash: string): Promise<FinalityStatus> {
+    const tx = await this.server.transactions().transaction(txHash).call();
+    const ledgerSequence = Number(tx.ledger);
+    const latest = await this.server.ledgers().order("desc").limit(1).call();
+    const currentLedger = Number(latest.records[0]?.sequence ?? ledgerSequence);
+    const confirmations = Math.max(0, currentLedger - ledgerSequence);
+    const successful = tx.successful === true || tx.result_successful === true;
+
+    return {
+      finalized: successful && confirmations >= this.config.minConfirmations,
+      confirmations,
+      ledgerSequence,
+    };
+  }
+}

src/index.ts

@@ -199,6 +199,10 @@ export {
   ClassifiedHorizonError,
   isClassifiedHorizonError,
   HorizonErrorClassification,
+  FinalityTimeoutError,
+  isFinalityTimeoutError,
+  ApprovalTimeoutError,
+  isApprovalTimeoutError,
 } from "./errors.js";
 
 // ---------------------------------------------------------------------------
@@ -269,6 +273,14 @@ export { Deduplicator } from "./dedup.js";
 export { TxQueue } from "./queue.js";
 
 export { replayEvents } from "./events.js";
+export { sdkEvents } from "./events.js";
+export { FinalityChecker } from "./finalityChecker.js";
+export type { FinalityServerLike } from "./finalityChecker.js";
+export { ApprovalWorkflowSequencer, ApprovalSession } from "./approvalWorkflowSequencer.js";
+export type { ApprovalWorkflowOptions, NotificationAdapter, SignatureApplier } from "./approvalWorkflowSequencer.js";
+export { OperationChunker, MAX_OPERATIONS_PER_TRANSACTION } from "./operationChunker.js";
+export { StreamHealthProbe } from "./streamHealthProbe.js";
+export type { MonitoredStream, StreamHealthProbeOptions } from "./streamHealthProbe.js";
 export {
   EventChecksumChain,
   verifyChain,
@@ -394,6 +406,12 @@ export type {
   DecodedTransactionMeta,
   DecodedLedgerEntry,
   DecodedOperation,
+  FinalityStatus,
+  FinalityCheckConfig,
+  MultiSigPolicy,
+  ApprovalSessionResult,
+  BatchPaymentResult,
+  ChunkSubmitter,
 } from "./types.js";
 export { InvalidTransitionError } from "./types.js";
 
@@ -580,6 +598,7 @@ export {
   generatePaymentReceipt,
   serializePaymentReceipt,
   deserializePaymentReceipt,
+  finalizePaymentReceipt,
 } from "./receipt.js";
 export type {
   PaymentReceipt,

src/operationChunker.ts

@@ -0,0 +1,64 @@
+import { IdempotencyManager } from "./idempotency.js";
+import type { BatchPaymentResult, ChunkSubmitter } from "./types.js";
+import type { OperationOptions } from "@stellar/stellar-base";
+
+export const MAX_OPERATIONS_PER_TRANSACTION = 100;
+
+export class OperationChunker {
+  private readonly idempotency: IdempotencyManager;
+
+  constructor(idempotency = new IdempotencyManager()) {
+    this.idempotency = idempotency;
+  }
+
+  static chunk<T extends OperationOptions>(operations: T[], chunkSize = MAX_OPERATIONS_PER_TRANSACTION): T[][] {
+    if (chunkSize <= 0) {
+      throw new Error("chunkSize must be greater than zero");
+    }
+
+    const chunks: T[][] = [];
+    for (let index = 0; index < operations.length; index += chunkSize) {
+      chunks.push(operations.slice(index, index + chunkSize));
+    }
+    return chunks;
+  }
+
+  chunk<T extends OperationOptions>(operations: T[], chunkSize = MAX_OPERATIONS_PER_TRANSACTION): T[][] {
+    return OperationChunker.chunk(operations, chunkSize);
+  }
+
+  async submitAll<T extends OperationOptions>(
+    chunks: T[][],
+    config: { sourceAddress: string; submitChunk: ChunkSubmitter<T> },
+  ): Promise<BatchPaymentResult> {
+    const result: BatchPaymentResult = { succeeded: [], failed: null };
+
+    for (let index = 0; index < chunks.length; index++) {
+      const operations = chunks[index]!;
+      const key = this.idempotency.generateKey(
+        config.sourceAddress,
+        JSON.stringify({ index, operations }),
+      );
+      const existing = this.idempotency.getResult(key);
+
+      if (existing) {
+        result.succeeded.push({ chunkIndex: index, txHash: existing.txHash });
+        continue;
+      }
+
+      try {
+        const submitted = await config.submitChunk(operations, index);
+        this.idempotency.tryClaim(key, { txHash: submitted.txHash });
+        result.succeeded.push({ chunkIndex: index, txHash: submitted.txHash });
+      } catch (error) {
+        result.failed = {
+          chunkIndex: index,
+          error: error instanceof Error ? error : new Error(String(error)),
+        };
+        break;
+      }
+    }
+
+    return result;
+  }
+}