fix(streams,events,soroban): serialize batchWithdraw, decode missing events, configurable fee; add Module44 (#378, #504, #506, #509)

CHANGELOG.md

@@ -18,6 +18,9 @@ All notable changes are documented here. Format based on [Keep a Changelog](http
 - Direct unit tests for `resolvePassphrase()` covering explicit passphrase present/blank, named network known/unknown, and neither-provided branches (#462)
 - `StreamsModule.forceCancel()` — wraps the contract's `force_cancel()` so a recipient can force-cancel a stream paused beyond the 30-day threshold (previously only a prose TODO; `StreamErrorCode.PauseThresholdNotMet` is now reachable through the SDK) (#453)
 - `StreamsModule.transferRecipient()` — wraps the contract's `transfer_recipient()` so the current recipient can reassign the recipient role (previously only a prose TODO) (#454)
+- `ConduitConfig.fee` / `ConduitConfig.feeMultiplier` — configure the inclusion (bid) fee for submitted transactions, via `resolveFee()` (exported from `src/soroban.ts`) and a new `fee` parameter on `buildContractCallTx()`. Previously every transaction hardcoded `fee: BASE_FEE` with no override, so under inclusion-fee pressure a 100-stroop bid would never be selected (#509)
+- `StreamEventHandlers.onCreated`, `.onForceCancel`, `.onRecipientTransfer`, `.onOperatorSet`, `.onOperatorRevoke` — `dispatchEvent()` previously silently ignored the `created`, `force_cxl`, `xfer_rec`, `set_op`, and `rm_op` contract topics, so a live subscriber was never notified when the recipient role was transferred, a paused stream was force-cancelled by the recipient, or an operator was delegated/revoked (#506)
+- `Module44` stream liquidity-risk / runway calculator for Feature #44 (#378); follows `Module26`/`Module36`/`Module48`'s shared `LruMemoCache` pattern and reports an honest, workload-dependent `measuredSpeedupPercent` rather than a fixed percentage. Classifies each stream's remaining runway into `'inactive' | 'critical' | 'warning' | 'healthy'` and exposes `estimateTopUpNeeded()` to compute the stroops needed to reach a target runway
 - `NonceManager` (the bigint-based, queue-with-cancellation implementation) is now exported from the package entry point, along with its `NonceLock`/`NonceManagerOptions` types. Previously neither `NonceManager` implementation was reachable outside SDK source (#483).
 - `subscribeToStream()` / `client.streams.subscribe()` now accept `maxBackoffMs` and `maxConsecutiveFailures` options controlling the new exponential-backoff-with-cutoff polling behaviour (#485).
 - `paramToScVal(value, type?)` accepts an explicit ScVal type hint (`'u64'`, `'i128'`, ...) and passes already-encoded `xdr.ScVal`s through untouched; `BatchOperation.types` supplies per-field type information for the `params` map (e.g. `{ streamId: 'u64', amount: 'i128' }`) so a u64 stream ID or an i128 amount encodes with the exact width the contract expects instead of the default inference (#497).
@@ -67,6 +70,8 @@ All notable changes are documented here. Format based on [Keep a Changelog](http
 - `catchNetworkError()` no longer reclassifies *any* `TypeError` whose text happens to contain `fetch`/`connect`/`network`/etc. It now only reclassifies errors that are provably transport failures: the canonical fetch/axios network messages (`fetch failed`, `Failed to fetch`, `Network Error`, `Load failed`) or an error (or its nested `cause`) carrying a network errno code such as `ECONNREFUSED`/`ENOTFOUND`/`ERR_NETWORK`. A programming `TypeError` (e.g. `Cannot read properties of undefined (reading 'connect')`) is re-thrown as-is instead of being masked as a network outage (#457).
 - `NonceManager` now throws a descriptive error for an unparseable nonce string (e.g. `startNonce: 'not-a-number'`) instead of silently coercing it to `0n`, which masked caller bugs as an explicit zero (#458).
 - `StreamBuilder.build()` now stringifies a numeric `ratePerSecond` so the runtime value matches the declared `ratePerSecond?: string` return type; previously a `number` input passed through unchanged, so callers trusting the type (`.trim()`, string concatenation) hit runtime errors (#459).
+- `StreamsModule.batchWithdraw()` now submits withdrawals one at a time instead of firing them all concurrently via `Promise.allSettled(...map)`. Concurrent submission meant every withdrawal read the same account sequence number from `getAccount()` and built a transaction with the same value — with a single keypair/wallet, at most one of the N withdrawals could ever land, the rest failing with `txBAD_SEQ`, defeating the whole purpose of a batch (#504)
+- `src/streams.ts` was missing its `Signer` type import, so `npm run build` / `npm run typecheck` failed unconditionally on `main` (`Cannot find name 'Signer'`) — added the missing `import type { Signer } from './signer.js'`.
 
 - `StreamsModule.withdraw()` / `topUp()` now reject `amount <= 0n` client-side (before any RPC round-trip), matching `create()`'s fail-fast validation philosophy instead of relying on the contract's `InvalidAmount` simulate+reject cycle (#451)
 - `StreamsModule.list()` no longer silently drops `recipient` when both `sender` and `recipient` are provided — it now returns the de-duplicated union of both filters (#452)

docs/api.md

@@ -20,6 +20,17 @@ new ConduitClient(config: ConduitConfig)
 | `factoryAddress` | `string` | | Deployed factory |
 | `governorAddress` | `string` | | Deployed governor |
 | `wallet` | `WalletAdapter` | | — |
+| `fee` | `string` | | Explicit inclusion (bid) fee in stroops for submitted transactions. Takes precedence over `feeMultiplier`. Defaults to `BASE_FEE` (100 stroops) |
+| `feeMultiplier` | `number` | | Multiplier applied to `BASE_FEE` to compute the inclusion fee, e.g. `10` bids 10x the network minimum. Ignored when `fee` is set |
+
+> **Inclusion fee.** Every submitted transaction previously used `BASE_FEE`
+> (the network minimum, 100 stroops) unconditionally, with no way to raise
+> it. Under inclusion-fee pressure (surge pricing, congested ledgers) a
+> 100-stroop bid is not selected, and the SDK's confirmation polling
+> eventually throws a misleading `Transaction timed out` instead of
+> surfacing "fee too low". Set `fee` (an exact stroops amount) or
+> `feeMultiplier` (a multiple of `BASE_FEE`) on `ConduitConfig` to bid
+> higher; `fee` wins if both are set.
 
 ### Convenience methods
 
@@ -86,6 +97,38 @@ const total = await client.streams.streamedTotal(streamId);
 
 ---
 
+### `batchWithdraw(withdrawals) → Promise<BatchWithdrawResult[]>`
+
+| Param | Type | Notes |
+|-------|------|-------|
+| `withdrawals` | `{ streamId: bigint \| string; amount?: bigint }[]` | One entry per stream to withdraw from |
+
+```typescript
+interface BatchWithdrawResult {
+  streamId: bigint;
+  success:  boolean;
+  txHash?:  string;  // present when success is true
+  error?:   string;  // present when success is false
+}
+```
+
+Withdraws from multiple streams. Soroban permits only one
+`invoke_host_function` operation per transaction, so this cannot be a single
+atomic transaction — each withdrawal is submitted as its own transaction and
+reported independently, so a failure on one streamId (e.g.
+`StreamNotFound`, insufficient balance) does not fail the others.
+
+Withdrawals are submitted **one at a time, not concurrently**. Each
+transaction is built from the caller account's current sequence number;
+submitting all N withdrawals at once means every one of them would read the
+same sequence number and collide, so with a single keypair/wallet at most
+one could ever land on-chain. Awaiting each withdrawal in turn guarantees
+every submission gets a distinct, ordered sequence number.
+
+**Requires:** `keypair`, `wallet`, or `signer` set (recipient)
+
+---
+
 ### `cancel(streamId) → Promise<string>`
 
 Atomically settles both parties (recipient gets owed amount, sender gets refund).
@@ -203,6 +246,11 @@ const sub = client.streams.subscribe(streamId, {
   onResume:    e => console.log('Resumed at:', e.resumedAt),
   onTopUp:     e => console.log('Topped up:', e.amount),
   onClawback:  e => console.log('Clawback:', e.amount),
+  onCreated:            e => console.log('Created:', e.recipient),
+  onForceCancel:        e => console.log('Force-cancelled by:', e.recipient),
+  onRecipientTransfer:  e => console.log('Recipient transferred to:', e.newRecipient),
+  onOperatorSet:        e => console.log('Operator set:', e.operator),
+  onOperatorRevoke:     e => console.log('Operator revoked:', e.operator),
   onError:     err => console.warn('Polling error:', err),
   pollInterval:            3000,   // ms; default 5000
   maxBackoffMs:            30000,  // ms; default 60000
@@ -222,10 +270,17 @@ sub.unsubscribe();
 > **All event payload fields are decoded.** `src/events.ts`'s `dispatchEvent()` parses every
 > multi-field event from its tuple `ScVal`s: `onWithdraw` → `{ recipient, amount,
 > totalWithdrawn, remaining }`, `onCancel` → `{ sender, refundAmount, withdrawnSoFar }`,
-> `onPause` → `{ sender, pausedAt, withdrawable }`, `onTopUp` → `{ sender, amount, newBalance }`.
-> Single-field events are decoded from their bare scalar: `onResume` → `{ sender, resumedAt }`,
-> `onClawback` → `{ sender, amount }`. These fields are real values from the chain — no
-> placeholder `0`/`0n` values remain.
+> `onPause` → `{ sender, pausedAt, withdrawable }`, `onTopUp` → `{ sender, amount, newBalance }`,
+> `onCreated` → `{ sender, recipient, token, depositAmount, ratePerSecond, startTime, endTime }`,
+> `onForceCancel` → `{ recipient, payoutAmount, refundAmount }`. Single-field events are decoded
+> from their bare scalar: `onResume` → `{ sender, resumedAt }`, `onClawback` → `{ sender, amount }`,
+> `onRecipientTransfer` → `{ previousRecipient, newRecipient }`, `onOperatorSet` → `{ sender,
+> operator }`, `onOperatorRevoke` → `{ sender, operator }`. These fields are real values from the
+> chain — no placeholder `0`/`0n` values remain.
+>
+> `onRecipientTransfer`, `onForceCancel`, `onOperatorSet`, and `onOperatorRevoke` were previously
+> silently ignored — a live subscriber was never notified when the recipient role moved to
+> another address, a paused stream was force-cancelled, or an operator was delegated/revoked.
 
 The first poll seeds its start ledger from `getLatestLedger()` before calling `getEvents()` (Soroban
 RPC's `getEvents` rejects without a start ledger). If that seeding call itself fails, it's retried on
@@ -813,4 +868,32 @@ new Module49(config?: Module49Config)
 * `clearCache(): void` - Clears the internal lookup cache and metrics.
 * `getPerformanceMetrics(): Module49Metrics` - Returns real-time metrics (`totalProcessed`, `cacheHits`, `cacheMisses`, `hitRate`, `averageExecutionTimeMs`).
 
+---
+
+## `Module44` (Feature #44)
+
+Stream liquidity-risk / runway calculator implementing Feature #44. For each stream, computes the remaining `runwaySecs` until `endTime` and classifies it into a `LiquidityRiskLevel` (`'inactive' | 'critical' | 'warning' | 'healthy'`), so a dashboard can flag streams about to run out of scheduled balance. Uses the same LRU-memoized lookup pattern as `Module26`/`Module36`/`Module48`; actual speedup is workload-dependent (proportional to cache hit rate) — see `getPerformanceMetrics()` for a measured value, never a fixed assumed percentage.
+
+### Constructor
+
+```typescript
+new Module44(config?: Module44Config)
+```
+
+| Option | Type | Default | Notes |
+|--------|------|---------|-------|
+| `cacheSize` | `number` | `1000` | Max entries in memoization cache |
+| `enableOptimization` | `boolean` | `true` | Enables memoized lookup caching |
+| `batchChunkSize` | `number` | `50` | Stream chunk size for batch processing |
+| `criticalThresholdSecs` | `number` | `86400` (1 day) | Runway below this is `'critical'`. Throws if negative |
+| `warningThresholdSecs` | `number` | `604800` (7 days) | Runway below this (and at/above `criticalThresholdSecs`) is `'warning'`. Throws if not greater than `criticalThresholdSecs` |
+
+### Methods
+
+* `assessSingleItem(item: StreamRiskItem): StreamRiskAssessment` — Computes one stream's `runwaySecs` and `riskLevel`. A cancelled, paused, or zero-rate stream is `'inactive'` with `runwaySecs: 0`. An open-ended stream (`endTime === 0`) is `'healthy'` with `runwaySecs: null` — its runway isn't bounded by a schedule, only by the sender keeping the balance topped up.
+* `assessBatch(items: StreamRiskItem[]): StreamRiskAssessment[]` — Assesses an array of streams in `batchChunkSize` chunks.
+* `estimateTopUpNeeded(stream: StreamInfo, targetRunwaySecs: number, nowSec?: number): bigint` — Stroops needed via `top_up()` for the stream's runway to reach `targetRunwaySecs`. Returns `0n` if the stream is inactive/paused/cancelled, the target is non-positive, or the target is already met (including any open-ended stream, whose runway is treated as unbounded).
+* `clearCache(): void` — Clears the internal lookup cache and metrics.
+* `getPerformanceMetrics(): Module44Metrics` — Returns `totalAssessed`, `cacheHits`, `cacheMisses`, `averageExecutionTimeMs`, and `measuredSpeedupPercent` (a real measurement derived from this instance's own accumulated hit/miss timings, `null` until both have occurred at least once).
+
 

src/events.ts

@@ -15,10 +15,21 @@ import type {
   ResumeEvent,
   TopUpEvent,
   ClawbackEvent,
+  CreatedEvent,
+  ForceCancelEvent,
+  RecipientTransferEvent,
+  OperatorSetEvent,
+  OperatorRevokedEvent,
 } from './types/index.js';
 import { scValToI128, scValToU64, createRpcServer } from './soroban.js';
 
 // ── Event topic names (match symbol_short!() values in Rust) ─────────────────
+//
+// `created`, `force_cxl`, `xfer_rec`, `set_op` and `rm_op` are emitted by
+// contracts/stream/src/events.rs alongside the six handled below, but were
+// never wired into TOPIC/dispatchEvent — a subscriber was silently never
+// notified of a recipient transfer, a recipient-initiated force-cancel, or
+// an operator delegation/revocation (see #506).
 
 const TOPIC = {
   WITHDRAWN: 'withdrawn',
@@ -27,6 +38,11 @@ const TOPIC = {
   RESUMED:   'resumed',
   TOPPED_UP: 'topped_up',
   CLAWBACK:  'clawback',
+  CREATED:               'created',
+  FORCE_CANCELLED:       'force_cxl',
+  RECIPIENT_TRANSFERRED: 'xfer_rec',
+  OPERATOR_SET:          'set_op',
+  OPERATOR_REVOKED:      'rm_op',
 } as const;
 
 // ── Parser helpers ────────────────────────────────────────────────────────────
@@ -51,6 +67,10 @@ function u64Field(fields: xdr.ScVal[], index: number): number {
   return field ? Number(scValToU64(field)) : 0;
 }
 
+function addressFieldAt(fields: xdr.ScVal[], index: number): string {
+  return addressField(fields[index]);
+}
+
 /**
  * Decodes an address topic to its G.../C... string. `ScVal.address()?.accountId()`
  * returns the raw XDR PublicKey object, not a string — calling `.toString()`
@@ -307,6 +327,75 @@ export function dispatchEvent(
       handlers.onClawback(data);
       break;
     }
+
+    case TOPIC.CREATED: {
+      if (!handlers.onCreated) break;
+      // data: (recipient, token, deposit_amount: i128, rate_per_second: i128, start_time: u64, end_time: u64)
+      const fields = tupleFields(event.value);
+      const data: CreatedEvent = {
+        sender:         actor,
+        recipient:      addressFieldAt(fields, 0),
+        token:          addressFieldAt(fields, 1),
+        depositAmount:  i128Field(fields, 2),
+        ratePerSecond:  i128Field(fields, 3),
+        startTime:      u64Field(fields, 4),
+        endTime:        u64Field(fields, 5),
+        sequence,
+      };
+      handlers.onCreated(data);
+      break;
+    }
+
+    case TOPIC.FORCE_CANCELLED: {
+      if (!handlers.onForceCancel) break;
+      // data: (payout_amount: i128, refund_amount: i128) — mirrors cancel()'s
+      // (refund_amount, withdrawn_so_far) shape but recipient-initiated.
+      const fields = tupleFields(event.value);
+      const data: ForceCancelEvent = {
+        recipient:    actor,
+        payoutAmount: i128Field(fields, 0),
+        refundAmount: i128Field(fields, 1),
+        sequence,
+      };
+      handlers.onForceCancel(data);
+      break;
+    }
+
+    case TOPIC.RECIPIENT_TRANSFERRED: {
+      if (!handlers.onRecipientTransfer) break;
+      // data: new_recipient: address (bare scalar, not a tuple)
+      const data: RecipientTransferEvent = {
+        previousRecipient: actor,
+        newRecipient:      addressField(event.value),
+        sequence,
+      };
+      handlers.onRecipientTransfer(data);
+      break;
+    }
+
+    case TOPIC.OPERATOR_SET: {
+      if (!handlers.onOperatorSet) break;
+      // data: operator: address (bare scalar, not a tuple)
+      const data: OperatorSetEvent = {
+        sender:   actor,
+        operator: addressField(event.value),
+        sequence,
+      };
+      handlers.onOperatorSet(data);
+      break;
+    }
+
+    case TOPIC.OPERATOR_REVOKED: {
+      if (!handlers.onOperatorRevoke) break;
+      // data: operator: address (bare scalar, not a tuple)
+      const data: OperatorRevokedEvent = {
+        sender:   actor,
+        operator: addressField(event.value),
+        sequence,
+      };
+      handlers.onOperatorRevoke(data);
+      break;
+    }
   }
 
   return sequence;

src/index.ts

@@ -109,3 +109,12 @@ export type {
   Module49Metrics,
 } from './module49.js';
 
+export { Module44 } from './module44.js';
+export type {
+  Module44Config,
+  StreamRiskItem,
+  LiquidityRiskLevel,
+  StreamRiskAssessment,
+  Module44Metrics,
+} from './module44.js';
+