feat: add createBatchStreams for bulk stream creation

src/streams.ts

@@ -15,6 +15,8 @@ import type {
   Subscription,
   BatchWithdrawItem,
   BatchWithdrawResult,
+  BatchCreateStreamResult,
+  StreamConfig,
   StreamOperation,
   FeeEstimate,
 } from './types/index.js';
@@ -33,6 +35,8 @@ import {
   DEFAULT_CONFIRMATION_POLL_INTERVAL_MS,
   createRpcServer,
 } from './soroban.js';
+import { buildBatchTransactions } from './batch-tx.js';
+import type { BatchTransactionContext } from './batch-tx.js';
 import { FactoryModule } from './factory.js';
 import { ConduitError, RateLimitError, StreamErrorCode } from './errors.js';
 
@@ -309,6 +313,154 @@ export class StreamsModule {
     });
   }
 
+  /**
+   * Create multiple streams in bulk.
+   *
+   * Soroban permits only one invoke_host_function operation per transaction
+   * (see {@link buildBatchTransactions}), so this builds and submits one
+   * transaction per config. Each transaction consumes the sender's next
+   * sequence number in order, so every config remains independently
+   * submittable regardless of how many configs are in the batch — unlike
+   * fetching a fresh account/sequence per config, which would hand out the
+   * same sequence number to more than one transaction once submitted out of
+   * order.
+   *
+   * Per-config client-side validation failures do not block or roll back the
+   * rest of the batch — each config is reported independently by its
+   * original index, mirroring {@link batchWithdraw}. Note this isolation
+   * does not extend to on-chain simulation: {@link buildBatchTransactions}
+   * simulates every config's transaction in parallel via `Promise.all`, so
+   * one config's simulation being rejected currently fails building the
+   * whole batch, not just that config.
+   */
+  async createBatchStreams(configs: StreamConfig[]): Promise<BatchCreateStreamResult[]> {
+    this._ensureCanMutate();
+    if (!Array.isArray(configs) || configs.length === 0) return [];
+
+    const senderAddr = await this._getSenderAddress();
+    const factoryId  = this.config.factoryAddress ?? '';
+
+    const built = await Promise.all(configs.map(async (params, index) => {
+      try {
+        const {
+          recipient, token, depositAmount,
+          durationSeconds, ratePerSecond,
+          startTime, clawbackEnabled = false,
+        } = params;
+
+        if (!recipient || typeof recipient !== 'string' || !recipient.trim()) {
+          throw new Error('Invalid recipient address: must be a non-empty string');
+        }
+        if (!token || typeof token !== 'string' || !token.trim()) {
+          throw new Error('Invalid token address: must be a non-empty string');
+        }
+        if (!depositAmount || typeof depositAmount !== 'string' || !depositAmount.trim()) {
+          throw new Error('Invalid deposit amount: must be a non-empty string');
+        }
+        if (durationSeconds !== undefined && (typeof durationSeconds !== 'number' || durationSeconds <= 0)) {
+          throw new Error('Invalid durationSeconds: must be a positive number');
+        }
+        if (ratePerSecond !== undefined && (typeof ratePerSecond !== 'string' || !ratePerSecond.trim())) {
+          throw new Error('Invalid ratePerSecond: must be a non-empty string');
+        }
+        if (!durationSeconds && !ratePerSecond) {
+          throw new Error('Either durationSeconds or ratePerSecond must be provided');
+        }
+
+        const decimals = await getTokenDecimals(this.rpcUrl, this.passphrase, senderAddr, token);
+        const depositStroops = toStroops(depositAmount, decimals);
+        const rateStroops    = ratePerSecond
+          ? BigInt(ratePerSecond)
+          : calculateRate(depositAmount, durationSeconds!, decimals);
+        const start = startTime ?? Math.floor(Date.now() / 1000);
+        const end   = durationSeconds ? start + durationSeconds : 0;
+
+        const args = [
+          new Address(senderAddr).toScVal(),
+          new Address(recipient).toScVal(),
+          new Address(token).toScVal(),
+          nativeToScVal(depositStroops, { type: 'i128' }),
+          nativeToScVal(rateStroops,    { type: 'i128' }),
+          nativeToScVal(start,          { type: 'u64'  }),
+          nativeToScVal(end,            { type: 'u64'  }),
+          boolToScVal(clawbackEnabled),
+        ];
+        return { index, args, error: undefined as string | undefined };
+      } catch (err) {
+        return { index, args: undefined, error: err instanceof Error ? err.message : String(err) };
+      }
+    }));
+
+    const results = new Map<number, BatchCreateStreamResult>();
+    for (const b of built) {
+      if (b.error !== undefined) {
+        results.set(b.index, { index: b.index, success: false, error: b.error });
+      }
+    }
+
+    const validEntries = built.filter(
+      (b): b is { index: number; args: xdr.ScVal[]; error: undefined } => b.args !== undefined,
+    );
+
+    if (validEntries.length > 0) {
+      const context: BatchTransactionContext = {
+        contractId: factoryId,
+        sourceAccount: senderAddr,
+        networkPassphrase: this.passphrase,
+        rpcUrl: this.rpcUrl,
+      };
+
+      let builtTxs: Awaited<ReturnType<typeof buildBatchTransactions>>;
+      try {
+        const operations = validEntries.map(b => ({ method: 'create_stream', args: b.args }));
+        builtTxs = await buildBatchTransactions(operations, context);
+      } catch (err) {
+        const message = err instanceof Error ? err.message : String(err);
+        for (const b of validEntries) {
+          results.set(b.index, { index: b.index, success: false, error: message });
+        }
+        builtTxs = [];
+      }
+
+      const server = this._server();
+      const settled = await Promise.allSettled(
+        builtTxs.map(async (bt) => {
+          const tx      = new Transaction(bt.xdr, this.passphrase);
+          const signed  = await this._signTx(tx);
+          const { hash: txHash, returnValue } = await this._sendAndPoll(server, signed);
+          if (!returnValue) {
+            throw new Error(`Transaction ${txHash} succeeded but returned no value`);
+          }
+          const streamId      = scValToU64(returnValue);
+          const streamAddress = await this._factory.streamAddress(streamId) ?? '';
+          return { streamId, streamAddress, txHash };
+        }),
+      );
+
+      settled.forEach((result, i) => {
+        const originalIndex = validEntries[i]!.index;
+        if (result.status === 'fulfilled') {
+          results.set(originalIndex, {
+            index: originalIndex,
+            success: true,
+            streamId: result.value.streamId,
+            streamAddress: result.value.streamAddress,
+            txHash: result.value.txHash,
+          });
+        } else {
+          const err = result.reason;
+          results.set(originalIndex, {
+            index: originalIndex,
+            success: false,
+            error: err instanceof Error ? err.message : String(err),
+          });
+        }
+      });
+    }
+
+    return configs.map((_, i) => results.get(i)!);
+  }
+
   /** Cancel the stream (sender only). Settles all balances atomically. */
   async cancel(streamId: bigint | string): Promise<string> {
     this._ensureCanMutate();