perf(batch-tx): parallelize batch simulation + test/doc coverage for indexer, adapters, client

CHANGELOG.md

@@ -10,6 +10,12 @@ All notable changes are documented here. Format based on [Keep a Changelog](http
 
 ### Performance
 - `FactoryModule.streamAddress()` now caches resolved stream→contract-address lookups in-memory, since the mapping is fixed at stream creation and never changes. Eliminates redundant RPC round trips on every `StreamsModule` read/write operation (`get`, `withdraw`, `cancel`, `pause`, `resume`, `topUp`, `clawback`) and on each page of `list()`, which previously re-resolved the same address for every stream on every call.
+- `buildBatchTransactions()` (the RPC-prepared batch path) now simulates all operations in a batch concurrently instead of one at a time, cutting the wall-clock time of an N-operation batch from N sequential RPC round trips to one.
+
+### Documentation
+- Added an API reference section for `GraphQLIndexer`, which was previously exported but undocumented.
+- Added a "Wallet Adapters" API reference section documenting `KeypairWalletAdapter`.
+- Documented `ConduitClient`'s `pauseStream()`, `unpauseStream()`, and `setWallet()` convenience methods in `docs/api.md`, and fixed `setWallet()`'s JSDoc block, which had been orphaned above `pauseStream()`/`unpauseStream()` and left `setWallet()` itself undocumented.
 
 ### Fixed
 - **Critical:** `FeeEstimator.estimateFee()` now uses `bigint` stroops instead of floating-point for fee representation, eliminating IEEE-754 precision loss. All monetary amounts in the SDK now consistently use bigint to avoid rounding errors.

docs/api.md

@@ -19,6 +19,13 @@ new ConduitClient(config: ConduitConfig)
 | `rpcUrl` | `string` | | Network default |
 | `factoryAddress` | `string` | | Deployed factory |
 | `governorAddress` | `string` | | Deployed governor |
+| `wallet` | `WalletAdapter` | | — |
+
+### Convenience methods
+
+* `pauseStream(streamId: string) → Promise<string>` — equivalent to `client.streams.pause(streamId)`.
+* `unpauseStream(streamId: string) → Promise<string>` — equivalent to `client.streams.resume(streamId)`.
+* `setWallet(wallet: WalletAdapter): void` — dynamically attach or change the active wallet adapter. Throws `UnsupportedChainError` if the wallet's `chainId` is on a different network than the client was configured for. See [Wallet Adapters](#wallet-adapters) below. Only propagates to `client.streams` — `client.factory` and `client.governor` are read-only and use `config.keypair` for simulation fee sourcing, so they are unaffected.
 
 ---
 
@@ -372,6 +379,100 @@ clearTokenDecimalsCache(); // force a fresh simulation on the next call
 
 ---
 
+## Wallet Adapters
+
+`WalletAdapter` is the interface `client.streams` signs transactions through — implement it to
+support any wallet. The SDK ships two implementations:
+
+### `KeypairWalletAdapter`
+
+Wraps a raw `@stellar/stellar-sdk` `Keypair` so `config.keypair` can be used through the same
+`WalletAdapter` interface as a browser or WalletConnect wallet. Constructed automatically by
+`ConduitClient`/`StreamsModule` when `config.keypair` is supplied and no `config.wallet` is given.
+
+```typescript
+new KeypairWalletAdapter(keypair: Keypair)
+```
+
+* `getPublicKey(): string` — the keypair's G-address.
+* `signTransaction(tx: Transaction | string, opts?: SignTransactionOptions): Promise<Transaction | string>` — signs and returns a `Transaction` instance as-is; for a raw XDR string, requires `opts.networkPassphrase` (throws otherwise) and returns signed XDR. `opts.accountToSign` is not applicable — a keypair only ever signs as itself.
+* `isConnected(): boolean` — always `true`.
+
+### `WalletConnectAdapter`
+
+Wraps a WalletConnect v2 session. See its JSDoc in `src/adapters/walletconnect.ts` for the full
+option set.
+
+---
+
+## `GraphQLIndexer`
+
+A client for a Conduit indexer's GraphQL endpoint — one-shot queries plus live subscriptions.
+
+```typescript
+new GraphQLIndexer(endpoint: string)
+```
+
+Throws if `endpoint` is empty.
+
+### `query(options) → Promise<unknown>`
+
+Issues a single GraphQL query as an HTTP POST and returns the parsed JSON response.
+
+| Field | Type | Required |
+|-------|------|----------|
+| `query` | `string` | ✓ |
+| `variables` | `Record<string, unknown>` | |
+| `headers` | `Record<string, string>` | |
+
+**Throws** if `query` is empty, or if the HTTP response is not `ok`.
+
+### `subscribe(options) → IndexerSubscription`
+
+Opens a live subscription. Prefers a `graphql-transport-ws` WebSocket connection derived from
+`endpoint` (`https://` → `wss://`, `http://` → `ws://`); when no `WebSocket` constructor is
+available (e.g. some non-browser, non-Node runtimes) it falls back to reading a
+`text/event-stream` HTTP response and parsing its `data:` lines.
+
+| Field | Type | Required |
+|-------|------|----------|
+| `query` | `string` | ✓ |
+| `variables` | `Record<string, unknown>` | |
+| `headers` | `Record<string, string>` | |
+| `onData` | `(data: unknown) => void` | ✓ |
+| `onError` | `(error: Error) => void` | |
+
+Returns `{ unsubscribe(): void }`. Calling `unsubscribe()` is idempotent — it sends a
+`complete` message (WebSocket transport) or aborts the underlying fetch (SSE fallback) and is
+safe to call more than once.
+
+### `getSubscriptionCount() → number`
+
+Number of subscriptions currently active on this indexer instance.
+
+### `cleanup(): void`
+
+Unsubscribes every active subscription and marks the indexer destroyed — subsequent calls to
+`query()` or `subscribe()` throw.
+
+```typescript
+import { GraphQLIndexer } from '@conduit-protocol/sdk';
+
+const indexer = new GraphQLIndexer('https://indexer.streamfi.io/graphql');
+
+const sub = indexer.subscribe({
+  query: 'subscription { streamUpdated(id: "1") { id withdrawn } }',
+  onData: (data) => console.log(data),
+  onError: (err) => console.error(err),
+});
+
+// later
+sub.unsubscribe();
+indexer.cleanup();
+```
+
+---
+
 ## Fluent Builder API
 
 The SDK provides `StreamBuilder` and `ConduitBatcher` to construct and execute stream operations fluently and in batches.

src/client.ts

@@ -82,26 +82,6 @@ export class ConduitClient {
     this.governor = new GovernorModule(this.config);
   }
 
-  /**
-   * Dynamically attach or change the active wallet adapter.
-   *
-   * Validates that the new wallet's chain matches the client's configured
-   * network before accepting it — prevents silent cross-chain mismatches
-   * from reaching the smart contract (fixes #157).
-   *
-   * **Wallet propagation contract:**
-   * - {@link StreamsModule}: Updated immediately — all subsequent stream
-   *   operations (create, withdraw, cancel, etc.) use the new wallet.
-   * - {@link FactoryModule}: NOT updated — this module is read-only and
-   *   uses `config.keypair` for simulation fee sourcing. It does not hold
-   *   a wallet reference and is unaffected by `setWallet()`.
-   * - {@link GovernorModule}: NOT updated — this module is read-only and
-   *   uses `config.keypair` for simulation fee sourcing. It does not hold
-   *   a wallet reference and is unaffected by `setWallet()`.
-   *
-   * @throws {UnsupportedChainError} if the wallet's `chainId` is on a
-   *   different network than the one this client was initialised with.
-   */
   /**
    * Pause an active stream (sender only).
    *
@@ -128,6 +108,26 @@ export class ConduitClient {
     return this.streams.resume(streamId);
   }
 
+  /**
+   * Dynamically attach or change the active wallet adapter.
+   *
+   * Validates that the new wallet's chain matches the client's configured
+   * network before accepting it — prevents silent cross-chain mismatches
+   * from reaching the smart contract (fixes #157).
+   *
+   * **Wallet propagation contract:**
+   * - {@link StreamsModule}: Updated immediately — all subsequent stream
+   *   operations (create, withdraw, cancel, etc.) use the new wallet.
+   * - {@link FactoryModule}: NOT updated — this module is read-only and
+   *   uses `config.keypair` for simulation fee sourcing. It does not hold
+   *   a wallet reference and is unaffected by `setWallet()`.
+   * - {@link GovernorModule}: NOT updated — this module is read-only and
+   *   uses `config.keypair` for simulation fee sourcing. It does not hold
+   *   a wallet reference and is unaffected by `setWallet()`.
+   *
+   * @throws {UnsupportedChainError} if the wallet's `chainId` is on a
+   *   different network than the one this client was initialised with.
+   */
   setWallet(wallet: WalletAdapter): void {
     assertWalletNetworkMatch(wallet, this.config.network);
     this.config.wallet = wallet;

src/tests/batch-tx-rpc-simulation.test.ts

@@ -0,0 +1,138 @@
+/**
+ * buildBatchTransactions() — the RPC-prepared path (context.rpcUrl set) —
+ * had no dedicated tests before this file. It simulated every batched
+ * operation one at a time in a `for` loop even though each simulation is an
+ * independent RPC round trip, so N operations meant N sequential round
+ * trips. These tests cover that path and lock in the concurrent behaviour.
+ */
+
+import { describe, it, expect, vi, beforeEach } from 'vitest';
+
+const { mockSimulateTransaction, mockGetAccount, mockAssembleTransaction } = vi.hoisted(() => ({
+  mockSimulateTransaction: vi.fn(),
+  mockGetAccount: vi.fn(),
+  mockAssembleTransaction: vi.fn(),
+}));
+
+vi.mock('@stellar/stellar-sdk', async () => {
+  const actual = await vi.importActual('@stellar/stellar-sdk');
+  return {
+    ...actual,
+    SorobanRpc: {
+      ...(actual as any).SorobanRpc,
+      Server: vi.fn().mockImplementation(function MockServer() {
+        return {
+          simulateTransaction: mockSimulateTransaction,
+          getAccount: mockGetAccount,
+        };
+      }),
+      Api: (actual as any).SorobanRpc.Api,
+      assembleTransaction: mockAssembleTransaction,
+    },
+  };
+});
+
+import { buildBatchTransactions, BatchBuildError, type BatchTransactionContext } from '../batch-tx.js';
+
+const CONTRACT_ID = 'CAAQCAIBAEAQCAIBAEAQCAIBAEAQCAIBAEAQCAIBAEAQCAIBAEAQC526';
+const SOURCE = 'GAAQCAIBAEAQCAIBAEAQCAIBAEAQCAIBAEAQCAIBAEAQCAIBAEAQDZ7H';
+
+const CONTEXT: BatchTransactionContext = {
+  contractId: CONTRACT_ID,
+  sourceAccount: SOURCE,
+  network: 'testnet',
+  sequence: '100',
+  rpcUrl: 'http://localhost:8000/soroban/rpc',
+};
+
+const SIMULATION_OK = { result: { retval: {} }, transactionData: {} };
+
+beforeEach(() => {
+  mockSimulateTransaction.mockReset();
+  mockGetAccount.mockReset();
+  // assembleTransaction(tx, sim).build() just needs to hand back something
+  // with a real toXDR() — the real, unmocked Transaction built by
+  // buildBatchTransactions already has one.
+  mockAssembleTransaction.mockReset().mockImplementation((tx: unknown) => ({ build: () => tx }));
+});
+
+describe('buildBatchTransactions() — RPC-prepared path', () => {
+  it('simulates every operation concurrently, not one at a time', async () => {
+    const resolvers: Array<(v: unknown) => void> = [];
+    mockSimulateTransaction.mockImplementation(
+      () => new Promise((resolve) => resolvers.push(resolve)),
+    );
+
+    const operations = [{ method: 'a' }, { method: 'b' }, { method: 'c' }];
+    const promise = buildBatchTransactions(operations, CONTEXT);
+    promise.catch(() => {});
+
+    // Flush pending microtasks without resolving any simulation. A
+    // sequential implementation would have called simulateTransaction() only
+    // for the first operation by this point; a concurrent one calls it for
+    // every operation up front.
+    await Promise.resolve();
+    await Promise.resolve();
+    await Promise.resolve();
+
+    expect(mockSimulateTransaction).toHaveBeenCalledTimes(3);
+
+    resolvers.forEach((resolve) => resolve(SIMULATION_OK));
+    const built = await promise;
+
+    expect(built.map((b) => b.index)).toEqual([0, 1, 2]);
+    expect(built.every((b) => b.prepared)).toBe(true);
+    expect(built.map((b) => b.method)).toEqual(['a', 'b', 'c']);
+  });
+
+  it('fetches the sequence once via getAccount when none is supplied, and derives per-op sequences from it', async () => {
+    mockGetAccount.mockResolvedValue({ sequenceNumber: () => '200' });
+    mockSimulateTransaction.mockResolvedValue(SIMULATION_OK);
+
+    const { rpcUrl, contractId, sourceAccount, network } = CONTEXT;
+    const built = await buildBatchTransactions(
+      [{ method: 'a' }, { method: 'b' }],
+      {
+        contractId,
+        sourceAccount,
+        ...(rpcUrl !== undefined ? { rpcUrl } : {}),
+        ...(network !== undefined ? { network } : {}),
+      },
+    );
+
+    expect(mockGetAccount).toHaveBeenCalledTimes(1);
+    expect(built).toHaveLength(2);
+  });
+
+  it('surfaces a simulation error as a BatchBuildError naming the failing operation', async () => {
+    mockSimulateTransaction.mockResolvedValue({ error: 'contract trapped' });
+
+    await expect(
+      buildBatchTransactions([{ method: 'boom' }], CONTEXT),
+    ).rejects.toThrow(BatchBuildError);
+    await expect(
+      buildBatchTransactions([{ method: 'boom' }], CONTEXT),
+    ).rejects.toThrow(/operation 0 \(boom\)/);
+  });
+
+  it('rejects an operation missing a method name without ever calling the RPC', async () => {
+    mockSimulateTransaction.mockResolvedValue(SIMULATION_OK);
+
+    await expect(
+      buildBatchTransactions([{ method: '' }], CONTEXT),
+    ).rejects.toThrow(/missing a method name/);
+  });
+
+  it('falls back to the offline sync path when no rpcUrl is configured', async () => {
+    const { contractId, sourceAccount, network, sequence } = CONTEXT;
+    const built = await buildBatchTransactions([{ method: 'ping' }], {
+      contractId,
+      sourceAccount,
+      ...(network !== undefined ? { network } : {}),
+      ...(sequence !== undefined ? { sequence } : {}),
+    });
+
+    expect(mockSimulateTransaction).not.toHaveBeenCalled();
+    expect(built[0]!.prepared).toBe(false);
+  });
+});

src/tests/client-pause-unpause.test.ts

@@ -0,0 +1,56 @@
+/**
+ * ConduitClient.pauseStream()/unpauseStream() are thin convenience wrappers
+ * around client.streams.pause()/resume() — StreamsModule's own pause/resume
+ * are covered in streams-success.test.ts, but nothing exercised the
+ * ConduitClient-level methods themselves.
+ */
+
+import { describe, it, expect, vi } from 'vitest';
+import { Keypair } from '@stellar/stellar-sdk';
+import { ConduitClient } from '../client.js';
+
+function makeClient(): ConduitClient {
+  return new ConduitClient({
+    network: 'testnet',
+    keypair: Keypair.random(),
+    factoryAddress: 'CDRIPFACTORY1234567890123456789012345678901234567890123',
+  });
+}
+
+describe('ConduitClient.pauseStream()', () => {
+  it('delegates to client.streams.pause() with the given streamId', async () => {
+    const client = makeClient();
+    const pauseSpy = vi.spyOn(client.streams, 'pause').mockResolvedValue('tx-hash-pause');
+
+    const result = await client.pauseStream('42');
+
+    expect(pauseSpy).toHaveBeenCalledWith('42');
+    expect(result).toBe('tx-hash-pause');
+  });
+
+  it('propagates a rejection from client.streams.pause()', async () => {
+    const client = makeClient();
+    vi.spyOn(client.streams, 'pause').mockRejectedValue(new Error('pause failed'));
+
+    await expect(client.pauseStream('42')).rejects.toThrow('pause failed');
+  });
+});
+
+describe('ConduitClient.unpauseStream()', () => {
+  it('delegates to client.streams.resume() with the given streamId', async () => {
+    const client = makeClient();
+    const resumeSpy = vi.spyOn(client.streams, 'resume').mockResolvedValue('tx-hash-resume');
+
+    const result = await client.unpauseStream('42');
+
+    expect(resumeSpy).toHaveBeenCalledWith('42');
+    expect(result).toBe('tx-hash-resume');
+  });
+
+  it('propagates a rejection from client.streams.resume()', async () => {
+    const client = makeClient();
+    vi.spyOn(client.streams, 'resume').mockRejectedValue(new Error('resume failed'));
+
+    await expect(client.unpauseStream('42')).rejects.toThrow('resume failed');
+  });
+});