Merge pull request #509 from lekescrew22/feat/batch-onprogress-callback

feat: expose onProgress callback on batchWithdraw and bulkCreateStreams

Author: Chuks-coderr

.changeset/batch-progress-callback.md

@@ -0,0 +1,10 @@
+---
+"@sorostream/sdk": minor
+---
+
+Add an optional `onProgress` callback to the batch operations so callers can render incremental progress:
+
+- `batchWithdraw(streamIds, batchSize?, onProgress?)` — fires after each chunk (transaction) completes, whether it succeeded or failed.
+- `bulkCreateStreams(rows, { ..., onProgress })` — fires after each chunk, or after each individual row for mixed-token chunks.
+
+Both callbacks receive `{ completed, total, processedIds }`, where `completed`/`total` count individual streams/rows and `processedIds` lists the stream IDs (withdraw) or recipient addresses (create) handled by the step that just finished. The new `BatchProgress` type is exported from `@sorostream/sdk`, `@sorostream/sdk/core`, and `@sorostream/sdk/batch`.

README.md

@@ -121,6 +121,7 @@ import type {
   BatchCancelResult,
   BatchWithdrawResult,
   BatchWithdrawPartialResult,
+  BatchProgress,
   BulkCreateOptions,
   BulkCreateResult,
   CancelStreamParams,
@@ -2473,6 +2474,9 @@ export class SoroStreamClient<TEventData = Record<string, unknown>> {
    * @param streamIds - Stream IDs to withdraw from.
    * @param batchSize - Maximum operations per transaction (default 8). Chunks
    *   are still attempted together, but a chunk failure is recorded per-stream.
+   * @param onProgress - Optional callback fired after each chunk completes
+   *   (success or failure), reporting cumulative `{ completed, total,
+   *   processedIds }` so callers can render incremental progress.
    * @returns `{ successes, failures }` — IDs that were successfully withdrawn
    *   and the IDs+errors that were not.
    *
@@ -2483,17 +2487,27 @@ export class SoroStreamClient<TEventData = Record<string, unknown>> {
    *
    * @example
    * ```ts
-   * const { successes, failures } = await client.batchWithdraw(["1", "2", "3"]);
+   * const { successes, failures } = await client.batchWithdraw(
+   *   ["1", "2", "3"],
+   *   2,
+   *   ({ completed, total }) => console.log(`Withdrawn ${completed}/${total}`),
+   * );
    * if (failures.length) {
    *   console.warn("Some withdrawals failed:", failures);
    * }
    * console.log("Withdrawn:", successes);
    * ```
    */
-  async batchWithdraw(streamIds: string[], batchSize = 8): Promise<BatchWithdrawPartialResult> {
+  async batchWithdraw(
+    streamIds: string[],
+    batchSize = 8,
+    onProgress?: (progress: BatchProgress) => void,
+  ): Promise<BatchWithdrawPartialResult> {
     const successes: string[] = [];
     const failures: { id: string; error: Error }[] = [];
     const recipient = await this.requireWalletAdapter().getPublicKey();
+    const total = streamIds.length;
+    let completed = 0;
 
     for (let i = 0; i < streamIds.length; i += batchSize) {
       const chunk = streamIds.slice(i, i + batchSize);
@@ -2522,6 +2536,9 @@ export class SoroStreamClient<TEventData = Record<string, unknown>> {
           failures.push({ id, error: err instanceof Error ? err : new Error(String(err)) });
         }
       }
+
+      completed += chunk.length;
+      onProgress?.({ completed, total, processedIds: chunk });
     }
 
     return { successes, failures };
@@ -4478,14 +4495,22 @@ export class SoroStreamClient<TEventData = Record<string, unknown>> {
    * @param options.token - Default SAC token contract address for rows that omit `token`.
    * @param options.autoRenew - Whether created streams auto-renew (default false).
    * @param options.batchSize - Maximum operations per transaction (default 8).
+   * @param options.onProgress - Optional callback fired after each chunk (or
+   *   row, for mixed-token chunks) is submitted, reporting cumulative
+   *   `{ completed, total, processedIds }` so callers can render incremental
+   *   progress. Fires on failures too — a failed row still counts as processed.
    * @returns `{ batches }` — one entry per submitted transaction, each with its `txHash` and the resulting `streamIds`.
    * @throws {BulkCreatePartialError} If one or more rows fail; carries `successfulBatches` and `failedSlots`.
    * @throws {TransactionFailedError} If a submitted transaction is rejected (wrapped into `failedSlots` rather than thrown directly).
    *
    * @example
    * ```ts
    * try {
-   *   const { batches } = await client.bulkCreateStreams(rows, { token: usdc });
+   *   const { batches } = await client.bulkCreateStreams(rows, {
+   *     token: usdc,
+   *     onProgress: ({ completed, total }) =>
+   *       console.log(`Created ${completed}/${total} streams`),
+   *   });
    * } catch (err) {
    *   if (err instanceof BulkCreatePartialError) {
    *     console.error(`${err.failedSlots.length} stream(s) failed:`, err.failedSlots);
@@ -4502,6 +4527,9 @@ export class SoroStreamClient<TEventData = Record<string, unknown>> {
       const defaultToken = options.token;
       const autoRenew = options.autoRenew ?? false;
       const batchSize = options.batchSize ?? 8;
+      const onProgress = options.onProgress;
+      const total = rows.length;
+      let completed = 0;
 
       // Validate cliff for all rows before submitting anything
       for (const row of rows) {
@@ -4543,6 +4571,9 @@ export class SoroStreamClient<TEventData = Record<string, unknown>> {
             } catch (error) {
               failedSlots.push({ index: i + j, row, error });
             }
+
+            completed += 1;
+            onProgress?.({ completed, total, processedIds: [row.recipient] });
           }
         } else {
           try {
@@ -4570,6 +4601,9 @@ export class SoroStreamClient<TEventData = Record<string, unknown>> {
               failedSlots.push({ index: i + j, row, error });
             });
           }
+
+          completed += chunk.length;
+          onProgress?.({ completed, total, processedIds: chunk.map((r) => r.recipient) });
         }
       }
 

src/SoroStreamClient.ts

@@ -27,6 +27,7 @@ export type {
   BatchCancelResult,
   BatchWithdrawResult,
   BatchWithdrawPartialResult,
+  BatchProgress,
 } from './types.js';
 export { BulkCreatePartialError } from './errors.js';
 export type { BulkCreateFailedSlot } from './errors.js';

src/batch.ts

@@ -129,6 +129,7 @@ export type {
   BatchCancelResult,
   BatchWithdrawResult,
   BatchWithdrawPartialResult,
+  BatchProgress,
   TokenAggregate,
   MultisigSigner,
   StreamEvent,

src/core.ts

@@ -197,6 +197,7 @@ export type {
   BatchCancelResult,
   BatchWithdrawResult,
   BatchWithdrawPartialResult,
+  BatchProgress,
   TokenAggregate,
   MultisigSigner,
   StreamEvent,

src/index.ts

@@ -545,6 +545,12 @@ export interface BulkCreateOptions {
   autoRenew?: boolean;
   /** Max operations per transaction (default 8). */
   batchSize?: number;
+  /**
+   * Optional callback fired after each chunk (or row, for mixed-token chunks)
+   * is submitted. Receives cumulative progress so callers can render
+   * incremental progress while the bulk create is in flight.
+   */
+  onProgress?: (progress: BatchProgress) => void;
 }
 
 /** Result of one batch within a bulk create. */
@@ -577,6 +583,24 @@ export interface BatchWithdrawPartialResult {
   failures: { id: string; error: Error }[];
 }
 
+/**
+ * Progress reported after each completed step of a batch operation, via the
+ * `onProgress` callback of {@link batchWithdraw} and {@link bulkCreateStreams}.
+ * `completed`/`total` count individual items (streams or rows), so callers can
+ * render incremental progress bars while the operation is in flight.
+ */
+export interface BatchProgress {
+  /** Number of items (streams or rows) processed so far, including this step. */
+  completed: number;
+  /** Total number of items to process. */
+  total: number;
+  /**
+   * Identifiers of the items handled by the step that just completed:
+   * stream IDs for `batchWithdraw`, recipient addresses for `bulkCreateStreams`.
+   */
+  processedIds: string[];
+}
+
 /** Per-token aggregate of a set of streams. */
 export interface TokenAggregate {
   token: string;

src/types.ts

@@ -13,6 +13,7 @@ import type {
   Stream,
   WalletAdapter,
   BulkStreamRow,
+  BatchProgress,
   PriceFeedAdapter,
   FeeBumpOptions,
 } from '../src/types.js';
@@ -917,6 +918,32 @@ describe('SoroStreamClient batchWithdraw', () => {
     expect(result.failures).toEqual([]);
   });
 
+  it('fires onProgress after each chunk with cumulative counts', async () => {
+    const progress: BatchProgress[] = [];
+    await client.batchWithdraw(['1', '2', '3', '4', '5'], 2, (p) => progress.push(p));
+
+    expect(progress).toEqual([
+      { completed: 2, total: 5, processedIds: ['1', '2'] },
+      { completed: 4, total: 5, processedIds: ['3', '4'] },
+      { completed: 5, total: 5, processedIds: ['5'] },
+    ]);
+  });
+
+  it('fires onProgress for failed chunks too', async () => {
+    vi.spyOn(client, 'executeBatch')
+      .mockResolvedValueOnce('tx1')
+      .mockRejectedValueOnce(new Error('chunk failed'));
+
+    const progress: BatchProgress[] = [];
+    const result = await client.batchWithdraw(['1', '2', '3', '4'], 2, (p) => progress.push(p));
+
+    expect(progress).toEqual([
+      { completed: 2, total: 4, processedIds: ['1', '2'] },
+      { completed: 4, total: 4, processedIds: ['3', '4'] },
+    ]);
+    expect(result.failures).toHaveLength(2);
+  });
+
   it('records failures when executeBatch rejects for a chunk', async () => {
     vi.spyOn(client, 'executeBatch')
       .mockResolvedValueOnce('tx1')
@@ -997,6 +1024,50 @@ describe('SoroStreamClient bulkCreateStreams', () => {
     expect(result.batches).toHaveLength(1);
     expect(result.batches[0]!.streamIds).toEqual([]);
   });
+
+  it('fires onProgress after each chunk with cumulative counts', async () => {
+    vi.spyOn(client, 'getStreamsBySender').mockResolvedValue([]);
+
+    const rows: BulkStreamRow[] = [
+      { recipient: TEST_PK, amount: 100n, durationSeconds: 3600 },
+      { recipient: TEST_PK, amount: 200n, durationSeconds: 7200 },
+      { recipient: TEST_PK, amount: 300n, durationSeconds: 3600 },
+    ];
+
+    const progress: BatchProgress[] = [];
+    await client.bulkCreateStreams(rows, {
+      token: TEST_TOKEN,
+      batchSize: 2,
+      onProgress: (p) => progress.push(p),
+    });
+
+    expect(progress).toEqual([
+      { completed: 2, total: 3, processedIds: [TEST_PK, TEST_PK] },
+      { completed: 3, total: 3, processedIds: [TEST_PK] },
+    ]);
+  });
+
+  it('fires onProgress per row when a chunk has mixed tokens', async () => {
+    vi.spyOn(client, 'getStreamsBySender').mockResolvedValue([]);
+    vi.spyOn(client, 'buildAndSubmit').mockResolvedValue({ txHash: 'tx_mixed', ledger: 0 });
+
+    const otherToken = VALID_CONTRACT;
+    const rows: BulkStreamRow[] = [
+      { recipient: TEST_PK, amount: 100n, durationSeconds: 3600, token: TEST_TOKEN },
+      { recipient: TEST_PK, amount: 200n, durationSeconds: 7200, token: otherToken },
+    ];
+
+    const progress: BatchProgress[] = [];
+    await client.bulkCreateStreams(rows, {
+      token: TEST_TOKEN,
+      onProgress: (p) => progress.push(p),
+    });
+
+    expect(progress).toEqual([
+      { completed: 1, total: 2, processedIds: [TEST_PK] },
+      { completed: 2, total: 2, processedIds: [TEST_PK] },
+    ]);
+  });
 });
 
 // ── Pre-flight validation tests (Issue 2) ────────────────────────────────────