From c7c01e57ad48f7fd1873ea61f3a6837045391384 Mon Sep 17 00:00:00 2001 From: King In The North Date: Wed, 29 Jul 2026 15:37:07 +0000 Subject: [PATCH 1/4] Add AccountDataManager for typed account data entry CRUD (#528) --- src/accountDataManager.ts | 151 ++++++++++++++++++++++++++++++++++++++ src/errors.ts | 23 ++++++ src/index.ts | 15 ++++ src/txBuilder.ts | 10 +++ src/types.ts | 15 ++++ 5 files changed, 214 insertions(+) create mode 100644 src/accountDataManager.ts diff --git a/src/accountDataManager.ts b/src/accountDataManager.ts new file mode 100644 index 0000000..4fc2417 --- /dev/null +++ b/src/accountDataManager.ts @@ -0,0 +1,151 @@ +/** + * Typed CRUD manager for Stellar account data entries. + * + * Wraps `Operation.manageData()` with validation for the protocol's 64-byte + * key/value limits and 64-entry-per-account cap, so callers can store custom + * metadata alongside SDK state without hand-rolling raw manageData calls. + */ + +import { + Account, + Horizon, + Keypair, + Operation, + TransactionBuilder, + BASE_FEE, +} from "@stellar/stellar-sdk"; +import type { AccountDataMap } from "./types.js"; +import { DataEntryValidationError } from "./errors.js"; + +/** Stellar protocol limit for both data entry keys and values, in bytes. */ +const MAX_DATA_ENTRY_BYTES = 64; + +/** Stellar protocol limit on the number of data entries per account. */ +const MAX_DATA_ENTRIES = 64; + +/** Result of submitting a manageData transaction. */ +export interface TransactionResult { + txHash: string; +} + +/** Configuration for {@link AccountDataManager}. */ +export interface AccountDataManagerConfig { + /** Horizon server URL. */ + horizonUrl: string; + /** Stellar network passphrase. */ + networkPassphrase: string; +} + +function byteLength(value: string): number { + return Buffer.byteLength(value, "utf8"); +} + +/** + * Typed CRUD manager for account data entries, built on top of + * `Operation.manageData()` and `Server.loadAccount().data_attr`. + */ +export class AccountDataManager { + private readonly server: Horizon.Server; + private readonly networkPassphrase: string; + + constructor(config: AccountDataManagerConfig) { + this.server = new Horizon.Server(config.horizonUrl); + this.networkPassphrase = config.networkPassphrase; + } + + /** + * Set (create or update) a data entry on `accountId`. + * + * @throws DataEntryValidationError if the key/value exceed 64 bytes, or if + * the account already has 64 entries and `key` is new. + */ + async set( + accountId: string, + key: string, + value: string, + signerSecret: string, + ): Promise { + await this.validateEntry(accountId, key, value); + return this.submitManageData(accountId, key, value, signerSecret); + } + + /** + * Fetch the current value of `key` on `accountId`, or `null` if absent. + */ + async get(accountId: string, key: string): Promise { + const entries = await this.list(accountId); + return Object.prototype.hasOwnProperty.call(entries, key) ? entries[key]! : null; + } + + /** + * Delete a data entry by submitting `manageData` with a `null` value. + */ + async delete( + accountId: string, + key: string, + signerSecret: string, + ): Promise { + return this.submitManageData(accountId, key, null, signerSecret); + } + + /** + * Return all data entries currently stored on `accountId`, decoded from + * base64 to UTF-8 strings. + */ + async list(accountId: string): Promise { + const account = await this.server.loadAccount(accountId); + const raw = account.data_attr as Record | undefined; + const result: AccountDataMap = {}; + for (const [key, base64Value] of Object.entries(raw ?? {})) { + result[key] = Buffer.from(base64Value, "base64").toString("utf8"); + } + return result; + } + + private async validateEntry(accountId: string, key: string, value: string): Promise { + if (byteLength(key) > MAX_DATA_ENTRY_BYTES) { + throw new DataEntryValidationError( + `key "${key}" exceeds ${MAX_DATA_ENTRY_BYTES} bytes`, + { key }, + ); + } + if (byteLength(value) > MAX_DATA_ENTRY_BYTES) { + throw new DataEntryValidationError( + `value for key "${key}" exceeds ${MAX_DATA_ENTRY_BYTES} bytes`, + { key }, + ); + } + + const existing = await this.list(accountId); + const isNewKey = !Object.prototype.hasOwnProperty.call(existing, key); + if (isNewKey && Object.keys(existing).length >= MAX_DATA_ENTRIES) { + throw new DataEntryValidationError( + `account ${accountId} already has ${MAX_DATA_ENTRIES} data entries`, + { accountId }, + ); + } + } + + private async submitManageData( + accountId: string, + key: string, + value: string | null, + signerSecret: string, + ): Promise { + const keypair = Keypair.fromSecret(signerSecret); + const loaded = await this.server.loadAccount(accountId); + const sourceAccount = new Account(loaded.accountId(), loaded.sequenceNumber()); + + const tx = new TransactionBuilder(sourceAccount, { + fee: BASE_FEE, + networkPassphrase: this.networkPassphrase, + }) + .addOperation(Operation.manageData({ name: key, value: value ?? null })) + .setTimeout(30) + .build(); + + tx.sign(keypair); + const result = await this.server.submitTransaction(tx); + return { txHash: result.hash }; + } +} diff --git a/src/errors.ts b/src/errors.ts index 47f1a5c..e26e65a 100644 --- a/src/errors.ts +++ b/src/errors.ts @@ -1699,3 +1699,26 @@ export class ClassifiedHorizonError extends StellarSplitError { export function isClassifiedHorizonError(err: unknown): err is ClassifiedHorizonError { return err instanceof ClassifiedHorizonError; } + +// --------------------------------------------------------------------------- +// Account Data Entry errors +// --------------------------------------------------------------------------- + +/** + * Thrown when an account data entry key/value exceeds the 64-byte Stellar + * protocol limit, or when adding a new key would exceed the 64-entry cap. + */ +export class DataEntryValidationError extends StellarSplitError { + readonly reason: string; + + constructor(reason: string, context?: Record) { + super(`Account data entry validation failed: ${reason}`, "DATA_ENTRY_VALIDATION_ERROR", context); + this.name = "DataEntryValidationError"; + this.reason = reason; + Object.setPrototypeOf(this, new.target.prototype); + } +} + +export function isDataEntryValidationError(err: unknown): err is DataEntryValidationError { + return err instanceof DataEntryValidationError; +} diff --git a/src/index.ts b/src/index.ts index 38d8fe1..79306bc 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1112,3 +1112,18 @@ export type { ScValPrimitive, ContractStorageExporterOptions, } from "./diagnostics/ContractStorageExporter.js"; + +// --------------------------------------------------------------------------- +// #528 — AccountDataManager: typed CRUD for account data entries +// --------------------------------------------------------------------------- + +export { AccountDataManager } from "./accountDataManager.js"; +export type { + AccountDataManagerConfig, + TransactionResult as AccountDataTransactionResult, +} from "./accountDataManager.js"; +export type { AccountDataEntry, AccountDataMap } from "./types.js"; +export { + DataEntryValidationError, + isDataEntryValidationError, +} from "./errors.js"; diff --git a/src/txBuilder.ts b/src/txBuilder.ts index 7469db2..79733d9 100644 --- a/src/txBuilder.ts +++ b/src/txBuilder.ts @@ -86,6 +86,16 @@ export class StellarSplitTxBuilder { return this; } + /** + * Add a ManageData operation for setting or clearing an account data entry. + * Pass `value: null` to clear the entry. + */ + addManageData(key: string, value: string | null): this { + const op = Operation.manageData({ name: key, value }); + this.operations.push(op); + return this; + } + /** * Add a PathPaymentStrictSend operation for cross-asset DEX-routed payments. */ diff --git a/src/types.ts b/src/types.ts index fff3f03..83c97cf 100644 --- a/src/types.ts +++ b/src/types.ts @@ -1647,3 +1647,18 @@ export interface CursorStore { /** Delete a saved cursor. */ delete(key: string): Promise; } + +// --------------------------------------------------------------------------- +// Account Data Entry Types (Issue #528) +// --------------------------------------------------------------------------- + +/** A single decoded key-value data entry stored on a Stellar account. */ +export interface AccountDataEntry { + /** Data entry key (max 64 bytes). */ + key: string; + /** Decoded (UTF-8) value, or null when the entry has been cleared. */ + value: string | null; +} + +/** All data entries currently stored on an account, keyed by entry name. */ +export type AccountDataMap = Record; From 4bada14db370e2033912fbb0cc13691e0c744fa6 Mon Sep 17 00:00:00 2001 From: King In The North Date: Wed, 29 Jul 2026 15:41:53 +0000 Subject: [PATCH 2/4] Add SorobanFeatureDetector for protocol upgrade detection (#529) --- src/client.ts | 20 ++++++ src/index.ts | 11 ++++ src/sorobanFeatureDetector.ts | 120 ++++++++++++++++++++++++++++++++++ src/types.ts | 24 +++++++ 4 files changed, 175 insertions(+) create mode 100644 src/sorobanFeatureDetector.ts diff --git a/src/client.ts b/src/client.ts index d7b0d69..75c9e2b 100644 --- a/src/client.ts +++ b/src/client.ts @@ -55,6 +55,8 @@ import { PluginRegistry } from "./plugin.js"; import type { SdkPlugin } from "./plugin.js"; import { checkRPCHealth } from "./health.js"; import { Deduplicator } from "./dedup.js"; +import { SorobanFeatureDetector } from "./sorobanFeatureDetector.js"; +import type { SorobanFeatureFlags } from "./types.js"; import { verifyBatchPayments } from "./batchVerifier.js"; import { type HealthCheckResult, HealthCheckTimeoutError } from "./types.js"; import type { @@ -652,6 +654,7 @@ export class StellarSplitClient extends TypedEventEmitter { private _advancedCircuitBreaker: AdvancedCircuitBreaker | null = null; /** Optimistic UI cache for Invoice reads during a pending pay() call. */ private _optimisticCache: OptimisticCache | null = null; + private _sorobanFeatureDetector: SorobanFeatureDetector; private _shutdownInProgress = false; private _pluginsDestroyed = false; private _runtimeShutdownPromise: Promise | null = null; @@ -806,6 +809,13 @@ export class StellarSplitClient extends TypedEventEmitter { this._stateMachine = new InvoiceStateMachine(config.stateMachine); + // Soroban protocol feature detection (Issue #529): probe once at startup; + // the detector caches internally and re-probes after its staleness window. + this._sorobanFeatureDetector = new SorobanFeatureDetector({ rpcUrl: primaryUrl }); + this._sorobanFeatureDetector.detect().catch(() => { + // Best-effort startup probe; getSorobanFeatures() will retry on demand. + }); + if ( !this._rpcClient && Array.isArray(config.rpcUrl) && @@ -1213,6 +1223,16 @@ export class StellarSplitClient extends TypedEventEmitter { * Performs a health check of the client's RPC connection and contract. * Resolves with status information or throws HealthCheckTimeoutError if taking > 5000ms. */ + /** + * Return the current Soroban protocol feature flags, detected once at + * startup and re-probed automatically after the detector's staleness + * window (default 1 hour). Emits `protocolUpgradeDetected` on the + * underlying detector when a re-probe observes a version change. + */ + async getSorobanFeatures(): Promise { + return this._sorobanFeatureDetector.detect(); + } + async healthCheck(): Promise { const start = Date.now(); try { diff --git a/src/index.ts b/src/index.ts index 79306bc..f8f7593 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1127,3 +1127,14 @@ export { DataEntryValidationError, isDataEntryValidationError, } from "./errors.js"; + +// --------------------------------------------------------------------------- +// #529 — SorobanFeatureDetector: protocol upgrade / feature flag detection +// --------------------------------------------------------------------------- + +export { SorobanFeatureDetector } from "./sorobanFeatureDetector.js"; +export type { + SorobanFeatureDetectorConfig, + SorobanFeatureDetectorEventMap, +} from "./sorobanFeatureDetector.js"; +export type { SorobanFeatureFlags } from "./types.js"; diff --git a/src/sorobanFeatureDetector.ts b/src/sorobanFeatureDetector.ts new file mode 100644 index 0000000..513ca52 --- /dev/null +++ b/src/sorobanFeatureDetector.ts @@ -0,0 +1,120 @@ +/** + * Detects Soroban network protocol upgrades and exposes typed feature flags. + * + * Queries `SorobanRpc.Server.getNetwork()` for the current protocol version + * and `getLedgerEntries()` for the `ConfigSettingContractComputeV0` entry to + * read resource limits. Results are cached and re-detected after a + * configurable staleness interval so the SDK never re-probes on every call. + */ + +import { rpc as SorobanRpc, xdr } from "@stellar/stellar-sdk"; +import { TypedEventEmitter } from "./events/TypedEventEmitter.js"; +import type { SorobanFeatureFlags } from "./types.js"; + +/** Protocol version at which ExtendFootprintTtl / RestoreFootprint became available. */ +const STATE_ARCHIVAL_PROTOCOL_VERSION = 20; + +/** Events emitted by {@link SorobanFeatureDetector}. */ +export interface SorobanFeatureDetectorEventMap { + [key: string]: unknown; + protocolUpgradeDetected: { previousVersion: number; newVersion: number }; +} + +/** Configuration for {@link SorobanFeatureDetector}. */ +export interface SorobanFeatureDetectorConfig { + /** Soroban RPC endpoint URL. */ + rpcUrl: string; + /** Milliseconds before cached flags are considered stale. Default: 1 hour. */ + staleAfterMs?: number; +} + +const DEFAULT_STALE_AFTER_MS = 60 * 60 * 1000; + +const FALLBACK_FLAGS: Omit = { + protocolVersion: 0, + supportsExtendFootprint: false, + supportsRestoreFootprint: false, + maxInstructionsPerTx: 0, + maxInstructionsPerLedger: 0, +}; + +/** + * Detects the current Soroban protocol version and derives typed feature + * flags from it, caching the result and emitting `protocolUpgradeDetected` + * when a re-check observes a version change. + */ +export class SorobanFeatureDetector extends TypedEventEmitter { + private readonly server: SorobanRpc.Server; + private readonly staleAfterMs: number; + private cached: SorobanFeatureFlags | null = null; + + constructor(config: SorobanFeatureDetectorConfig) { + super(); + this.server = new SorobanRpc.Server(config.rpcUrl, { + allowHttp: config.rpcUrl.startsWith("http://"), + }); + this.staleAfterMs = config.staleAfterMs ?? DEFAULT_STALE_AFTER_MS; + } + + /** + * Return the current feature flags, using the cached value unless it is + * stale (older than `staleAfterMs`). + */ + async detect(): Promise { + if (this.cached && Date.now() - this.cached.detectedAt < this.staleAfterMs) { + return this.cached; + } + + const previousVersion = this.cached?.protocolVersion; + const flags = await this.probe(); + + if (previousVersion !== undefined && previousVersion !== flags.protocolVersion) { + this.emit("protocolUpgradeDetected", { + previousVersion, + newVersion: flags.protocolVersion, + }); + } + + this.cached = flags; + return flags; + } + + /** Force a re-detection on the next {@link detect} call. */ + invalidate(): void { + this.cached = null; + } + + private async probe(): Promise { + const network = await this.server.getNetwork(); + const protocolVersion = parseInt(String(network.protocolVersion), 10) || 0; + + let maxInstructionsPerTx = FALLBACK_FLAGS.maxInstructionsPerTx; + let maxInstructionsPerLedger = FALLBACK_FLAGS.maxInstructionsPerLedger; + + try { + const key = xdr.LedgerKey.configSetting( + new xdr.LedgerKeyConfigSetting({ + configSettingId: xdr.ConfigSettingId.configSettingContractComputeV0(), + }), + ); + const response = await this.server.getLedgerEntries(key); + const entry = response.entries[0]; + const computeV0 = entry?.val.configSetting().contractCompute(); + if (computeV0) { + maxInstructionsPerTx = Number(computeV0.txMaxInstructions()); + maxInstructionsPerLedger = Number(computeV0.ledgerMaxInstructions()); + } + } catch { + // Resource limits unavailable — fall back to defaults, protocol version is still valid. + } + + return { + protocolVersion, + supportsExtendFootprint: protocolVersion >= STATE_ARCHIVAL_PROTOCOL_VERSION, + supportsRestoreFootprint: protocolVersion >= STATE_ARCHIVAL_PROTOCOL_VERSION, + maxInstructionsPerTx, + maxInstructionsPerLedger, + detectedAt: Date.now(), + }; + } +} diff --git a/src/types.ts b/src/types.ts index 83c97cf..5026170 100644 --- a/src/types.ts +++ b/src/types.ts @@ -1662,3 +1662,27 @@ export interface AccountDataEntry { /** All data entries currently stored on an account, keyed by entry name. */ export type AccountDataMap = Record; + +// --------------------------------------------------------------------------- +// Soroban Feature Detection Types (Issue #529) +// --------------------------------------------------------------------------- + +/** + * Typed flags for protocol-version-gated Soroban features, plus the raw + * resource limits pulled from the network's `ConfigSettingEntry` ledger + * entries. + */ +export interface SorobanFeatureFlags { + /** Current Stellar protocol version integer. */ + protocolVersion: number; + /** Whether the network supports the `ExtendFootprintTtl` operation (protocol >= 20). */ + supportsExtendFootprint: boolean; + /** Whether the network supports archived-entry restoration (protocol >= 20). */ + supportsRestoreFootprint: boolean; + /** Maximum Soroban instructions allowed per transaction. */ + maxInstructionsPerTx: number; + /** Maximum Soroban instructions allowed per ledger. */ + maxInstructionsPerLedger: number; + /** Unix timestamp (ms) when these flags were detected. */ + detectedAt: number; +} From 63919586f446427d037c633f099dde3e36548772 Mon Sep 17 00:00:00 2001 From: King In The North Date: Wed, 29 Jul 2026 15:44:16 +0000 Subject: [PATCH 3/4] Add StreamDeduplicator for paging-token-based stream event dedup (#530) --- src/index.ts | 17 ++++++ src/snapshot.ts | 40 ++++++++++++++ src/sse.ts | 14 +++++ src/stream.ts | 8 ++- src/streamDeduplicator.ts | 108 ++++++++++++++++++++++++++++++++++++++ 5 files changed, 186 insertions(+), 1 deletion(-) create mode 100644 src/streamDeduplicator.ts diff --git a/src/index.ts b/src/index.ts index f8f7593..c3cfc23 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1138,3 +1138,20 @@ export type { SorobanFeatureDetectorEventMap, } from "./sorobanFeatureDetector.js"; export type { SorobanFeatureFlags } from "./types.js"; + +// --------------------------------------------------------------------------- +// #530 — StreamDeduplicator: paging-token-based stream event dedup +// --------------------------------------------------------------------------- + +export { StreamDeduplicator } from "./streamDeduplicator.js"; +export type { + StreamDeduplicatorOptions, + StreamDeduplicatorEventMap, +} from "./streamDeduplicator.js"; +export { + InMemoryDedupTokenStore, + setDefaultDedupTokenStore, + saveDedupTokens, + loadDedupTokens, +} from "./snapshot.js"; +export type { DedupTokenStore } from "./snapshot.js"; diff --git a/src/snapshot.ts b/src/snapshot.ts index 6923437..e4067a7 100644 --- a/src/snapshot.ts +++ b/src/snapshot.ts @@ -31,3 +31,43 @@ export function snapshotInvoice(invoice: Invoice): InvoiceSnapshot { payments: frozenPayments, }); } + +// --------------------------------------------------------------------------- +// Stream dedup token persistence (Issue #530) +// --------------------------------------------------------------------------- + +/** Pluggable persistence for a stream deduplicator's seen-token set. */ +export interface DedupTokenStore { + save(namespace: string, tokens: string[]): Promise; + load(namespace: string): Promise; +} + +/** In-memory token store; state is lost on process exit. */ +export class InMemoryDedupTokenStore implements DedupTokenStore { + private store = new Map(); + + async save(namespace: string, tokens: string[]): Promise { + this.store.set(namespace, [...tokens]); + } + + async load(namespace: string): Promise { + return this.store.get(namespace) ?? null; + } +} + +let defaultDedupTokenStore: DedupTokenStore = new InMemoryDedupTokenStore(); + +/** Override the default dedup token store (e.g. with a durable backend). */ +export function setDefaultDedupTokenStore(store: DedupTokenStore): void { + defaultDedupTokenStore = store; +} + +/** Persist a deduplicator's current token set under `namespace`. */ +export async function saveDedupTokens(namespace: string, tokens: string[]): Promise { + await defaultDedupTokenStore.save(namespace, tokens); +} + +/** Load a previously persisted token set for `namespace`, or null if none exists. */ +export async function loadDedupTokens(namespace: string): Promise { + return defaultDedupTokenStore.load(namespace); +} diff --git a/src/sse.ts b/src/sse.ts index 1abd2cf..d5ef7f5 100644 --- a/src/sse.ts +++ b/src/sse.ts @@ -3,6 +3,7 @@ */ import { getCursor, setCursor } from "./cursorTracker.js"; +import type { StreamDeduplicator } from "./streamDeduplicator.js"; /** The three event types emitted by the invoice SSE stream. */ export type SSEInvoiceEventType = @@ -33,6 +34,12 @@ export interface SubscribeToInvoiceOptions { maxBackoffMs?: number; /** Factory for EventSource (injectable for testing). Default: global EventSource. */ eventSourceFactory?: (url: string) => EventSourceLike; + /** + * Optional deduplicator applied before dispatching to `handler`. When the + * incoming message includes a `paging_token` field and it has already been + * seen, the event is discarded instead of delivered. + */ + dedup?: StreamDeduplicator; } /** Minimal EventSource interface (subset of the browser API). */ @@ -71,6 +78,7 @@ export function subscribeToInvoice( initialBackoffMs = 1000, maxBackoffMs = 30_000, eventSourceFactory, + dedup, } = options; // Build URL with persisted cursor for resume-on-restart. @@ -99,6 +107,7 @@ export function subscribeToInvoice( type?: unknown; invoiceId?: unknown; data?: unknown; + paging_token?: unknown; }; if ( @@ -109,6 +118,11 @@ export function subscribeToInvoice( return; } + if (dedup && typeof raw.paging_token === "string") { + const isNew = dedup.filter({ paging_token: raw.paging_token }); + if (!isNew) return; + } + backoff = initialBackoffMs; // reset on successful message // Persist event timestamp as cursor for resume-on-restart. diff --git a/src/stream.ts b/src/stream.ts index 61b22de..f408e73 100644 --- a/src/stream.ts +++ b/src/stream.ts @@ -3,6 +3,7 @@ import type { InvoiceEventCallbacks, Payment } from "./types.js"; import type { SSEInvoiceEvent } from "./sse.js"; import { TooManySubscriptionsError } from "./errors.js"; import { getCursor, setCursor } from "./cursorTracker.js"; +import type { StreamDeduplicator } from "./streamDeduplicator.js"; /** Maximum concurrent subscriptions allowed. */ const MAX_SUBSCRIPTIONS = 10; @@ -117,6 +118,8 @@ function extractPayment(event: SorobanRpc.Api.EventResponse): Payment | null { * @param invoiceId - The invoice ID to watch * @param handlerOrCallbacks - Called with InvoiceEvent[] (events since last poll), or callbacks object for legacy API * @param intervalMs - Poll interval in milliseconds (default: 5000, max: 30000) + * @param dedup - Optional deduplicator; events whose pagingToken has already + * been processed are discarded before reaching handlers/callbacks. * @returns Unsubscribe function that stops the stream * @throws TooManySubscriptionsError if more than 10 subscriptions are created */ @@ -125,7 +128,8 @@ export function subscribeToInvoice( contractId: string, invoiceId: string, handlerOrCallbacks: ((events: SSEInvoiceEvent[]) => void) | InvoiceEventCallbacks, - intervalMs: number = 5000 + intervalMs: number = 5000, + dedup?: StreamDeduplicator ): () => void { // Resume from last persisted cursor when available. const streamId = `contract:${contractId}:invoice:${invoiceId}`; @@ -217,6 +221,8 @@ export function subscribeToInvoice( const eventInvoiceId = extractInvoiceId(event); if (eventInvoiceId !== invoiceId) continue; + if (dedup && !dedup.filter({ paging_token: event.pagingToken })) continue; + hasChanges = true; const invoiceEvent = convertToInvoiceEvent( diff --git a/src/streamDeduplicator.ts b/src/streamDeduplicator.ts new file mode 100644 index 0000000..7a74716 --- /dev/null +++ b/src/streamDeduplicator.ts @@ -0,0 +1,108 @@ +/** + * Paging-token-based deduplicator for Horizon SSE / stream consumers. + * + * Horizon streams occasionally redeliver the same event (reconnects, Horizon + * restarts). This maintains a fixed-size circular buffer of recently seen + * paging tokens and discards records whose token has already been processed, + * persisting the token set periodically so restarts don't reprocess events. + */ + +import { TypedEventEmitter } from "./events/TypedEventEmitter.js"; +import { saveDedupTokens, loadDedupTokens } from "./snapshot.js"; + +/** Events emitted by {@link StreamDeduplicator}. */ +export interface StreamDeduplicatorEventMap { + [key: string]: unknown; + duplicateEventDiscarded: { pagingToken: string }; +} + +/** Configuration for {@link StreamDeduplicator}. */ +export interface StreamDeduplicatorOptions { + /** Maximum number of paging tokens retained. Oldest tokens are evicted first. Default: 1000. */ + windowSize?: number; + /** Persist the token set after this many newly seen tokens. Default: 100. */ + flushIntervalTokens?: number; + /** Namespace used to key persisted state. Default: "default". */ + namespace?: string; +} + +const DEFAULT_WINDOW_SIZE = 1000; +const DEFAULT_FLUSH_INTERVAL_TOKENS = 100; + +/** + * Discards duplicate stream records by paging token using a fixed-size + * circular buffer, with periodic persistence for restart survival. + */ +export class StreamDeduplicator extends TypedEventEmitter { + private readonly windowSize: number; + private readonly flushIntervalTokens: number; + private readonly namespace: string; + + private readonly seen = new Set(); + private readonly buffer: string[] = []; + private writeIndex = 0; + private newTokensSinceFlush = 0; + + constructor(options: StreamDeduplicatorOptions = {}) { + super(); + this.windowSize = options.windowSize ?? DEFAULT_WINDOW_SIZE; + this.flushIntervalTokens = options.flushIntervalTokens ?? DEFAULT_FLUSH_INTERVAL_TOKENS; + this.namespace = options.namespace ?? "default"; + } + + /** + * Load the previously persisted token set for this deduplicator's + * namespace. Call once at startup, before processing any new events. + */ + async restore(): Promise { + const tokens = await loadDedupTokens(this.namespace); + if (!tokens) return; + + for (const token of tokens.slice(-this.windowSize)) { + if (this.seen.has(token)) continue; + this.seen.add(token); + this.buffer.push(token); + } + this.writeIndex = this.buffer.length % this.windowSize; + } + + /** + * Returns `false` when `record.paging_token` has already been seen within + * the current window (i.e. it is a duplicate and should be discarded). + * Returns `true` (and records the token) otherwise. + */ + filter(record: T): boolean { + const token = record.paging_token; + + if (this.seen.has(token)) { + this.emit("duplicateEventDiscarded", { pagingToken: token }); + return false; + } + + this.record(token); + return true; + } + + /** Number of tokens currently tracked. */ + get size(): number { + return this.seen.size; + } + + private record(token: string): void { + if (this.buffer.length >= this.windowSize) { + const evicted = this.buffer[this.writeIndex]; + if (evicted !== undefined) this.seen.delete(evicted); + this.buffer[this.writeIndex] = token; + } else { + this.buffer.push(token); + } + this.writeIndex = (this.writeIndex + 1) % this.windowSize; + this.seen.add(token); + + this.newTokensSinceFlush++; + if (this.newTokensSinceFlush >= this.flushIntervalTokens) { + this.newTokensSinceFlush = 0; + void saveDedupTokens(this.namespace, [...this.buffer]); + } + } +} From 3b35c10186c81dfcf07a2535c9e7d058bd4c205b Mon Sep 17 00:00:00 2001 From: King In The North Date: Wed, 29 Jul 2026 15:45:32 +0000 Subject: [PATCH 4/4] Add per-split audit log entries to AuditLogger and compliance export (#531) --- src/auditLogger.ts | 39 ++++++++++++++++++++++++++++++++++++++- src/complianceExporter.ts | 39 ++++++++++++++++++++++++++++++++++++++- src/index.ts | 12 ++++++++++++ src/types.ts | 24 ++++++++++++++++++++++++ 4 files changed, 112 insertions(+), 2 deletions(-) diff --git a/src/auditLogger.ts b/src/auditLogger.ts index dc2361c..063713a 100644 --- a/src/auditLogger.ts +++ b/src/auditLogger.ts @@ -1,6 +1,6 @@ import { truncateAddress } from "./utils.js"; import { decodeXDR } from "./xdrDecoder.js"; -import type { XDRType, DecodedXDR } from "./types.js"; +import type { XDRType, DecodedXDR, SplitAuditEntry } from "./types.js"; export interface AuditEntry { timestamp: number; @@ -22,6 +22,7 @@ const MIN_XDR_LENGTH = 40; export class AuditLogger { private readonly sink: (entry: AuditEntry) => void; + private readonly splitAuditTrails = new Map(); constructor(sink: (entry: AuditEntry) => void) { this.sink = sink; @@ -101,4 +102,40 @@ export class AuditLogger { this.log(entry); } + + /** + * Record a `SplitAuditEntry` for a single settled leg of a multi-recipient + * split payment. Writes immediately to the configured sink (as an + * `AuditEntry`) and to the in-memory per-invoice trail returned by + * {@link exportSplitAuditTrail}. + */ + recordSplitLeg(entry: SplitAuditEntry): void { + const trail = this.splitAuditTrails.get(entry.invoiceId) ?? []; + trail.push(entry); + this.splitAuditTrails.set(entry.invoiceId, trail); + + this.log({ + timestamp: entry.settledAt, + method: "split_leg_settled", + params: this.sanitize({ + invoiceId: entry.invoiceId, + legIndex: entry.legIndex, + recipientId: entry.recipientId, + assetCode: entry.assetCode, + amount: entry.amount.toString(), + operationId: entry.operationId, + ledgerSequence: entry.ledgerSequence, + }), + success: true, + durationMs: 0, + }); + } + + /** + * Return all recorded `SplitAuditEntry` records for `invoiceId`, in the + * order they were settled. + */ + async exportSplitAuditTrail(invoiceId: string): Promise { + return [...(this.splitAuditTrails.get(invoiceId) ?? [])]; + } } diff --git a/src/complianceExporter.ts b/src/complianceExporter.ts index b152e92..d6ca37f 100644 --- a/src/complianceExporter.ts +++ b/src/complianceExporter.ts @@ -1,4 +1,4 @@ -import type { Invoice } from "./types.js"; +import type { Invoice, SplitAuditEntry } from "./types.js"; /** A single row in the compliance audit export. */ export interface ComplianceExportRecord { @@ -142,3 +142,40 @@ function csvEscape(value: string): string { } return value; } + +// --------------------------------------------------------------------------- +// Per-split audit trail export (Issue #531) +// --------------------------------------------------------------------------- + +/** Stable, ordered list of CSV column names for the split audit trail export. */ +export const SPLIT_AUDIT_CSV_COLUMNS = [ + "invoiceId", + "legIndex", + "recipientId", + "assetCode", + "amount", + "operationId", + "ledgerSequence", + "settledAt", +] as const; + +/** + * Produces a CSV export of per-split-leg audit entries (as recorded by + * `AuditLogger.recordSplitLeg`), suitable for inclusion in compliance reports. + */ +export function exportSplitAuditTrail(entries: SplitAuditEntry[]): string { + const header = SPLIT_AUDIT_CSV_COLUMNS.join(","); + const rows = entries.map((e) => + [ + e.invoiceId, + String(e.legIndex), + e.recipientId, + e.assetCode, + String(e.amount), + e.operationId, + String(e.ledgerSequence), + String(e.settledAt), + ].join(","), + ); + return [header, ...rows].join("\n"); +} diff --git a/src/index.ts b/src/index.ts index c3cfc23..56c4975 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1155,3 +1155,15 @@ export { loadDedupTokens, } from "./snapshot.js"; export type { DedupTokenStore } from "./snapshot.js"; + +// --------------------------------------------------------------------------- +// #531 — Per-Split Audit Log Emitter +// --------------------------------------------------------------------------- + +export { AuditLogger } from "./auditLogger.js"; +export type { AuditEntry } from "./auditLogger.js"; +export type { SplitAuditEntry } from "./types.js"; +export { + exportSplitAuditTrail, + SPLIT_AUDIT_CSV_COLUMNS, +} from "./complianceExporter.js"; diff --git a/src/types.ts b/src/types.ts index 5026170..df948e6 100644 --- a/src/types.ts +++ b/src/types.ts @@ -1686,3 +1686,27 @@ export interface SorobanFeatureFlags { /** Unix timestamp (ms) when these flags were detected. */ detectedAt: number; } + +// --------------------------------------------------------------------------- +// Per-Split Audit Log Types (Issue #531) +// --------------------------------------------------------------------------- + +/** A granular audit record for a single settled leg of a multi-recipient split payment. */ +export interface SplitAuditEntry { + /** Invoice the split payment belongs to. */ + invoiceId: string; + /** Zero-based index of this leg within the split. */ + legIndex: number; + /** Stellar address of the recipient for this leg. */ + recipientId: string; + /** Asset code paid out for this leg. */ + assetCode: string; + /** Amount paid to this recipient, in stroops. */ + amount: bigint; + /** Operation ID of the settlement operation. */ + operationId: string; + /** Ledger sequence number at which the leg settled. */ + ledgerSequence: number; + /** Unix timestamp (seconds) when the leg settled. */ + settledAt: number; +}