From fdccd584d2a92e3a1b274779e37dce7152127efe Mon Sep 17 00:00:00 2001
From: flourishbar <you@example.com>
Date: Tue, 28 Jul 2026 00:02:39 +0100
Subject: [PATCH] feat: implement API response streaming for large datasets
 (#768)

---
 .../__tests__/streamingService.test.ts        | 192 +++++++++
 app/services/batchTransactionService.ts       |  62 +++
 .../hooks/__tests__/useStreamingList.test.ts  | 156 ++++++++
 app/services/hooks/useExportStream.ts         | 180 +++++++++
 app/services/hooks/useStreamingList.ts        | 132 ++++++
 app/services/streamingService.ts              | 375 ++++++++++++++++++
 backend/server.ts                             | 164 +++++++-
 .../accountingExportService.streaming.test.ts | 195 +++++++++
 .../billing/accountingExportService.ts        | 155 ++++++++
 backend/services/index.ts                     |  31 +-
 .../shared/__tests__/streaming.test.ts        | 187 +++++++++
 backend/services/shared/sseEmitter.ts         | 161 ++++++++
 backend/services/shared/streaming.ts          | 227 +++++++++++
 .../subscription/subscriptionEventStore.ts    |  79 ++++
 developer-portal/docs/api-reference.md        |  34 ++
 developer-portal/docs/streaming-api.md        | 228 +++++++++++
 docs/streaming-architecture.md                | 147 +++++++
 17 files changed, 2701 insertions(+), 4 deletions(-)
 create mode 100644 app/services/__tests__/streamingService.test.ts
 create mode 100644 app/services/hooks/__tests__/useStreamingList.test.ts
 create mode 100644 app/services/hooks/useExportStream.ts
 create mode 100644 app/services/hooks/useStreamingList.ts
 create mode 100644 app/services/streamingService.ts
 create mode 100644 backend/services/billing/__tests__/accountingExportService.streaming.test.ts
 create mode 100644 backend/services/shared/__tests__/streaming.test.ts
 create mode 100644 backend/services/shared/sseEmitter.ts
 create mode 100644 backend/services/shared/streaming.ts
 create mode 100644 developer-portal/docs/streaming-api.md
 create mode 100644 docs/streaming-architecture.md

diff --git a/app/services/__tests__/streamingService.test.ts b/app/services/__tests__/streamingService.test.ts
new file mode 100644
index 00000000..40b3ac95
--- /dev/null
+++ b/app/services/__tests__/streamingService.test.ts
@@ -0,0 +1,192 @@
+/**
+ * Issue #768 – Tests for app/services/streamingService.ts
+ *
+ * Uses Jest mocks for `fetch` and `ReadableStream`.
+ */
+
+import {
+  fetchCursorPage,
+  collectAllPages,
+  streamNdjson,
+  subscribeToSse,
+  getClientMemoryStats,
+} from '../streamingService';
+import type { CursorPage } from '../streamingService';
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Helpers
+// ─────────────────────────────────────────────────────────────────────────────
+
+function mockFetchJson(body: unknown, status = 200): void {
+  global.fetch = jest.fn().mockResolvedValueOnce({
+    ok: status >= 200 && status < 300,
+    status,
+    statusText: status === 200 ? 'OK' : 'Error',
+    json: () => Promise.resolve(body),
+    text: () => Promise.resolve(JSON.stringify(body)),
+    body: null,
+  } as unknown as Response);
+}
+
+function mockFetchNdjson(lines: string[], status = 200): void {
+  const text = lines.join('\n') + '\n';
+  global.fetch = jest.fn().mockResolvedValueOnce({
+    ok: status >= 200 && status < 300,
+    status,
+    statusText: status === 200 ? 'OK' : 'Error',
+    text: () => Promise.resolve(text),
+    body: null, // triggers text fallback path
+  } as unknown as Response);
+}
+
+function mockFetchSse(blocks: string[], status = 200): void {
+  // Build SSE text with double-newline separators
+  const text = blocks.join('\n\n') + '\n\n';
+  global.fetch = jest.fn().mockResolvedValueOnce({
+    ok: status >= 200 && status < 300,
+    status,
+    statusText: status === 200 ? 'OK' : 'Error',
+    text: () => Promise.resolve(text),
+    body: null, // triggers SSE text fallback – we test subscribeToSse via handler verification
+  } as unknown as Response);
+}
+
+beforeEach(() => {
+  jest.resetAllMocks();
+});
+
+// ─────────────────────────────────────────────────────────────────────────────
+// fetchCursorPage
+// ─────────────────────────────────────────────────────────────────────────────
+
+describe('fetchCursorPage', () => {
+  it('fetches the first page without a cursor', async () => {
+    const page: CursorPage<{ id: string }> = {
+      items: [{ id: 'a' }, { id: 'b' }],
+      nextCursor: 'cursor-2',
+      total: 10,
+      pageSize: 2,
+    };
+    mockFetchJson(page);
+
+    const result = await fetchCursorPage<{ id: string }>('/api/items', { limit: 2 });
+    expect(result.items).toHaveLength(2);
+    expect(result.nextCursor).toBe('cursor-2');
+    expect(result.total).toBe(10);
+  });
+
+  it('includes cursor in URL when provided', async () => {
+    const page: CursorPage<number> = { items: [3, 4], nextCursor: null, pageSize: 2 };
+    mockFetchJson(page);
+
+    await fetchCursorPage<number>('/api/items', { cursor: 'cursor-2', limit: 2 });
+
+    const calledUrl = (global.fetch as jest.Mock).mock.calls[0][0] as string;
+    expect(calledUrl).toContain('cursor=cursor-2');
+    expect(calledUrl).toContain('limit=2');
+  });
+
+  it('throws on non-200 responses', async () => {
+    mockFetchJson({ error: 'Not Found' }, 404);
+    await expect(fetchCursorPage('/api/items')).rejects.toThrow('HTTP 404');
+  });
+});
+
+// ─────────────────────────────────────────────────────────────────────────────
+// collectAllPages
+// ─────────────────────────────────────────────────────────────────────────────
+
+describe('collectAllPages', () => {
+  it('collects items from all pages', async () => {
+    const page1: CursorPage<number> = { items: [1, 2], nextCursor: 'c2', pageSize: 2 };
+    const page2: CursorPage<number> = { items: [3, 4], nextCursor: 'c3', pageSize: 2 };
+    const page3: CursorPage<number> = { items: [5], nextCursor: null, pageSize: 2 };
+
+    global.fetch = jest.fn()
+      .mockResolvedValueOnce({ ok: true, status: 200, json: () => Promise.resolve(page1), body: null } as unknown as Response)
+      .mockResolvedValueOnce({ ok: true, status: 200, json: () => Promise.resolve(page2), body: null } as unknown as Response)
+      .mockResolvedValueOnce({ ok: true, status: 200, json: () => Promise.resolve(page3), body: null } as unknown as Response);
+
+    const all = await collectAllPages<number>('/api/items', { limit: 2 });
+    expect(all).toEqual([1, 2, 3, 4, 5]);
+    expect(global.fetch).toHaveBeenCalledTimes(3);
+  });
+});
+
+// ─────────────────────────────────────────────────────────────────────────────
+// streamNdjson
+// ─────────────────────────────────────────────────────────────────────────────
+
+describe('streamNdjson', () => {
+  it('calls onItem for each valid NDJSON line (fallback path)', async () => {
+    const lines = ['{"id":"x1"}', '{"id":"x2"}', '{"id":"x3"}'];
+    mockFetchNdjson(lines);
+
+    const items: { id: string }[] = [];
+    await streamNdjson<{ id: string }>('/stream', (item) => { items.push(item); });
+
+    expect(items).toHaveLength(3);
+    expect(items.map((i) => i.id)).toEqual(['x1', 'x2', 'x3']);
+  });
+
+  it('skips malformed lines silently', async () => {
+    mockFetchNdjson(['{"id":"ok"}', 'not-json', '{"id":"also-ok"}']);
+
+    const items: { id: string }[] = [];
+    await streamNdjson<{ id: string }>('/stream', (item) => { items.push(item); });
+
+    expect(items).toHaveLength(2);
+    expect(items[0].id).toBe('ok');
+    expect(items[1].id).toBe('also-ok');
+  });
+
+  it('throws on non-200 responses', async () => {
+    mockFetchNdjson([], 500);
+    await expect(streamNdjson('/stream', () => {})).rejects.toThrow('HTTP 500');
+  });
+});
+
+// ─────────────────────────────────────────────────────────────────────────────
+// subscribeToSse – basic handler invocation
+// ─────────────────────────────────────────────────────────────────────────────
+
+describe('subscribeToSse', () => {
+  it('returns a cancel function synchronously', () => {
+    global.fetch = jest.fn().mockReturnValue(new Promise(() => {})); // never resolves
+    const cancel = subscribeToSse('/sse', {});
+    expect(typeof cancel).toBe('function');
+    cancel(); // should not throw
+  });
+
+  it('calls onError when fetch returns non-200', async () => {
+    global.fetch = jest.fn().mockResolvedValueOnce({
+      ok: false,
+      status: 403,
+      body: null,
+    } as unknown as Response);
+
+    const onError = jest.fn();
+    const onClose = jest.fn();
+
+    subscribeToSse('/sse', { onError, onClose });
+
+    // Wait for the async internals to run
+    await new Promise((r) => setTimeout(r, 50));
+
+    expect(onError).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining('403') }));
+    expect(onClose).toHaveBeenCalled();
+  });
+});
+
+// ─────────────────────────────────────────────────────────────────────────────
+// getClientMemoryStats
+// ─────────────────────────────────────────────────────────────────────────────
+
+describe('getClientMemoryStats', () => {
+  it('returns a stats object with capturedAt', () => {
+    const stats = getClientMemoryStats();
+    expect(typeof stats.capturedAt).toBe('string');
+    // In Jest/Node, performance.memory is not available so heap fields should be undefined
+    expect(stats.usedJSHeapSize).toBeUndefined();
+  });
+});
diff --git a/app/services/batchTransactionService.ts b/app/services/batchTransactionService.ts
index 4f621ed9..0922c592 100644
--- a/app/services/batchTransactionService.ts
+++ b/app/services/batchTransactionService.ts
@@ -981,6 +981,68 @@ export class BatchTransactionService {
   clearIdempotencyKeys(): void {
     this.idempotencyKeys.clear();
   }
+
+  // ══════════════════════════════════════════════════════════════
+  // Issue #768 – Streaming batch operations
+  // ══════════════════════════════════════════════════════════════
+
+  /**
+   * Execute a batch create by POSTing inputs in chunks and reading
+   * streamed NDJSON per-item results.
+   *
+   * Unlike `executeBatchCreate` (which loads all results into memory),
+   * this variant calls `onResult` as each item result arrives.
+   *
+   * @param inputs  Items to create
+   * @param onResult  Called with each per-item result as it streams in
+   * @param options  Chunk size and abort signal
+   */
+  async executeBatchCreateStream(
+    inputs: BatchCreateInput[],
+    onResult: (result: PerItemResult) => void | Promise<void>,
+    options: { chunkSize?: number; signal?: AbortSignal } = {}
+  ): Promise<{ totalSent: number; streamingComplete: boolean }> {
+    const { chunkSize = this.chunkSize, signal } = options;
+    let totalSent = 0;
+
+    for (let i = 0; i < inputs.length; i += chunkSize) {
+      if (signal?.aborted) break;
+
+      const chunk = inputs.slice(i, i + chunkSize);
+      totalSent += chunk.length;
+
+      // In a real implementation this POSTs to a streaming endpoint and
+      // pipes the NDJSON response through streamNdjson(). Here we simulate
+      // the streaming behaviour locally so the interface is correct.
+      for (let j = 0; j < chunk.length; j++) {
+        const input = chunk[j];
+        const result: PerItemResult = {
+          index: i + j,
+          subscriptionId: `pending_${i + j}`,
+          subscriptionName: input.name,
+          status: 'pending',
+          retryCount: 0,
+        };
+        await onResult(result);
+        // Yield to the event loop between items
+        await new Promise<void>((resolve) => setTimeout(resolve, 0));
+      }
+    }
+
+    return { totalSent, streamingComplete: !signal?.aborted };
+  }
+
+  /**
+   * Estimate the memory pressure (bytes) of the current batch result.
+   *
+   * Uses a conservative estimate: 512 bytes per item result.
+   * Useful for callers that want to decide whether to switch to streaming mode.
+   */
+  memoryPressure(): number {
+    if (!this.currentResult) return 0;
+    const BYTES_PER_ITEM = 512;
+    return this.currentResult.totalItems * BYTES_PER_ITEM;
+  }
 }
 
 export default BatchTransactionService;
diff --git a/app/services/hooks/__tests__/useStreamingList.test.ts b/app/services/hooks/__tests__/useStreamingList.test.ts
new file mode 100644
index 00000000..01304ed9
--- /dev/null
+++ b/app/services/hooks/__tests__/useStreamingList.test.ts
@@ -0,0 +1,156 @@
+/**
+ * Issue #768 – Tests for useStreamingList hook
+ */
+
+import React from 'react';
+import { renderHook, act, waitFor } from '@testing-library/react-native';
+import { useStreamingList } from '../useStreamingList';
+
+// ─────────────────────────────────────────────────────────────────────────────
+// fetch mock helpers
+// ─────────────────────────────────────────────────────────────────────────────
+
+function makePage<T>(items: T[], nextCursor: string | null, total?: number) {
+  return {
+    ok: true,
+    status: 200,
+    json: () => Promise.resolve({ items, nextCursor, total, pageSize: items.length }),
+    body: null,
+  };
+}
+
+beforeEach(() => {
+  jest.resetAllMocks();
+});
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Tests
+// ─────────────────────────────────────────────────────────────────────────────
+
+describe('useStreamingList', () => {
+  it('starts in idle state with no items', () => {
+    global.fetch = jest.fn();
+    const { result } = renderHook(() =>
+      useStreamingList<string>('/api/items')
+    );
+
+    expect(result.current.items).toHaveLength(0);
+    expect(result.current.isLoading).toBe(false);
+    expect(result.current.hasMore).toBe(true);
+    expect(result.current.error).toBeNull();
+    expect(result.current.totalLoaded).toBe(0);
+  });
+
+  it('does NOT auto-fetch when autoLoad is false (default)', () => {
+    global.fetch = jest.fn();
+    renderHook(() => useStreamingList<string>('/api/items'));
+    expect(global.fetch).not.toHaveBeenCalled();
+  });
+
+  it('auto-fetches first page when autoLoad is true', async () => {
+    global.fetch = jest.fn().mockResolvedValue(makePage(['a', 'b'], null));
+
+    const { result } = renderHook(() =>
+      useStreamingList<string>('/api/items', { autoLoad: true, pageSize: 2 })
+    );
+
+    await waitFor(() => expect(result.current.isLoading).toBe(false));
+
+    expect(result.current.items).toEqual(['a', 'b']);
+    expect(result.current.hasMore).toBe(false);
+  });
+
+  it('loadMore appends items from subsequent pages', async () => {
+    global.fetch = jest.fn()
+      .mockResolvedValueOnce(makePage([1, 2], 'cursor-2', 4) as unknown as Response)
+      .mockResolvedValueOnce(makePage([3, 4], null, 4) as unknown as Response);
+
+    const { result } = renderHook(() =>
+      useStreamingList<number>('/api/items', { pageSize: 2 })
+    );
+
+    // Load first page
+    await act(async () => { await result.current.loadMore(); });
+    expect(result.current.items).toEqual([1, 2]);
+    expect(result.current.hasMore).toBe(true);
+    expect(result.current.total).toBe(4);
+
+    // Load second page
+    await act(async () => { await result.current.loadMore(); });
+    expect(result.current.items).toEqual([1, 2, 3, 4]);
+    expect(result.current.hasMore).toBe(false);
+    expect(result.current.totalLoaded).toBe(4);
+  });
+
+  it('does not call fetch again when hasMore is false', async () => {
+    global.fetch = jest.fn().mockResolvedValue(makePage(['x'], null));
+
+    const { result } = renderHook(() => useStreamingList<string>('/api/items'));
+
+    await act(async () => { await result.current.loadMore(); });
+    expect(result.current.hasMore).toBe(false);
+
+    await act(async () => { await result.current.loadMore(); });
+    expect(global.fetch).toHaveBeenCalledTimes(1); // no second call
+  });
+
+  it('sets error state on fetch failure', async () => {
+    global.fetch = jest.fn().mockResolvedValue({
+      ok: false,
+      status: 500,
+      statusText: 'Server Error',
+      text: () => Promise.resolve('Internal server error'),
+      body: null,
+    });
+
+    const { result } = renderHook(() => useStreamingList<string>('/api/items'));
+
+    await act(async () => { await result.current.loadMore(); });
+
+    expect(result.current.error).toBeInstanceOf(Error);
+    expect(result.current.items).toHaveLength(0);
+  });
+
+  it('reset() clears all state and allows re-loading', async () => {
+    global.fetch = jest.fn()
+      .mockResolvedValueOnce(makePage(['a'], 'c2') as unknown as Response)
+      .mockResolvedValueOnce(makePage(['b'], null) as unknown as Response);
+
+    const { result } = renderHook(() => useStreamingList<string>('/api/items'));
+
+    await act(async () => { await result.current.loadMore(); });
+    expect(result.current.items).toHaveLength(1);
+
+    act(() => { result.current.reset(); });
+
+    expect(result.current.items).toHaveLength(0);
+    expect(result.current.hasMore).toBe(true);
+    expect(result.current.error).toBeNull();
+    expect(result.current.totalLoaded).toBe(0);
+
+    await act(async () => { await result.current.loadMore(); });
+    // Reset re-enables loading from the beginning (second mock resolves new data)
+    expect(result.current.items).toHaveLength(1);
+    expect(result.current.items[0]).toBe('b');
+  });
+
+  it('does not double-fetch when loadMore is called while loading', async () => {
+    let resolveFirst!: (v: unknown) => void;
+    global.fetch = jest.fn().mockImplementationOnce(
+      () => new Promise((resolve) => { resolveFirst = resolve; })
+    );
+
+    const { result } = renderHook(() => useStreamingList<string>('/api/items'));
+
+    // Start loading
+    const p1 = act(async () => { await result.current.loadMore(); });
+    // Call loadMore again while in-flight (should be no-op)
+    act(() => { void result.current.loadMore(); });
+
+    // Resolve the fetch
+    resolveFirst(makePage(['z'], null));
+    await p1;
+
+    expect(global.fetch).toHaveBeenCalledTimes(1);
+  });
+});
diff --git a/app/services/hooks/useExportStream.ts b/app/services/hooks/useExportStream.ts
new file mode 100644
index 00000000..74988d9a
--- /dev/null
+++ b/app/services/hooks/useExportStream.ts
@@ -0,0 +1,180 @@
+/**
+ * Issue #768 – useExportStream hook
+ *
+ * Connects to the SSE export progress endpoint and surfaces:
+ *  - Progress percentage + stage message
+ *  - Final download URL once complete
+ *  - Error state
+ *  - Cancel function
+ *
+ * Usage:
+ * ```tsx
+ * const { progress, stage, downloadUrl, error, startExport, cancel } =
+ *   useExportStream('exp_abc123');
+ *
+ * // Start the export
+ * <Button onPress={() => startExport({ format: 'csv' })} title="Export CSV" />
+ *
+ * // Show progress
+ * {progress > 0 && <ProgressBar value={progress} />}
+ *
+ * // Download link
+ * {downloadUrl && <Link href={downloadUrl}>Download</Link>}
+ * ```
+ */
+
+import { useState, useCallback, useRef } from 'react';
+import { subscribeToSse } from '../streamingService';
+
+export type ExportStage = 'idle' | 'connecting' | 'processing' | 'complete' | 'error';
+
+export interface UseExportStreamOptions {
+  /** Base URL for the SSE endpoint. Default: '/exports/stream'. */
+  baseUrl?: string;
+}
+
+export interface StartExportParams {
+  /** Export format. Default: 'json'. */
+  format?: 'json' | 'csv';
+  /** Additional query params forwarded to the SSE endpoint. */
+  params?: Record<string, string>;
+}
+
+export interface UseExportStreamResult {
+  /** Export progress 0–100. 0 when idle. */
+  progress: number;
+  /** Human-readable stage label. */
+  stage: ExportStage;
+  /** Message from the latest progress event. */
+  statusMessage: string;
+  /** Number of records processed so far. */
+  recordsProcessed: number;
+  /** Total records to export (if known). */
+  totalRecords: number | undefined;
+  /** Final download URL (set when stage === 'complete'). */
+  downloadUrl: string | null;
+  /** Final checksum for validation (set when stage === 'complete'). */
+  checksum: string | null;
+  /** Error message (set when stage === 'error'). */
+  error: string | null;
+  /** Start the export stream for the given exportId. */
+  startExport: (exportId: string, params?: StartExportParams) => void;
+  /** Cancel the in-flight SSE connection. */
+  cancel: () => void;
+  /** Reset state back to idle. */
+  reset: () => void;
+}
+
+const initialState = {
+  progress: 0,
+  stage: 'idle' as ExportStage,
+  statusMessage: '',
+  recordsProcessed: 0,
+  totalRecords: undefined as number | undefined,
+  downloadUrl: null as string | null,
+  checksum: null as string | null,
+  error: null as string | null,
+};
+
+export function useExportStream(
+  options: UseExportStreamOptions = {}
+): UseExportStreamResult {
+  const { baseUrl = '/exports/stream' } = options;
+
+  const [state, setState] = useState(initialState);
+  const cancelRef = useRef<(() => void) | null>(null);
+
+  const cancel = useCallback(() => {
+    cancelRef.current?.();
+    cancelRef.current = null;
+    setState((prev) =>
+      prev.stage !== 'idle' && prev.stage !== 'complete' && prev.stage !== 'error'
+        ? { ...prev, stage: 'idle' }
+        : prev
+    );
+  }, []);
+
+  const reset = useCallback(() => {
+    cancelRef.current?.();
+    cancelRef.current = null;
+    setState(initialState);
+  }, []);
+
+  const startExport = useCallback(
+    (exportId: string, params: StartExportParams = {}) => {
+      // Cancel any existing connection
+      cancelRef.current?.();
+
+      const { format = 'json', params: extraParams = {} } = params;
+
+      const url = new URL(`${baseUrl}/${exportId}`, 'http://localhost');
+      url.searchParams.set('format', format);
+      for (const [k, v] of Object.entries(extraParams)) {
+        url.searchParams.set(k, v);
+      }
+
+      setState({
+        ...initialState,
+        stage: 'connecting',
+        statusMessage: 'Connecting…',
+      });
+
+      const stop = subscribeToSse(
+        // Strip the dummy base if running on a real server
+        url.pathname + url.search,
+        {
+          onProgress: (data) => {
+            setState((prev) => ({
+              ...prev,
+              stage: 'processing',
+              progress: data.percent,
+              statusMessage: data.message,
+              recordsProcessed: data.recordsProcessed,
+              totalRecords: data.totalRecords ?? prev.totalRecords,
+            }));
+          },
+          onChunk: (_data) => {
+            // Chunks are handled server-side for the SSE flow;
+            // in a download-and-assemble scenario callers can override this.
+          },
+          onComplete: (data) => {
+            cancelRef.current = null;
+            setState((prev) => ({
+              ...prev,
+              stage: 'complete',
+              progress: 100,
+              statusMessage: 'Export complete',
+              totalRecords: data.totalRecords,
+              downloadUrl: data.downloadUrl ?? null,
+              checksum: data.checksum ?? null,
+            }));
+          },
+          onError: (data) => {
+            cancelRef.current = null;
+            setState((prev) => ({
+              ...prev,
+              stage: 'error',
+              error: data.message,
+              statusMessage: 'Export failed',
+            }));
+          },
+          onClose: () => {
+            cancelRef.current = null;
+          },
+        }
+      );
+
+      cancelRef.current = stop;
+    },
+    [baseUrl]
+  );
+
+  return {
+    ...state,
+    startExport,
+    cancel,
+    reset,
+  };
+}
+
+export default useExportStream;
diff --git a/app/services/hooks/useStreamingList.ts b/app/services/hooks/useStreamingList.ts
new file mode 100644
index 00000000..77016066
--- /dev/null
+++ b/app/services/hooks/useStreamingList.ts
@@ -0,0 +1,132 @@
+/**
+ * Issue #768 – useStreamingList hook
+ *
+ * Incrementally loads cursor-paginated data from a streaming endpoint.
+ * Items are appended as each page arrives — no full dataset is held in memory.
+ *
+ * Usage:
+ * ```tsx
+ * const { items, isLoading, hasMore, loadMore } = useStreamingList<MyRecord>(
+ *   '/subscriptions/stream',
+ *   { pageSize: 50, autoLoad: true }
+ * );
+ * ```
+ */
+
+import { useState, useCallback, useRef, useEffect } from 'react';
+import { fetchCursorPage } from '../streamingService';
+import type { CursorPage } from '../streamingService';
+
+export interface UseStreamingListOptions {
+  /** Records per page. Default: 100. */
+  pageSize?: number;
+  /** If true, fetches the first page automatically on mount. Default: false. */
+  autoLoad?: boolean;
+  /** Additional query params forwarded to the endpoint. */
+  params?: Record<string, string>;
+}
+
+export interface UseStreamingListResult<T> {
+  /** All items loaded so far (appended page-by-page). */
+  items: T[];
+  /** True while a page fetch is in flight. */
+  isLoading: boolean;
+  /** True if there is at least one more page to load. */
+  hasMore: boolean;
+  /** Total records available on the server (if the endpoint returns it). */
+  total: number | undefined;
+  /** Number of items currently loaded. */
+  totalLoaded: number;
+  /** Fetch the next page and append items. No-op if already loading or no more pages. */
+  loadMore: () => Promise<void>;
+  /** Reset to the initial empty state and re-fetch from the first page. */
+  reset: () => void;
+  /** Error from the last failed fetch, if any. */
+  error: Error | null;
+}
+
+export function useStreamingList<T>(
+  url: string,
+  options: UseStreamingListOptions = {}
+): UseStreamingListResult<T> {
+  const { pageSize = 100, autoLoad = false, params } = options;
+
+  const [items, setItems] = useState<T[]>([]);
+  const [isLoading, setIsLoading] = useState(false);
+  const [hasMore, setHasMore] = useState(true);
+  const [total, setTotal] = useState<number | undefined>(undefined);
+  const [error, setError] = useState<Error | null>(null);
+
+  // Store cursor in a ref so loadMore closure always sees the latest value
+  const cursorRef = useRef<string | null | undefined>(undefined);
+  const abortRef = useRef<AbortController | null>(null);
+  const loadingRef = useRef(false);
+
+  const loadMore = useCallback(async () => {
+    if (loadingRef.current || !hasMore) return;
+
+    loadingRef.current = true;
+    setIsLoading(true);
+    setError(null);
+
+    const controller = new AbortController();
+    abortRef.current = controller;
+
+    try {
+      const page: CursorPage<T> = await fetchCursorPage<T>(url, {
+        cursor: cursorRef.current ?? undefined,
+        limit: pageSize,
+        params,
+        signal: controller.signal,
+      });
+
+      cursorRef.current = page.nextCursor;
+      setItems((prev) => [...prev, ...page.items]);
+      setHasMore(page.nextCursor !== null);
+      if (page.total !== undefined) setTotal(page.total);
+    } catch (err) {
+      if ((err as { name?: string })?.name !== 'AbortError') {
+        setError(err instanceof Error ? err : new Error(String(err)));
+      }
+    } finally {
+      loadingRef.current = false;
+      setIsLoading(false);
+      abortRef.current = null;
+    }
+  }, [url, pageSize, params, hasMore]);
+
+  const reset = useCallback(() => {
+    abortRef.current?.abort();
+    cursorRef.current = undefined;
+    setItems([]);
+    loadingRef.current = false;
+    setIsLoading(false);
+    setHasMore(true);
+    setTotal(undefined);
+    setError(null);
+  }, []);
+
+  // Auto-load first page on mount
+  useEffect(() => {
+    if (autoLoad) {
+      void loadMore();
+    }
+    return () => {
+      abortRef.current?.abort();
+    };
+    // eslint-disable-next-line react-hooks/exhaustive-deps
+  }, [autoLoad, url]);
+
+  return {
+    items,
+    isLoading,
+    hasMore,
+    total,
+    totalLoaded: items.length,
+    loadMore,
+    reset,
+    error,
+  };
+}
+
+export default useStreamingList;
diff --git a/app/services/streamingService.ts b/app/services/streamingService.ts
new file mode 100644
index 00000000..a7614a9c
--- /dev/null
+++ b/app/services/streamingService.ts
@@ -0,0 +1,375 @@
+/**
+ * Issue #768 – Client-Side Streaming Service
+ *
+ * Provides typed utilities for:
+ *  - Cursor-based pagination  (fetchCursorPage, collectAllPages)
+ *  - NDJSON streaming         (streamNdjson)
+ *  - Server-Sent Events       (subscribeToSse)
+ *  - Memory stats             (getClientMemoryStats)
+ *
+ * Designed to work in React Native (no DOM EventSource) using fetch +
+ * a manual NDJSON/SSE line parser over ReadableStream.
+ */
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Shared types
+// ─────────────────────────────────────────────────────────────────────────────
+
+/** A page of items from a cursor-paginated endpoint. */
+export interface CursorPage<T> {
+  items: T[];
+  /** Opaque cursor for the next page. `null` = last page. */
+  nextCursor: string | null;
+  total?: number;
+  pageSize: number;
+}
+
+/** Options for fetchCursorPage. */
+export interface FetchPageOptions {
+  /** Cursor returned by a previous page response. */
+  cursor?: string | null;
+  /** Maximum records per page. */
+  limit?: number;
+  /** Additional query params to append. */
+  params?: Record<string, string>;
+  /** AbortSignal for cancellation. */
+  signal?: AbortSignal;
+}
+
+/** SSE event handlers used by subscribeToSse. */
+export interface SseHandlers {
+  onProgress?: (data: {
+    percent: number;
+    message: string;
+    recordsProcessed: number;
+    totalRecords?: number;
+  }) => void;
+  onChunk?: (data: { payload: string; index: number }) => void;
+  onComplete?: (data: {
+    downloadUrl?: string;
+    data?: unknown;
+    totalRecords: number;
+    checksum?: string;
+  }) => void;
+  onError?: (data: { message: string; code?: string }) => void;
+  /** Called when the SSE connection closes (cleanly or on error). */
+  onClose?: () => void;
+}
+
+/** Client-side memory stats (where available). */
+export interface ClientMemoryStats {
+  /** Total JS heap size limit in bytes (Chrome / V8 only). */
+  jsHeapSizeLimit?: number;
+  /** Total JS heap size allocated in bytes. */
+  totalJSHeapSize?: number;
+  /** Current JS heap in use in bytes. */
+  usedJSHeapSize?: number;
+  capturedAt: string;
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Cursor-based pagination
+// ─────────────────────────────────────────────────────────────────────────────
+
+/**
+ * Fetch a single cursor page from a paginated endpoint.
+ *
+ * The server must return a JSON body conforming to `CursorPage<T>`.
+ */
+export async function fetchCursorPage<T>(
+  baseUrl: string,
+  options: FetchPageOptions = {}
+): Promise<CursorPage<T>> {
+  const { cursor, limit = 100, params = {}, signal } = options;
+
+  const isRelative = !/^[a-zA-Z][a-zA-Z0-9+.-]*:/.test(baseUrl);
+  const url = new URL(baseUrl, isRelative ? 'http://localhost' : undefined);
+  if (cursor) url.searchParams.set('cursor', cursor);
+  url.searchParams.set('limit', String(limit));
+  for (const [k, v] of Object.entries(params)) {
+    url.searchParams.set(k, v);
+  }
+
+  const fetchUrl = isRelative ? url.pathname + url.search : url.toString();
+  const res = await fetch(fetchUrl, {
+    method: 'GET',
+    headers: { Accept: 'application/json' },
+    signal,
+  });
+
+  if (!res.ok) {
+    const text = await res.text().catch(() => res.statusText);
+    throw new Error(`fetchCursorPage: HTTP ${res.status} – ${text}`);
+  }
+
+  return (await res.json()) as CursorPage<T>;
+}
+
+/**
+ * Collect ALL pages from a cursor-paginated endpoint into a flat array.
+ *
+ * ⚠️  Only use for datasets you are sure fit in memory.
+ */
+export async function collectAllPages<T>(
+  baseUrl: string,
+  options: Omit<FetchPageOptions, 'cursor'> = {}
+): Promise<T[]> {
+  const all: T[] = [];
+  let cursor: string | null | undefined = undefined;
+
+  do {
+    const page = await fetchCursorPage<T>(baseUrl, { ...options, cursor });
+    all.push(...page.items);
+    cursor = page.nextCursor;
+  } while (cursor !== null && cursor !== undefined);
+
+  return all;
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// NDJSON streaming
+// ─────────────────────────────────────────────────────────────────────────────
+
+/**
+ * Fetch an NDJSON stream from `url` and call `onItem` for each parsed record.
+ *
+ * Processes data incrementally using the Streams API — never buffers the full
+ * body in memory. Works in React Native via the Hermes fetch implementation.
+ *
+ * @param url      Endpoint returning `application/x-ndjson`
+ * @param onItem   Called with each parsed record as it arrives
+ * @param signal   Optional AbortSignal to cancel mid-stream
+ */
+export async function streamNdjson<T>(
+  url: string,
+  onItem: (item: T) => void | Promise<void>,
+  signal?: AbortSignal
+): Promise<void> {
+  const res = await fetch(url, {
+    method: 'GET',
+    headers: { Accept: 'application/x-ndjson, application/json' },
+    signal,
+  });
+
+  if (!res.ok) {
+    const text = await res.text().catch(() => res.statusText);
+    throw new Error(`streamNdjson: HTTP ${res.status} – ${text}`);
+  }
+
+  if (!res.body) {
+    // Fallback: body not streamable (e.g., old RN or test env)
+    const text = await res.text();
+    const lines = text.split('\n').filter((l) => l.trim().length > 0);
+    for (const line of lines) {
+      try {
+        await onItem(JSON.parse(line) as T);
+      } catch {
+        // skip malformed lines
+      }
+    }
+    return;
+  }
+
+  const reader = res.body.getReader();
+  const decoder = new TextDecoder('utf-8');
+  let buffer = '';
+
+  try {
+    while (true) {
+      const { done, value } = await reader.read();
+      if (done) break;
+
+      buffer += decoder.decode(value, { stream: true });
+
+      // Process all complete lines
+      let newlineIdx: number;
+      while ((newlineIdx = buffer.indexOf('\n')) !== -1) {
+        const line = buffer.slice(0, newlineIdx).trim();
+        buffer = buffer.slice(newlineIdx + 1);
+        if (line.length === 0) continue;
+        try {
+          await onItem(JSON.parse(line) as T);
+        } catch {
+          // skip malformed lines
+        }
+      }
+    }
+
+    // Process any remaining data in buffer
+    const remaining = buffer.trim();
+    if (remaining.length > 0) {
+      try {
+        await onItem(JSON.parse(remaining) as T);
+      } catch {
+        // skip malformed trailing data
+      }
+    }
+  } finally {
+    reader.releaseLock();
+  }
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Server-Sent Events (fetch-based, works in React Native)
+// ─────────────────────────────────────────────────────────────────────────────
+
+/** Parsed SSE message. */
+interface SseMessage {
+  event: string;
+  data: string;
+  id?: string;
+}
+
+/** Parse raw SSE text block into a structured message. */
+function parseSseBlock(block: string): SseMessage | null {
+  const lines = block.split('\n');
+  let event = 'message';
+  let data = '';
+  let id: string | undefined;
+
+  for (const line of lines) {
+    if (line.startsWith('event:')) {
+      event = line.slice('event:'.length).trim();
+    } else if (line.startsWith('data:')) {
+      data += line.slice('data:'.length).trim();
+    } else if (line.startsWith('id:')) {
+      id = line.slice('id:'.length).trim();
+    }
+    // ignore 'retry:' and comments for now
+  }
+
+  if (!data) return null;
+  return { event, data, id };
+}
+
+/**
+ * Subscribe to a Server-Sent Events endpoint using `fetch` + ReadableStream.
+ *
+ * Compatible with React Native (no native `EventSource` required).
+ *
+ * Returns a cleanup function that aborts the connection.
+ *
+ * ```ts
+ * const stop = subscribeToSse('/exports/stream/exp_123', {
+ *   onProgress: (p) => setPercent(p.percent),
+ *   onComplete: (d) => setDownloadUrl(d.downloadUrl),
+ * });
+ * // later: stop();
+ * ```
+ */
+export function subscribeToSse(
+  url: string,
+  handlers: SseHandlers,
+  signal?: AbortSignal
+): () => void {
+  const controller = new AbortController();
+  const combinedSignal = signal ?? controller.signal;
+
+  // Start in background — we return the cleanup function synchronously
+  (async () => {
+    try {
+      const res = await fetch(url, {
+        method: 'GET',
+        headers: {
+          Accept: 'text/event-stream',
+          'Cache-Control': 'no-cache',
+        },
+        signal: combinedSignal,
+      });
+
+      if (!res.ok) {
+        handlers.onError?.({ message: `SSE connection failed: HTTP ${res.status}` });
+        handlers.onClose?.();
+        return;
+      }
+
+      if (!res.body) {
+        handlers.onError?.({ message: 'SSE: response body not readable' });
+        handlers.onClose?.();
+        return;
+      }
+
+      const reader = res.body.getReader();
+      const decoder = new TextDecoder('utf-8');
+      let buffer = '';
+
+      try {
+        while (true) {
+          const { done, value } = await reader.read();
+          if (done) break;
+          if (combinedSignal.aborted) break;
+
+          buffer += decoder.decode(value, { stream: true });
+
+          // SSE events are separated by double newlines
+          const blocks = buffer.split('\n\n');
+          buffer = blocks.pop() ?? '';
+
+          for (const block of blocks) {
+            const trimmed = block.trim();
+            if (!trimmed || trimmed.startsWith(':')) continue; // heartbeat/comment
+
+            const msg = parseSseBlock(trimmed);
+            if (!msg) continue;
+
+            try {
+              const payload = JSON.parse(msg.data) as unknown;
+              switch (msg.event) {
+                case 'progress':
+                  handlers.onProgress?.(payload as Parameters<NonNullable<SseHandlers['onProgress']>>[0]);
+                  break;
+                case 'chunk':
+                  handlers.onChunk?.(payload as Parameters<NonNullable<SseHandlers['onChunk']>>[0]);
+                  break;
+                case 'complete':
+                  handlers.onComplete?.(payload as Parameters<NonNullable<SseHandlers['onComplete']>>[0]);
+                  break;
+                case 'error':
+                  handlers.onError?.(payload as Parameters<NonNullable<SseHandlers['onError']>>[0]);
+                  break;
+              }
+            } catch {
+              // ignore parse errors for individual events
+            }
+          }
+        }
+      } finally {
+        reader.releaseLock();
+      }
+    } catch (err) {
+      if ((err as { name?: string })?.name !== 'AbortError') {
+        handlers.onError?.({
+          message: err instanceof Error ? err.message : 'SSE connection error',
+        });
+      }
+    } finally {
+      handlers.onClose?.();
+    }
+  })();
+
+  return () => controller.abort();
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Client memory stats
+// ─────────────────────────────────────────────────────────────────────────────
+
+/**
+ * Return client-side memory statistics where available.
+ * In React Native / Hermes, `performance.memory` is not exposed,
+ * so all heap fields will be `undefined`.
+ */
+export function getClientMemoryStats(): ClientMemoryStats {
+  const mem = (
+    typeof performance !== 'undefined'
+      ? (performance as unknown as { memory?: Record<string, number> }).memory
+      : undefined
+  );
+
+  return {
+    jsHeapSizeLimit: mem?.['jsHeapSizeLimit'],
+    totalJSHeapSize: mem?.['totalJSHeapSize'],
+    usedJSHeapSize: mem?.['usedJSHeapSize'],
+    capturedAt: new Date().toISOString(),
+  };
+}
diff --git a/backend/server.ts b/backend/server.ts
index 522687f0..5e3991aa 100644
--- a/backend/server.ts
+++ b/backend/server.ts
@@ -36,6 +36,14 @@ import { rateLimitingService } from './services/shared/rateLimitingService';
 import { createRateLimitMiddleware, RATE_LIMIT_HEADERS } from './services/shared/rateLimitMiddleware';
 import { applyETagToRawHandler } from './shared/middleware/etagMiddleware';
 import { SubscriptionTier } from '../src/types/subscription';
+// Issue #768 — streaming imports
+import { MemoryMonitor } from './services/shared/streaming';
+import { SseEmitter } from './services/shared/sseEmitter';
+import {
+  streamExportNdjson,
+  streamExportWithProgress,
+} from './services/billing/accountingExportService';
+import type { TransactionRecord } from './services/billing/accountingExportService';
 
 export interface StartServerOptions {
   port?: number;
@@ -145,6 +153,69 @@ function sendJson(
   res.end(JSON.stringify(body));
 }
 
+// ---------------------------------------------------------------------------
+// Issue #768 — Streaming helpers
+// ---------------------------------------------------------------------------
+
+/**
+ * Pipe an async generator of NDJSON strings to a chunked HTTP response.
+ * Uses Transfer-Encoding: chunked automatically (Node default for res.write).
+ */
+async function streamNdjsonResponse(
+  res: http.ServerResponse,
+  gen: AsyncGenerator<string>,
+): Promise<void> {
+  res.writeHead(200, {
+    'Content-Type': 'application/x-ndjson; charset=utf-8',
+    'Transfer-Encoding': 'chunked',
+    'Cache-Control': 'no-cache',
+    'X-Content-Type-Options': 'nosniff',
+  });
+  for await (const line of gen) {
+    res.write(line);
+  }
+  res.end();
+}
+
+/**
+ * Send a streaming file download (CSV or JSON) with appropriate headers.
+ */
+async function streamFileDownload(
+  res: http.ServerResponse,
+  gen: AsyncGenerator<string>,
+  filename: string,
+  mimeType: string,
+): Promise<void> {
+  res.writeHead(200, {
+    'Content-Type': mimeType,
+    'Content-Disposition': `attachment; filename="${filename}"`,
+    'Transfer-Encoding': 'chunked',
+    'Cache-Control': 'no-store',
+  });
+  for await (const chunk of gen) {
+    res.write(chunk);
+  }
+  res.end();
+}
+
+/** Shared memory monitor instance for the server process. */
+const serverMemoryMonitor = new MemoryMonitor({
+  rssThresholdBytes: 500 * 1024 * 1024,
+  heapThresholdBytes: 256 * 1024 * 1024,
+  onThresholdExceeded: (snap, field) => {
+    console.warn(`[MemoryMonitor] Threshold exceeded: ${field} — ${MemoryMonitor.format(snap)}`);
+  },
+});
+
+/**
+ * Stub: fetch transaction records for the requesting merchant.
+ * Replace with a real DB query when Postgres is wired in.
+ */
+function getTransactionRecordsForMerchant(merchantId: string): TransactionRecord[] {
+  void merchantId;
+  return []; // real implementation queries DB
+}
+
 function matchPlanId(pathname: string): string | null {
   const match = pathname.match(/^\/plans\/([^/]+)$/);
   return match?.[1] ?? null;
@@ -337,6 +408,89 @@ export async function startServer(options: StartServerOptions = {}): Promise<Run
         return;
       }
 
+      // -----------------------------------------------------------------
+      // Issue #768 — GET /subscriptions/stream
+      // Cursor-based NDJSON stream of transaction records.
+      // Query params: cursor, limit, merchantId
+      // -----------------------------------------------------------------
+      if (pathname === '/subscriptions/stream' && method === 'GET') {
+        const merchantId = url.searchParams.get('merchantId') ?? 'default';
+        const limit = Math.min(500, Math.max(1, parseInt(url.searchParams.get('limit') ?? '100', 10)));
+        const records = getTransactionRecordsForMerchant(merchantId);
+        const gen = streamExportNdjson(records, {
+          chunkSize: limit,
+          memoryMonitor: serverMemoryMonitor,
+        });
+        await streamNdjsonResponse(res, gen);
+        return;
+      }
+
+      // -----------------------------------------------------------------
+      // Issue #768 — GET /exports/stream/:exportId
+      // SSE endpoint: emits progress events while building an export.
+      // -----------------------------------------------------------------
+      if (pathname.startsWith('/exports/stream/') && method === 'GET') {
+        const exportId = pathname.slice('/exports/stream/'.length);
+        if (!exportId) { sendJson(res, 400, { error: 'exportId is required' }); return; }
+        const format = (url.searchParams.get('format') ?? 'json') as 'csv' | 'json';
+        const merchantId = url.searchParams.get('merchantId') ?? 'default';
+        const records = getTransactionRecordsForMerchant(merchantId);
+        const sse = new SseEmitter(req, res, { heartbeatMs: 15_000, retryMs: 3000 });
+        let chunkIndex = 0;
+        try {
+          const result = await streamExportWithProgress(
+            records,
+            { format, chunkSize: 500, memoryMonitor: serverMemoryMonitor },
+            async (progress) => {
+              if (!sse.isConnected) return;
+              sse.progress({
+                percent: progress.percent,
+                message: `Processing chunk ${chunkIndex + 1}`,
+                recordsProcessed: progress.recordsProcessed,
+                totalRecords: progress.totalRecords,
+              });
+              sse.chunk({ payload: progress.chunk, index: chunkIndex });
+              chunkIndex++;
+            }
+          );
+          sse.complete({
+            downloadUrl: `/exports/download/${exportId}?format=${format}`,
+            totalRecords: result.totalRecords,
+            checksum: result.checksum,
+          });
+        } catch (err) {
+          sse.error({ message: err instanceof Error ? err.message : 'Export failed' });
+        }
+        return;
+      }
+
+      // -----------------------------------------------------------------
+      // Issue #768 — GET /exports/download/:token
+      // Streaming file download (CSV or JSON).
+      // -----------------------------------------------------------------
+      if (pathname.startsWith('/exports/download/') && method === 'GET') {
+        const token = pathname.slice('/exports/download/'.length);
+        if (!token) { sendJson(res, 400, { error: 'token is required' }); return; }
+        const format = (url.searchParams.get('format') ?? 'json') as 'csv' | 'json';
+        const merchantId = url.searchParams.get('merchantId') ?? 'default';
+        const records = getTransactionRecordsForMerchant(merchantId);
+        const mimeType = format === 'csv' ? 'text/csv; charset=utf-8' : 'application/json; charset=utf-8';
+        const filename = `export-${token}.${format}`;
+        const gen = streamExportNdjson(records, { memoryMonitor: serverMemoryMonitor });
+        await streamFileDownload(res, gen, filename, mimeType);
+        return;
+      }
+
+      // -----------------------------------------------------------------
+      // Issue #768 — GET /metrics/memory
+      // Returns current process memory stats.
+      // -----------------------------------------------------------------
+      if (pathname === '/metrics/memory' && method === 'GET') {
+        const snap = serverMemoryMonitor.snapshot();
+        sendJson(res, 200, snap);
+        return;
+      }
+
       sendJson(res, 404, { error: 'Not found' });
     } catch (err) {
       console.error('[Server] Request error:', err);
@@ -361,9 +515,13 @@ export async function startServer(options: StartServerOptions = {}): Promise<Run
     await new Promise<void>((resolve) => {
       server.listen(port, host, () => {
         console.info(`[Server] Listening on http://${host}:${port}`);
-        console.info(`[Server] GraphQL  → POST /graphql`);
-        console.info(`[Server] Plans    → /plans`);
-        console.info(`[Server] Metrics  → GET /metrics/plan-cache`);
+        console.info(`[Server] GraphQL   → POST /graphql`);
+        console.info(`[Server] Plans     → /plans`);
+        console.info(`[Server] Metrics   → GET /metrics/plan-cache`);
+        console.info(`[Server] Memory    → GET /metrics/memory`);
+        console.info(`[Server] Stream    → GET /subscriptions/stream`);
+        console.info(`[Server] SSE       → GET /exports/stream/:exportId`);
+        console.info(`[Server] Download  → GET /exports/download/:token`);
         console.info(`[Server] RateLimit → GET /rate-limits/analytics`);
         console.info(`[Server] RateLimit → GET /rate-limits/status?apiKey=...`);
         console.info(`[Server] RateLimit → POST /rate-limits/bypass`);
diff --git a/backend/services/billing/__tests__/accountingExportService.streaming.test.ts b/backend/services/billing/__tests__/accountingExportService.streaming.test.ts
new file mode 100644
index 00000000..cb19cd0a
--- /dev/null
+++ b/backend/services/billing/__tests__/accountingExportService.streaming.test.ts
@@ -0,0 +1,195 @@
+/**
+ * Issue #768 – Tests for the async streaming variants added to accountingExportService.ts
+ */
+
+import {
+  streamExportAsync,
+  streamExportNdjson,
+  streamExportWithProgress,
+} from '../accountingExportService';
+import type { TransactionRecord, AsyncStreamExportOptions } from '../accountingExportService';
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Fixtures
+// ─────────────────────────────────────────────────────────────────────────────
+
+function makeRecords(count: number): TransactionRecord[] {
+  return Array.from({ length: count }, (_, i) => ({
+    id: `tx_${i}`,
+    merchantId: `merchant_1`,
+    subscriptionId: `sub_${i % 5}`,
+    subscriptionName: `Subscription ${i}`,
+    description: `Desc ${i}`,
+    category: 'streaming',
+    transactionType: 'revenue' as const,
+    amount: 9.99 + i,
+    currency: 'usd',
+    billingCycle: 'monthly',
+    billingDate: Date.now() - i * 86400000,
+    deferredRevenue: 0,
+    createdAt: Date.now() - i * 86400000,
+  }));
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// streamExportAsync
+// ─────────────────────────────────────────────────────────────────────────────
+
+describe('streamExportAsync', () => {
+  async function collect(gen: AsyncGenerator<string>): Promise<string> {
+    let out = '';
+    for await (const chunk of gen) out += chunk;
+    return out;
+  }
+
+  it('json format: produces valid JSON array for 0 records', async () => {
+    const gen = streamExportAsync([], { format: 'json' });
+    const out = await collect(gen);
+    expect(JSON.parse(out)).toEqual([]);
+  });
+
+  it('json format: produces valid JSON array for N records', async () => {
+    const records = makeRecords(15);
+    const gen = streamExportAsync(records, { format: 'json', chunkSize: 5 });
+    const out = await collect(gen);
+    const parsed = JSON.parse(out) as TransactionRecord[];
+    expect(parsed).toHaveLength(15);
+    expect(parsed[0].id).toBe('tx_0');
+    expect(parsed[14].id).toBe('tx_14');
+  });
+
+  it('csv format: first line is the header', async () => {
+    const records = makeRecords(3);
+    const gen = streamExportAsync(records, { format: 'csv', chunkSize: 10 });
+    const out = await collect(gen);
+    const lines = out.split('\n').filter(Boolean);
+    expect(lines[0]).toContain('TransactionId');
+    expect(lines).toHaveLength(4); // header + 3 data rows
+  });
+
+  it('quickbooks format: header row matches QB columns', async () => {
+    const records = makeRecords(2);
+    const gen = streamExportAsync(records, { format: 'quickbooks' });
+    const out = await collect(gen);
+    const firstLine = out.split('\n')[0];
+    expect(firstLine).toContain('Customer');
+    expect(firstLine).toContain('Amount');
+  });
+
+  it('applies filter correctly', async () => {
+    const records = makeRecords(10);
+    const gen = streamExportAsync(records, {
+      format: 'json',
+      filter: { merchantId: 'merchant_1', transactionTypes: ['revenue'] },
+    });
+    const out = await collect(gen);
+    const parsed = JSON.parse(out) as TransactionRecord[];
+    expect(parsed).toHaveLength(10); // all match
+  });
+
+  it('respects chunkSize (generator yields multiple chunks)', async () => {
+    const records = makeRecords(11);
+    const chunks: string[] = [];
+    const gen = streamExportAsync(records, { format: 'csv', chunkSize: 5 });
+    for await (const chunk of gen) chunks.push(chunk);
+    // header + 3 data chunks (5, 5, 1)
+    expect(chunks.length).toBeGreaterThanOrEqual(3);
+  });
+});
+
+// ─────────────────────────────────────────────────────────────────────────────
+// streamExportNdjson
+// ─────────────────────────────────────────────────────────────────────────────
+
+describe('streamExportNdjson', () => {
+  it('yields one NDJSON line per record', async () => {
+    const records = makeRecords(5);
+    const lines: string[] = [];
+    for await (const line of streamExportNdjson(records, {})) {
+      lines.push(line.trim());
+    }
+    expect(lines).toHaveLength(5);
+    for (const line of lines) {
+      expect(() => JSON.parse(line)).not.toThrow();
+    }
+  });
+
+  it('produces empty output for 0 records', async () => {
+    const lines: string[] = [];
+    for await (const line of streamExportNdjson([], {})) {
+      lines.push(line);
+    }
+    expect(lines).toHaveLength(0);
+  });
+
+  it('each line parses to the original record shape', async () => {
+    const records = makeRecords(3);
+    const parsed: TransactionRecord[] = [];
+    for await (const line of streamExportNdjson(records, { chunkSize: 2 })) {
+      parsed.push(JSON.parse(line.trim()) as TransactionRecord);
+    }
+    expect(parsed.map((r) => r.id)).toEqual(['tx_0', 'tx_1', 'tx_2']);
+  });
+});
+
+// ─────────────────────────────────────────────────────────────────────────────
+// streamExportWithProgress
+// ─────────────────────────────────────────────────────────────────────────────
+
+describe('streamExportWithProgress', () => {
+  it('fires onProgress for each chunk and reaches 100%', async () => {
+    const records = makeRecords(20);
+    const progresses: number[] = [];
+
+    await streamExportWithProgress(
+      records,
+      { format: 'json', chunkSize: 5 },
+      async (p) => { progresses.push(p.percent); }
+    );
+
+    expect(progresses[progresses.length - 1]).toBe(100);
+    // Should be monotonically non-decreasing
+    for (let i = 1; i < progresses.length; i++) {
+      expect(progresses[i]).toBeGreaterThanOrEqual(progresses[i - 1]);
+    }
+  });
+
+  it('returns correct totalRecords and a checksum string', async () => {
+    const records = makeRecords(7);
+    const result = await streamExportWithProgress(
+      records,
+      { format: 'csv', chunkSize: 3 },
+      async () => {}
+    );
+    expect(result.totalRecords).toBe(7);
+    expect(typeof result.checksum).toBe('string');
+    expect(result.checksum.length).toBeGreaterThan(0);
+  });
+
+  it('calls onProgress with chunk payload containing data', async () => {
+    const records = makeRecords(5);
+    const chunks: string[] = [];
+
+    await streamExportWithProgress(
+      records,
+      { format: 'csv', chunkSize: 5 },
+      async (p) => { chunks.push(p.chunk); }
+    );
+
+    // There is at least the header + one data chunk
+    expect(chunks.length).toBeGreaterThanOrEqual(1);
+    expect(chunks.some((c) => c.includes('tx_'))).toBe(true);
+  });
+
+  it('returns totalRecords 0 and 0 percent for empty records', async () => {
+    const percents: number[] = [];
+    const result = await streamExportWithProgress(
+      [],
+      { format: 'json', chunkSize: 10 },
+      async (p) => { percents.push(p.percent); }
+    );
+    expect(result.totalRecords).toBe(0);
+    // all progress events should be 100% for empty set
+    expect(percents.every((p) => p === 100)).toBe(true);
+  });
+});
diff --git a/backend/services/billing/accountingExportService.ts b/backend/services/billing/accountingExportService.ts
index 096e84da..0baf6abe 100644
--- a/backend/services/billing/accountingExportService.ts
+++ b/backend/services/billing/accountingExportService.ts
@@ -3,8 +3,12 @@
  *
  * Handles large-dataset streaming exports, reconciliation checks,
  * and encoding-safe output for CSV/JSON/QuickBooks/Xero formats.
+ *
+ * Issue #768: Added async-generator streaming exports and SSE progress variants.
  */
 
+import { MemoryMonitor, toNdjsonLine } from '../shared/streaming';
+
 export type AccountingFormat = 'csv' | 'json' | 'quickbooks' | 'xero';
 export type TransactionType = 'revenue' | 'refund' | 'credit' | 'fee';
 
@@ -41,6 +45,32 @@ export interface StreamExportOptions {
   chunkSize?: number;
 }
 
+// ── Async streaming types (Issue #768) ──────────────────────────────────────
+
+/** Progress callback fired for each chunk during a streaming export. */
+export type ExportProgressCallback = (progress: {
+  /** 0–100 */
+  percent: number;
+  /** Records processed so far */
+  recordsProcessed: number;
+  /** Total records in the filtered result */
+  totalRecords: number;
+  /** Serialised chunk payload */
+  chunk: string;
+  /** Chunk index starting at 0 */
+  chunkIndex: number;
+}) => void | Promise<void>;
+
+/** Options for the async-generator streaming export variants. */
+export interface AsyncStreamExportOptions {
+  format: AccountingFormat;
+  filter?: ExportFilter;
+  /** Chunk size in records. Default: 500. */
+  chunkSize?: number;
+  /** Optional memory monitor — will call check() between chunks. */
+  memoryMonitor?: MemoryMonitor;
+}
+
 export interface ReconciliationResult {
   totalRecords: number;
   totalAmount: number;
@@ -206,6 +236,131 @@ export function streamExport(
   return { totalRecords: filtered.length, checksum: Math.abs(cs).toString(16) };
 }
 
+// ── Async-generator streaming export (Issue #768) ───────────────────────────
+
+/**
+ * Async-generator variant of `streamExport`.
+ *
+ * Yields string chunks lazily — the full result set is never held in memory at
+ * once. Each yielded value is a self-contained fragment that can be piped
+ * directly to an HTTP response or a file stream.
+ *
+ * ```ts
+ * const gen = streamExportAsync(records, { format: 'csv' });
+ * for await (const chunk of gen) {
+ *   res.write(chunk);
+ * }
+ * ```
+ */
+export async function* streamExportAsync(
+  records: TransactionRecord[],
+  options: AsyncStreamExportOptions
+): AsyncGenerator<string> {
+  const { format, filter = {}, chunkSize = 500, memoryMonitor } = options;
+  const filtered = applyFilter(records, filter);
+
+  if (format === 'json') {
+    yield '[';
+    for (let i = 0; i < filtered.length; i += chunkSize) {
+      const batch = filtered.slice(i, i + chunkSize);
+      const separator = i === 0 ? '' : ',';
+      yield separator + batch.map((r) => JSON.stringify(r)).join(',');
+      memoryMonitor?.check();
+      // Yield to event loop between chunks to avoid blocking
+      await new Promise<void>((resolve) => setImmediate(resolve));
+    }
+    yield ']';
+  } else {
+    const headers = headersForFormat(format);
+    yield headers.map(csvEscape).join(',') + '\n';
+    for (let i = 0; i < filtered.length; i += chunkSize) {
+      const batch = filtered.slice(i, i + chunkSize);
+      yield batch.map((r) => recordToCsvRow(r, format)).join('\n') + '\n';
+      memoryMonitor?.check();
+      await new Promise<void>((resolve) => setImmediate(resolve));
+    }
+  }
+}
+
+/**
+ * NDJSON streaming export — each record is its own JSON line.
+ *
+ * Suitable for endpoints that need line-by-line client parsing (e.g., the
+ * `GET /subscriptions/stream` endpoint).
+ */
+export async function* streamExportNdjson(
+  records: TransactionRecord[],
+  options: Omit<AsyncStreamExportOptions, 'format'>
+): AsyncGenerator<string> {
+  const { filter = {}, chunkSize = 500, memoryMonitor } = options;
+  const filtered = applyFilter(records, filter);
+
+  for (let i = 0; i < filtered.length; i += chunkSize) {
+    const batch = filtered.slice(i, i + chunkSize);
+    for (const record of batch) {
+      yield toNdjsonLine(record);
+    }
+    memoryMonitor?.check();
+    await new Promise<void>((resolve) => setImmediate(resolve));
+  }
+}
+
+/**
+ * Streaming export with per-chunk progress callbacks — intended for SSE
+ * endpoints where the caller needs to emit `progress` events to the client
+ * while the export is being built.
+ *
+ * Returns a summary `{ totalRecords, checksum }` once complete (same as
+ * the synchronous `streamExport`).
+ */
+export async function streamExportWithProgress(
+  records: TransactionRecord[],
+  options: AsyncStreamExportOptions,
+  onProgress: ExportProgressCallback
+): Promise<{ totalRecords: number; checksum: string }> {
+  const { format, filter = {}, chunkSize = 500, memoryMonitor } = options;
+  const filtered = applyFilter(records, filter);
+  const totalRecords = filtered.length;
+  let chunkIndex = 0;
+
+  let checksum = 0;
+
+  const emitChunk = async (chunk: string, recordsProcessed: number) => {
+    const percent = totalRecords === 0 ? 100 : Math.round((recordsProcessed / totalRecords) * 100);
+    await onProgress({ percent, recordsProcessed, totalRecords, chunk, chunkIndex });
+    chunkIndex++;
+    memoryMonitor?.check();
+    await new Promise<void>((resolve) => setImmediate(resolve));
+  };
+
+  if (format === 'json') {
+    await emitChunk('[', 0);
+    for (let i = 0; i < filtered.length; i += chunkSize) {
+      const batch = filtered.slice(i, i + chunkSize);
+      const separator = i === 0 ? '' : ',';
+      await emitChunk(separator + batch.map((r) => JSON.stringify(r)).join(','), i + batch.length);
+    }
+    await emitChunk(']', totalRecords);
+  } else {
+    const headers = headersForFormat(format);
+    await emitChunk(headers.map(csvEscape).join(',') + '\n', 0);
+    for (let i = 0; i < filtered.length; i += chunkSize) {
+      const batch = filtered.slice(i, i + chunkSize);
+      await emitChunk(
+        batch.map((r) => recordToCsvRow(r, format)).join('\n') + '\n',
+        i + batch.length
+      );
+    }
+  }
+
+  checksum = filtered.reduce(
+    (acc, r) => acc ^ r.id.split('').reduce((h, c) => h + c.charCodeAt(0), 0),
+    0
+  );
+
+  return { totalRecords, checksum: Math.abs(checksum).toString(16) };
+}
+
 // ── Reconciliation ────────────────────────────────────────────────────────────
 
 /**
diff --git a/backend/services/index.ts b/backend/services/index.ts
index a1a2b04c..4ac0bde0 100644
--- a/backend/services/index.ts
+++ b/backend/services/index.ts
@@ -2,6 +2,32 @@
 export { ConnectionPool, getPool, stellarPool } from './connectionPool';
 export type { PoolConfig, PoolMetrics } from './connectionPool';
 
+// ── Streaming primitives (Issue #768) ─────────────────────────────────────────
+export {
+  createCursorStream,
+  collectStream,
+  encodeOpaqueCursor,
+  decodeOpaqueCursor,
+  MemoryMonitor,
+  toNdjsonLine,
+  parseNdjsonBuffer,
+} from './shared/streaming';
+export type {
+  CursorPage,
+  CursorQueryOptions,
+  PageFetcher,
+  MemorySnapshot,
+  MemoryMonitorConfig,
+} from './shared/streaming';
+export { SseEmitter } from './shared/sseEmitter';
+export type {
+  SseEventName,
+  SseProgressData,
+  SseChunkData,
+  SseCompleteData,
+  SseErrorData,
+} from './shared/sseEmitter';
+
 // ── Repository pattern (#405) ─────────────────────────────────────────────────
 export * from './repositories';
 
@@ -112,6 +138,7 @@ export type {
 } from './subscription/ElasticsearchService';
 export type { ISubscriptionEventStore, IElasticsearchService } from './subscription/interfaces';
 export { SubscriptionError } from './subscription/errors';
+export type { CursorQuery } from './subscription/subscriptionEventStore';
 
 // ── Billing Module ────────────────────────────────────────────────────────────
 export { MeteringService } from './billing/meteringService';
@@ -120,13 +147,15 @@ export { PricingService } from './billing/pricingService';
 export type { PriceRecommendation, ABTestScenario, PricingContext } from './billing/pricingService';
 export { TaxService } from './billing/taxService';
 export { DunningService, dunningService } from './billing/dunningService';
-export { streamExport, reconcile } from './billing/accountingExportService';
+export { streamExport, reconcile, streamExportAsync, streamExportNdjson, streamExportWithProgress } from './billing/accountingExportService';
 export type {
   AccountingFormat,
   TransactionType,
   TransactionRecord,
   ExportFilter,
   StreamExportOptions,
+  AsyncStreamExportOptions,
+  ExportProgressCallback,
   ReconciliationResult,
 } from './billing/accountingExportService';
 export type {
diff --git a/backend/services/shared/__tests__/streaming.test.ts b/backend/services/shared/__tests__/streaming.test.ts
new file mode 100644
index 00000000..1e7989c0
--- /dev/null
+++ b/backend/services/shared/__tests__/streaming.test.ts
@@ -0,0 +1,187 @@
+/**
+ * Issue #768 – Unit tests for backend/services/shared/streaming.ts
+ */
+
+import {
+  encodeOpaqueCursor,
+  decodeOpaqueCursor,
+  createCursorStream,
+  collectStream,
+  MemoryMonitor,
+  toNdjsonLine,
+  parseNdjsonBuffer,
+} from '../streaming';
+import type { CursorPage, PageFetcher } from '../streaming';
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Cursor helpers
+// ─────────────────────────────────────────────────────────────────────────────
+
+describe('encodeOpaqueCursor / decodeOpaqueCursor', () => {
+  it('round-trips a simple payload', () => {
+    const payload = { offset: 42, subscriptionId: 'sub_abc' };
+    const token = encodeOpaqueCursor(payload);
+    expect(typeof token).toBe('string');
+    expect(token.length).toBeGreaterThan(0);
+    expect(decodeOpaqueCursor(token)).toEqual(payload);
+  });
+
+  it('returns null for invalid tokens', () => {
+    expect(decodeOpaqueCursor('not-valid-base64url!!!')).toBeNull();
+    expect(decodeOpaqueCursor('')).toBeNull();
+  });
+
+  it('returns null when decoded JSON is not a plain object', () => {
+    const arrayToken = Buffer.from(JSON.stringify([1, 2, 3])).toString('base64url');
+    expect(decodeOpaqueCursor(arrayToken)).toBeNull();
+  });
+
+  it('produces URL-safe tokens (no +, /, =)', () => {
+    const token = encodeOpaqueCursor({ data: 'hello world!', n: 9999 });
+    expect(token).not.toMatch(/[+/=]/);
+  });
+});
+
+// ─────────────────────────────────────────────────────────────────────────────
+// createCursorStream
+// ─────────────────────────────────────────────────────────────────────────────
+
+function makeFetcher<T>(allItems: T[], pageSize: number): PageFetcher<T> {
+  return async (afterCursor, limit): Promise<CursorPage<T>> => {
+    const offset = afterCursor ? parseInt(afterCursor, 10) : 0;
+    const items = allItems.slice(offset, offset + limit);
+    const nextOffset = offset + items.length;
+    const nextCursor = nextOffset < allItems.length ? String(nextOffset) : null;
+    return { items, nextCursor, total: allItems.length, pageSize: limit };
+  };
+}
+
+describe('createCursorStream', () => {
+  it('yields all pages until exhausted', async () => {
+    const data = Array.from({ length: 25 }, (_, i) => i);
+    const fetcher = makeFetcher(data, 10);
+    const pages: CursorPage<number>[] = [];
+    for await (const page of createCursorStream(fetcher, { limit: 10 })) {
+      pages.push(page);
+    }
+    expect(pages).toHaveLength(3); // 10 + 10 + 5
+    expect(pages.flatMap((p) => p.items)).toEqual(data);
+  });
+
+  it('yields a single page when data fits', async () => {
+    const data = [1, 2, 3];
+    const fetcher = makeFetcher(data, 10);
+    const pages: CursorPage<number>[] = [];
+    for await (const page of createCursorStream(fetcher, { limit: 10 })) {
+      pages.push(page);
+    }
+    expect(pages).toHaveLength(1);
+    expect(pages[0].nextCursor).toBeNull();
+  });
+
+  it('handles empty data source', async () => {
+    const fetcher = makeFetcher([], 10);
+    const pages: CursorPage<number>[] = [];
+    for await (const page of createCursorStream(fetcher, { limit: 10 })) {
+      pages.push(page);
+    }
+    // One empty page is yielded (the fetcher always returns at least one call)
+    expect(pages).toHaveLength(1);
+    expect(pages[0].items).toHaveLength(0);
+    expect(pages[0].nextCursor).toBeNull();
+  });
+});
+
+describe('collectStream', () => {
+  it('collects all items into a flat array', async () => {
+    const data = Array.from({ length: 15 }, (_, i) => `item-${i}`);
+    const fetcher = makeFetcher(data, 5);
+    const result = await collectStream(fetcher, { limit: 5 });
+    expect(result).toEqual(data);
+  });
+});
+
+// ─────────────────────────────────────────────────────────────────────────────
+// MemoryMonitor
+// ─────────────────────────────────────────────────────────────────────────────
+
+describe('MemoryMonitor', () => {
+  it('snapshot() returns a valid snapshot with all required fields', () => {
+    const monitor = new MemoryMonitor();
+    const snap = monitor.snapshot();
+    expect(typeof snap.rss).toBe('number');
+    expect(typeof snap.heapTotal).toBe('number');
+    expect(typeof snap.heapUsed).toBe('number');
+    expect(typeof snap.external).toBe('number');
+    expect(typeof snap.arrayBuffers).toBe('number');
+    expect(typeof snap.capturedAt).toBe('string');
+  });
+
+  it('check() fires onThresholdExceeded when RSS threshold is 0', () => {
+    const onThresholdExceeded = jest.fn();
+    const monitor = new MemoryMonitor({
+      rssThresholdBytes: 0,  // always exceeded
+      onThresholdExceeded,
+    });
+    monitor.check();
+    expect(onThresholdExceeded).toHaveBeenCalled();
+    const [, field] = onThresholdExceeded.mock.calls[0] as [unknown, string];
+    expect(field).toBe('rss');
+  });
+
+  it('check() does not fire callback when thresholds are very high', () => {
+    const onThresholdExceeded = jest.fn();
+    const monitor = new MemoryMonitor({
+      rssThresholdBytes: Number.MAX_SAFE_INTEGER,
+      heapThresholdBytes: Number.MAX_SAFE_INTEGER,
+      onThresholdExceeded,
+    });
+    monitor.check();
+    expect(onThresholdExceeded).not.toHaveBeenCalled();
+  });
+
+  it('MemoryMonitor.format() returns a readable string', () => {
+    const monitor = new MemoryMonitor();
+    const snap = monitor.snapshot();
+    const formatted = MemoryMonitor.format(snap);
+    expect(formatted).toMatch(/RSS=/);
+    expect(formatted).toMatch(/heap=/);
+  });
+});
+
+// ─────────────────────────────────────────────────────────────────────────────
+// NDJSON helpers
+// ─────────────────────────────────────────────────────────────────────────────
+
+describe('toNdjsonLine', () => {
+  it('serialises a value and appends a newline', () => {
+    const line = toNdjsonLine({ id: 'x', value: 42 });
+    expect(line).toBe('{"id":"x","value":42}\n');
+  });
+
+  it('handles primitive values', () => {
+    expect(toNdjsonLine(123)).toBe('123\n');
+    expect(toNdjsonLine(null)).toBe('null\n');
+  });
+});
+
+describe('parseNdjsonBuffer', () => {
+  it('parses multi-line NDJSON', () => {
+    const buffer = '{"a":1}\n{"b":2}\n{"c":3}\n';
+    expect(parseNdjsonBuffer(buffer)).toEqual([{ a: 1 }, { b: 2 }, { c: 3 }]);
+  });
+
+  it('skips blank lines', () => {
+    const buffer = '{"a":1}\n\n{"b":2}\n';
+    expect(parseNdjsonBuffer(buffer)).toHaveLength(2);
+  });
+
+  it('silently skips malformed JSON lines', () => {
+    const buffer = '{"a":1}\nnot-json\n{"b":2}\n';
+    expect(parseNdjsonBuffer(buffer)).toEqual([{ a: 1 }, { b: 2 }]);
+  });
+
+  it('returns empty array for empty string', () => {
+    expect(parseNdjsonBuffer('')).toEqual([]);
+  });
+});
diff --git a/backend/services/shared/sseEmitter.ts b/backend/services/shared/sseEmitter.ts
new file mode 100644
index 00000000..704f6ef8
--- /dev/null
+++ b/backend/services/shared/sseEmitter.ts
@@ -0,0 +1,161 @@
+/**
+ * Issue #768 – Server-Sent Events (SSE) Emitter
+ *
+ * Wraps an `http.ServerResponse` and provides a typed interface for sending
+ * SSE events. Correctly handles:
+ *  - Required headers (Content-Type, Cache-Control, Connection)
+ *  - Client-disconnect cleanup
+ *  - Event ID and retry fields
+ *  - Heartbeat keep-alive pings
+ */
+
+import type http from 'node:http';
+
+// ─────────────────────────────────────────────────────────────────────────────
+// SSE event types (matches what the client-side hook expects)
+// ─────────────────────────────────────────────────────────────────────────────
+
+export type SseEventName = 'progress' | 'chunk' | 'complete' | 'error' | 'ping';
+
+export interface SseProgressData {
+  /** 0–100 */
+  percent: number;
+  /** Human-readable stage label */
+  message: string;
+  /** Records processed so far */
+  recordsProcessed: number;
+  /** Total records (if known) */
+  totalRecords?: number;
+}
+
+export interface SseChunkData {
+  /** Serialised payload fragment */
+  payload: string;
+  /** Sequential chunk index starting at 0 */
+  index: number;
+}
+
+export interface SseCompleteData {
+  /** Final download URL or inline payload for small exports */
+  downloadUrl?: string;
+  /** Inline payload (if small enough to embed) */
+  data?: unknown;
+  /** Total records exported */
+  totalRecords: number;
+  /** Export checksum for client-side validation */
+  checksum?: string;
+}
+
+export interface SseErrorData {
+  message: string;
+  code?: string;
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// SseEmitter
+// ─────────────────────────────────────────────────────────────────────────────
+
+export class SseEmitter {
+  private readonly res: http.ServerResponse;
+  private eventId = 0;
+  private closed = false;
+  private heartbeatTimer: ReturnType<typeof setInterval> | null = null;
+
+  constructor(
+    req: http.IncomingMessage,
+    res: http.ServerResponse,
+    options: { heartbeatMs?: number; retryMs?: number } = {}
+  ) {
+    this.res = res;
+
+    // Set SSE headers before any data is written
+    res.writeHead(200, {
+      'Content-Type': 'text/event-stream; charset=utf-8',
+      'Cache-Control': 'no-cache, no-transform',
+      Connection: 'keep-alive',
+      'X-Accel-Buffering': 'no', // disable nginx proxy buffering
+      'Access-Control-Allow-Origin': '*',
+    });
+
+    // Send retry hint if configured
+    if (options.retryMs) {
+      res.write(`retry: ${options.retryMs}\n\n`);
+    }
+
+    // Clean up on client disconnect
+    req.on('close', () => {
+      this.closed = true;
+      this.stopHeartbeat();
+    });
+
+    // Optional keep-alive heartbeat
+    if (options.heartbeatMs && options.heartbeatMs > 0) {
+      this.heartbeatTimer = setInterval(() => {
+        if (!this.closed) this.ping();
+      }, options.heartbeatMs);
+    }
+  }
+
+  /** Whether the underlying connection has been closed by the client. */
+  get isConnected(): boolean {
+    return !this.closed;
+  }
+
+  /** Send an SSE event with typed data. */
+  send<D>(event: SseEventName, data: D): void {
+    if (this.closed) return;
+    const id = ++this.eventId;
+    const payload = [
+      `id: ${id}`,
+      `event: ${event}`,
+      `data: ${JSON.stringify(data)}`,
+      '',
+      '',
+    ].join('\n');
+    this.res.write(payload);
+  }
+
+  /** Emit a progress update (0–100%). */
+  progress(data: SseProgressData): void {
+    this.send('progress', data);
+  }
+
+  /** Emit a data chunk (for large exports sent in pieces). */
+  chunk(data: SseChunkData): void {
+    this.send('chunk', data);
+  }
+
+  /** Signal successful completion. Closes the stream. */
+  complete(data: SseCompleteData): void {
+    this.send('complete', data);
+    this.close();
+  }
+
+  /** Signal an error. Closes the stream. */
+  error(data: SseErrorData): void {
+    this.send('error', data);
+    this.close();
+  }
+
+  /** Send a heartbeat ping (keeps the connection alive through proxies). */
+  ping(): void {
+    if (this.closed) return;
+    // SSE comment — proxies must forward this
+    this.res.write(': ping\n\n');
+  }
+
+  /** Gracefully close the SSE stream. */
+  close(): void {
+    if (this.closed) return;
+    this.closed = true;
+    this.stopHeartbeat();
+    this.res.end();
+  }
+
+  private stopHeartbeat(): void {
+    if (this.heartbeatTimer) {
+      clearInterval(this.heartbeatTimer);
+      this.heartbeatTimer = null;
+    }
+  }
+}
diff --git a/backend/services/shared/streaming.ts b/backend/services/shared/streaming.ts
new file mode 100644
index 00000000..fc465ead
--- /dev/null
+++ b/backend/services/shared/streaming.ts
@@ -0,0 +1,227 @@
+/**
+ * Issue #768 – API Response Streaming for Large Datasets
+ *
+ * Core streaming primitives:
+ *  - CursorPage<T>            page envelope returned by cursor-paginated endpoints
+ *  - AsyncCursorStream<T>     async-generator helper that lazily yields cursor pages
+ *  - MemoryMonitor            lightweight RSS/heap watcher with configurable thresholds
+ *  - encodeOpaqueCursor /
+ *    decodeOpaqueCursor       base64-JSON cursor encode/decode (store-agnostic)
+ */
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Cursor pagination types
+// ─────────────────────────────────────────────────────────────────────────────
+
+/** A single page of results returned by a cursor-paginated endpoint. */
+export interface CursorPage<T> {
+  /** Items in this page. */
+  items: T[];
+  /**
+   * Opaque cursor pointing to the next page.
+   * `null` means this is the last page.
+   */
+  nextCursor: string | null;
+  /** Total number of records matching the query (may be omitted for performance). */
+  total?: number;
+  /** Page size used for this response. */
+  pageSize: number;
+}
+
+/** Options shared by all cursor-paginated queries. */
+export interface CursorQueryOptions {
+  /** Opaque cursor returned by a previous page request. */
+  afterCursor?: string;
+  /** Maximum records per page. Default: 100. */
+  limit?: number;
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Opaque cursor helpers
+// ─────────────────────────────────────────────────────────────────────────────
+
+/**
+ * Encode an arbitrary payload into an opaque, URL-safe cursor token.
+ * Uses base64url encoding of a JSON-serialised object.
+ */
+export function encodeOpaqueCursor(payload: Record<string, unknown>): string {
+  const json = JSON.stringify(payload);
+  return Buffer.from(json, 'utf8').toString('base64url');
+}
+
+/**
+ * Decode an opaque cursor token back into its original payload.
+ * Returns `null` if the token is invalid or cannot be parsed.
+ */
+export function decodeOpaqueCursor(token: string): Record<string, unknown> | null {
+  try {
+    const json = Buffer.from(token, 'base64url').toString('utf8');
+    const parsed = JSON.parse(json) as unknown;
+    if (parsed && typeof parsed === 'object' && !Array.isArray(parsed)) {
+      return parsed as Record<string, unknown>;
+    }
+    return null;
+  } catch {
+    return null;
+  }
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// AsyncCursorStream
+// ─────────────────────────────────────────────────────────────────────────────
+
+/**
+ * A function that fetches one page of data given a cursor and limit.
+ * The implementation is store-specific (in-memory, Postgres keyset, etc.).
+ */
+export type PageFetcher<T> = (
+  afterCursor: string | null,
+  limit: number
+) => Promise<CursorPage<T>>;
+
+/**
+ * Async generator that lazily yields pages from a cursor-paginated data source.
+ *
+ * Usage:
+ * ```ts
+ * const stream = createCursorStream(fetchPage, { limit: 50 });
+ * for await (const page of stream) {
+ *   // process page.items without holding the whole dataset in memory
+ * }
+ * ```
+ */
+export async function* createCursorStream<T>(
+  fetcher: PageFetcher<T>,
+  options: CursorQueryOptions = {}
+): AsyncGenerator<CursorPage<T>> {
+  const limit = Math.max(1, options.limit ?? 100);
+  let cursor: string | null = options.afterCursor ?? null;
+
+  do {
+    const page = await fetcher(cursor, limit);
+    yield page;
+    cursor = page.nextCursor;
+  } while (cursor !== null);
+}
+
+/**
+ * Collect all pages from a cursor stream into a single flat array.
+ * Only use for small datasets where you are sure the result fits in memory.
+ */
+export async function collectStream<T>(
+  fetcher: PageFetcher<T>,
+  options: CursorQueryOptions = {}
+): Promise<T[]> {
+  const all: T[] = [];
+  for await (const page of createCursorStream(fetcher, options)) {
+    all.push(...page.items);
+  }
+  return all;
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// MemoryMonitor
+// ─────────────────────────────────────────────────────────────────────────────
+
+export interface MemorySnapshot {
+  /** Resident Set Size in bytes. */
+  rss: number;
+  /** Total heap allocated in bytes. */
+  heapTotal: number;
+  /** Heap actually used in bytes. */
+  heapUsed: number;
+  /** V8 external memory in bytes. */
+  external: number;
+  /** Array buffer memory in bytes. */
+  arrayBuffers: number;
+  /** ISO-8601 timestamp. */
+  capturedAt: string;
+}
+
+export interface MemoryMonitorConfig {
+  /** RSS threshold in bytes. Warning emitted when exceeded. Default: 500 MB. */
+  rssThresholdBytes?: number;
+  /** Heap-used threshold in bytes. Warning emitted when exceeded. Default: 256 MB. */
+  heapThresholdBytes?: number;
+  /** Called when a threshold is crossed. */
+  onThresholdExceeded?: (snapshot: MemorySnapshot, field: keyof MemorySnapshot) => void;
+}
+
+/** Lightweight memory watcher. Captures snapshots on demand or periodically. */
+export class MemoryMonitor {
+  private readonly rssThreshold: number;
+  private readonly heapThreshold: number;
+  private readonly onThreshold: MemoryMonitorConfig['onThresholdExceeded'];
+
+  constructor(config: MemoryMonitorConfig = {}) {
+    this.rssThreshold = config.rssThresholdBytes ?? 500 * 1024 * 1024; // 500 MB
+    this.heapThreshold = config.heapThresholdBytes ?? 256 * 1024 * 1024; // 256 MB
+    this.onThreshold = config.onThresholdExceeded;
+  }
+
+  /** Capture current memory snapshot. */
+  snapshot(): MemorySnapshot {
+    const mem = process.memoryUsage();
+    return {
+      rss: mem.rss,
+      heapTotal: mem.heapTotal,
+      heapUsed: mem.heapUsed,
+      external: mem.external,
+      arrayBuffers: mem.arrayBuffers,
+      capturedAt: new Date().toISOString(),
+    };
+  }
+
+  /**
+   * Check memory and fire the threshold callback if exceeded.
+   * Returns the snapshot for callers that want to surface metrics.
+   */
+  check(): MemorySnapshot {
+    const snap = this.snapshot();
+    if (snap.rss > this.rssThreshold) {
+      this.onThreshold?.(snap, 'rss');
+    }
+    if (snap.heapUsed > this.heapThreshold) {
+      this.onThreshold?.(snap, 'heapUsed');
+    }
+    return snap;
+  }
+
+  /** Format snapshot as human-readable string (for logging). */
+  static format(snap: MemorySnapshot): string {
+    const mb = (bytes: number) => `${(bytes / 1024 / 1024).toFixed(1)} MB`;
+    return (
+      `RSS=${mb(snap.rss)} heap=${mb(snap.heapUsed)}/${mb(snap.heapTotal)} ` +
+      `ext=${mb(snap.external)} ts=${snap.capturedAt}`
+    );
+  }
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// NDJSON helpers (used by streaming HTTP endpoints)
+// ─────────────────────────────────────────────────────────────────────────────
+
+/**
+ * Serialise a value to a single NDJSON line (JSON + newline).
+ * Safe for use with chunked-transfer HTTP responses.
+ */
+export function toNdjsonLine(value: unknown): string {
+  return JSON.stringify(value) + '\n';
+}
+
+/**
+ * Parse a buffer of NDJSON text into typed records.
+ * Lines that fail to parse are silently skipped.
+ */
+export function parseNdjsonBuffer<T>(buffer: string): T[] {
+  return buffer
+    .split('\n')
+    .filter((line) => line.trim().length > 0)
+    .flatMap((line) => {
+      try {
+        return [JSON.parse(line) as T];
+      } catch {
+        return [];
+      }
+    });
+}
diff --git a/backend/services/subscription/subscriptionEventStore.ts b/backend/services/subscription/subscriptionEventStore.ts
index 58f53c65..38e79321 100644
--- a/backend/services/subscription/subscriptionEventStore.ts
+++ b/backend/services/subscription/subscriptionEventStore.ts
@@ -1,3 +1,6 @@
+import { encodeOpaqueCursor, decodeOpaqueCursor } from '../shared/streaming';
+import type { CursorPage } from '../shared/streaming';
+
 export type SubscriptionEventType =
   | 'subscription.created'
   | 'subscription.updated'
@@ -34,6 +37,17 @@ export interface SubscriptionEventPage {
   nextCursor?: number;
 }
 
+/**
+ * Cursor-paginated query options used by `queryStream`. (Issue #768)
+ * The cursor is an opaque token returned by a previous `queryStream` call.
+ */
+export interface CursorQuery extends Omit<SubscriptionEventQuery, 'cursor'> {
+  /** Opaque cursor from a previous page. Omit to start from the beginning. */
+  afterCursor?: string;
+  /** Maximum events per page. Default: 50. */
+  limit?: number;
+}
+
 export class SubscriptionEventStore {
   private readonly events: SubscriptionEvent[] = [];
   private readonly sequenceBySubscription = new Map<string, number>();
@@ -103,6 +117,71 @@ export class SubscriptionEventStore {
     }
     return archived;
   }
+
+  // ── Cursor-based streaming query (Issue #768) ──────────────────────────────
+
+  /**
+   * Async generator that yields cursor pages of events lazily.
+   *
+   * Each page contains at most `query.limit` events (default 50).
+   * The caller can iterate to exhaustion or stop early — no memory is wasted
+   * holding the full event set.
+   *
+   * ```ts
+   * for await (const page of store.queryStream({ subscriptionId: 'sub_1' })) {
+   *   for (const event of page.items) { ... }
+   * }
+   * ```
+   */
+  async *queryStream(
+    query: CursorQuery = {}
+  ): AsyncGenerator<CursorPage<SubscriptionEvent>> {
+    const limit = Math.max(1, query.limit ?? 50);
+
+    // Decode the opaque cursor back to a numeric offset
+    let offset = 0;
+    if (query.afterCursor) {
+      const decoded = decodeOpaqueCursor(query.afterCursor);
+      if (decoded && typeof decoded['offset'] === 'number') {
+        offset = decoded['offset'];
+      }
+    }
+
+    // Apply all filters (same logic as `query()`) on the in-memory store.
+    // In a Postgres-backed implementation this would be keyset pagination SQL.
+    const filtered = this.events.filter((event) => {
+      if (!query.includeArchived && event.archivedAt) return false;
+      if (query.subscriptionId && event.subscriptionId !== query.subscriptionId) return false;
+      if (query.type && event.type !== query.type) return false;
+      if (query.from && event.occurredAt < query.from) return false;
+      if (query.to && event.occurredAt > query.to) return false;
+      return true;
+    });
+
+    const total = filtered.length;
+
+    while (offset < total) {
+      const items = filtered.slice(offset, offset + limit);
+      const nextOffset = offset + items.length;
+      const nextCursor =
+        nextOffset < total
+          ? encodeOpaqueCursor({ offset: nextOffset })
+          : null;
+
+      yield {
+        items,
+        nextCursor,
+        total,
+        pageSize: limit,
+      };
+
+      if (nextCursor === null) break;
+      offset = nextOffset;
+
+      // Yield to the event loop between pages
+      await new Promise<void>((resolve) => setImmediate(resolve));
+    }
+  }
 }
 
 export const subscriptionEventStore = new SubscriptionEventStore();
diff --git a/developer-portal/docs/api-reference.md b/developer-portal/docs/api-reference.md
index 1f930301..f77826e5 100644
--- a/developer-portal/docs/api-reference.md
+++ b/developer-portal/docs/api-reference.md
@@ -368,6 +368,40 @@ Reset your sandbox data via the API:
 POST /v1/sandbox/reset
 ```
 
+## Streaming (Issue #768)
+
+For large datasets, SubTrackr provides memory-efficient streaming endpoints. See the full reference in [streaming-api.md](./streaming-api.md).
+
+### Quick Reference
+
+| Endpoint | Protocol | Description |
+|---|---|---|
+| `GET /subscriptions/stream` | NDJSON (chunked) | Stream transaction records line by line |
+| `GET /exports/stream/:exportId` | Server-Sent Events | Real-time export progress + chunks |
+| `GET /exports/download/:token` | Chunked download | Download exported file without buffering |
+| `GET /metrics/memory` | JSON | Current server memory usage |
+
+### Basic Usage
+
+**Stream records incrementally (JavaScript)**
+
+```js
+const res = await fetch('/subscriptions/stream?limit=100');
+const reader = res.body.getReader();
+// read line-by-line — see streaming-api.md for full example
+```
+
+**Track export progress (React Native)**
+
+```tsx
+import { useExportStream } from '../services/hooks/useExportStream';
+
+const { progress, downloadUrl, startExport } = useExportStream();
+// startExport('exp_123', { format: 'csv' });
+```
+
+---
+
 ## Support
 
 - **Documentation**: [docs.subtrackr.io](https://docs.subtrackr.io)
diff --git a/developer-portal/docs/streaming-api.md b/developer-portal/docs/streaming-api.md
new file mode 100644
index 00000000..5f0f9027
--- /dev/null
+++ b/developer-portal/docs/streaming-api.md
@@ -0,0 +1,228 @@
+# Streaming API Reference
+
+This document describes the streaming endpoints added in **Issue #768** to address memory pressure when working with large datasets.
+
+---
+
+## Overview
+
+SubTrackr exposes three streaming patterns:
+
+| Pattern | Protocol | Endpoint |
+|---|---|---|
+| **NDJSON list stream** | HTTP chunked transfer | `GET /subscriptions/stream` |
+| **Export progress** | Server-Sent Events (SSE) | `GET /exports/stream/:exportId` |
+| **File download** | HTTP chunked transfer | `GET /exports/download/:token` |
+| **Memory metrics** | JSON | `GET /metrics/memory` |
+
+---
+
+## NDJSON Streaming — `GET /subscriptions/stream`
+
+Returns transaction records as Newline-Delimited JSON (NDJSON). Each line is a self-contained JSON object. The connection uses **chunked transfer encoding** — records are streamed as they are processed without loading the full result into memory.
+
+### Query Parameters
+
+| Parameter | Type | Default | Description |
+|---|---|---|---|
+| `merchantId` | string | `"default"` | Filter records by merchant |
+| `limit` | integer | `100` | Records per internal chunk (1–500) |
+| `cursor` | string | — | Opaque cursor from a previous response for pagination |
+
+### Response
+
+`Content-Type: application/x-ndjson`
+
+One JSON object per line:
+
+```json
+{"id":"tx_001","merchantId":"m1","subscriptionId":"sub_1","amount":9.99,...}
+{"id":"tx_002","merchantId":"m1","subscriptionId":"sub_2","amount":14.99,...}
+```
+
+### Client Example (JavaScript)
+
+```js
+const res = await fetch('/subscriptions/stream?limit=100');
+const reader = res.body.getReader();
+const decoder = new TextDecoder();
+let buffer = '';
+
+while (true) {
+  const { done, value } = await reader.read();
+  if (done) break;
+  buffer += decoder.decode(value, { stream: true });
+  const lines = buffer.split('\n');
+  buffer = lines.pop(); // save incomplete line
+  for (const line of lines) {
+    if (!line.trim()) continue;
+    const record = JSON.parse(line);
+    console.log(record);
+  }
+}
+```
+
+### Client Example (React Native — using `streamNdjson`)
+
+```tsx
+import { streamNdjson } from '../services/streamingService';
+
+await streamNdjson('/subscriptions/stream', (record) => {
+  // called once per record, never buffers the full response
+  dispatch(addRecord(record));
+});
+```
+
+---
+
+## Export Progress — `GET /exports/stream/:exportId`
+
+Opens a **Server-Sent Events** connection. The server builds the export in the background and emits typed events as each chunk is processed.
+
+### Query Parameters
+
+| Parameter | Type | Default | Description |
+|---|---|---|---|
+| `format` | `"json"` \| `"csv"` | `"json"` | Export format |
+| `merchantId` | string | `"default"` | Merchant scope |
+
+### SSE Events
+
+All event data is JSON-encoded.
+
+#### `progress`
+
+```
+event: progress
+data: {"percent":42,"message":"Processing chunk 3","recordsProcessed":2100,"totalRecords":5000}
+```
+
+#### `chunk`
+
+```
+event: chunk
+data: {"payload":"[{\"id\":\"tx_1\",...}]","index":2}
+```
+
+#### `complete`
+
+```
+event: complete
+data: {"downloadUrl":"/exports/download/tok_abc?format=csv","totalRecords":5000,"checksum":"a1b2c3"}
+```
+
+#### `error`
+
+```
+event: error
+data: {"message":"Export failed: connection timeout","code":"TIMEOUT"}
+```
+
+### Heartbeat
+
+The server sends `: ping` comments every **15 seconds** to keep the connection alive through proxies.
+
+### Client Example (React Native — using `useExportStream`)
+
+```tsx
+import { useExportStream } from '../services/hooks/useExportStream';
+
+function ExportButton({ exportId }: { exportId: string }) {
+  const { progress, stage, downloadUrl, startExport } = useExportStream();
+
+  return (
+    <>
+      <Button title="Export CSV" onPress={() => startExport(exportId, { format: 'csv' })} />
+      {stage === 'processing' && <ProgressBar value={progress} />}
+      {downloadUrl && <Text>Ready: {downloadUrl}</Text>}
+    </>
+  );
+}
+```
+
+---
+
+## Streaming File Download — `GET /exports/download/:token`
+
+Downloads the exported file using **chunked transfer encoding**. No `Content-Length` is set — the client reads until the stream closes.
+
+### Query Parameters
+
+| Parameter | Type | Default | Description |
+|---|---|---|---|
+| `format` | `"json"` \| `"csv"` | `"json"` | File format |
+| `merchantId` | string | `"default"` | Merchant scope |
+
+### Response Headers
+
+```
+Content-Type: text/csv; charset=utf-8
+Content-Disposition: attachment; filename="export-<token>.csv"
+Transfer-Encoding: chunked
+Cache-Control: no-store
+```
+
+### cURL Example
+
+```bash
+curl --no-buffer \
+  "http://localhost:3001/exports/download/tok_abc?format=csv" \
+  -o export.csv
+```
+
+---
+
+## Memory Metrics — `GET /metrics/memory`
+
+Returns current Node.js process memory statistics. Useful for monitoring memory usage during or after large streaming operations.
+
+### Response
+
+```json
+{
+  "rss": 52428800,
+  "heapTotal": 31457280,
+  "heapUsed": 21233664,
+  "external": 1048576,
+  "arrayBuffers": 65536,
+  "capturedAt": "2026-07-28T00:00:00.000Z"
+}
+```
+
+All values are in **bytes**.
+
+---
+
+## Cursor-Based Pagination Protocol
+
+Cursor-paginated endpoints return a `nextCursor` field. Pass it as the `cursor` query parameter to retrieve the next page.
+
+```
+GET /subscriptions/stream?limit=100
+→ Returns records 1–100, nextCursor: "eyJvZmZzZXQiOjEwMH0"
+
+GET /subscriptions/stream?limit=100&cursor=eyJvZmZzZXQiOjEwMH0
+→ Returns records 101–200, nextCursor: "eyJvZmZzZXQiOjIwMH0"
+
+GET /subscriptions/stream?limit=100&cursor=eyJvZmZzZXQiOjIwMH0
+→ Returns records 201–250, nextCursor: null  ← last page
+```
+
+Cursors are **opaque base64url-encoded tokens**. Do not attempt to parse or construct them manually.
+
+---
+
+## Rate Limiting
+
+Streaming connections count as **one request** against the rate-limit budget. The connection is not throttled per-chunk. See [rate-limiting.md](../rate-limiting.md) for tier limits.
+
+---
+
+## Error Handling
+
+| Scenario | HTTP Status | SSE Event |
+|---|---|---|
+| Invalid exportId | `400` | — |
+| Export failed mid-stream | — | `error` event |
+| Client disconnects | Connection closed | — |
+| Memory threshold exceeded | Server logs warning | — |
diff --git a/docs/streaming-architecture.md b/docs/streaming-architecture.md
new file mode 100644
index 00000000..ddbd71e0
--- /dev/null
+++ b/docs/streaming-architecture.md
@@ -0,0 +1,147 @@
+# Streaming Architecture (Issue #768)
+
+Internal engineering reference for the streaming infrastructure added in issue #768.
+
+---
+
+## Problem Statement
+
+Before this change, all list/export endpoints loaded entire result sets into memory before sending a response. For large merchant datasets (10k+ transaction records) this caused:
+
+- RSS spikes exceeding 500 MB
+- GC pressure and latency during export requests
+- Mobile clients holding large JSON arrays in JavaScript heap
+
+## Solution Overview
+
+```
+┌─────────────────────────┐      NDJSON (chunked)      ┌───────────────────────┐
+│  GET /subscriptions/    │ ─────────────────────────► │  streamNdjson()       │
+│  stream?limit=100       │                             │  (line-by-line parse) │
+└─────────────────────────┘                             └───────────────────────┘
+
+┌─────────────────────────┐        SSE stream          ┌───────────────────────┐
+│  GET /exports/stream/   │ ─────────────────────────► │  useExportStream()    │
+│  :exportId              │  progress/chunk/complete   │  progress % + URL     │
+└─────────────────────────┘                             └───────────────────────┘
+
+┌─────────────────────────┐    Chunked file download   ┌───────────────────────┐
+│  GET /exports/download/ │ ─────────────────────────► │  fetch() → save file  │
+│  :token                 │                             │                       │
+└─────────────────────────┘                             └───────────────────────┘
+```
+
+---
+
+## Key Components
+
+### `backend/services/shared/streaming.ts`
+
+Core primitives:
+
+| Export | Purpose |
+|---|---|
+| `createCursorStream<T>` | Async generator wrapping any `PageFetcher<T>` |
+| `collectStream<T>` | Convenience: collect all pages (small datasets only) |
+| `encodeOpaqueCursor` | base64url-encode a JSON payload into a cursor token |
+| `decodeOpaqueCursor` | Decode a cursor token back to its payload |
+| `MemoryMonitor` | Check RSS/heap between chunks; fire callback on threshold |
+| `toNdjsonLine` | Serialise one value to `JSON\n` |
+| `parseNdjsonBuffer` | Split a string buffer into typed records |
+
+### `backend/services/shared/sseEmitter.ts`
+
+`SseEmitter` wraps `http.ServerResponse` and provides:
+- `send(event, data)` — typed SSE event emission
+- `progress(data)` / `chunk(data)` / `complete(data)` / `error(data)` — typed helpers
+- `ping()` — heartbeat comment to keep proxies alive
+- `close()` — graceful stream termination
+- Client disconnect handled via `req.on('close', …)`
+
+### Cursor Encoding Scheme
+
+Cursors are `base64url(JSON.stringify(payload))` where `payload` is store-specific:
+
+| Store | Payload |
+|---|---|
+| In-memory array | `{ offset: number }` |
+| PostgreSQL (future) | `{ id: string, createdAt: number }` (keyset) |
+
+The `encodeOpaqueCursor` / `decodeOpaqueCursor` helpers are store-agnostic — swap the payload without changing the HTTP contract.
+
+---
+
+## Memory Budget
+
+| Metric | Target | Monitored by |
+|---|---|---|
+| Server RSS | ≤ 500 MB | `MemoryMonitor` (warn) |
+| Server heap used | ≤ 256 MB | `MemoryMonitor` (warn) |
+| Chunk size | 500 records (default) | `chunkSize` option |
+| Client JS heap | ≤ platform limit | `getClientMemoryStats()` |
+
+Between each chunk, `setImmediate` is called to yield to the event loop and allow the GC to collect short-lived chunk buffers.
+
+---
+
+## Performance Benchmarks (target)
+
+Benchmarks should be measured with the existing `npm run performance:ci` pipeline.
+
+| Scenario | Target p95 |
+|---|---|
+| NDJSON stream (10k records, 500/chunk) | < 1 200 ms first byte |
+| SSE export progress (5k records) | < 250 ms first `progress` event |
+| Chunked download (10k records, CSV) | < 2 000 ms total |
+| Memory delta after 10k stream | < 50 MB RSS |
+
+---
+
+## SSE Reconnection Behaviour
+
+- The server sends `retry: 3000` at connection open — browsers/clients will retry after 3 s.
+- The server sends `: ping` heartbeats every 15 s to prevent proxy timeouts.
+- Each SSE event includes an `id:` field for safe reconnect (client can resume from last seen event in future).
+- React Native's `subscribeToSse` in `streamingService.ts` does **not** auto-reconnect — callers are expected to call `startExport()` again if needed.
+
+---
+
+## Testing Strategy
+
+### Unit tests
+
+| File | What is tested |
+|---|---|
+| `streaming.test.ts` | Cursor encode/decode, async generator page count, MemoryMonitor thresholds, NDJSON helpers |
+| `accountingExportService.streaming.test.ts` | `streamExportAsync` JSON/CSV correctness, `streamExportNdjson` per-line, `streamExportWithProgress` monotonic progress |
+| `streamingService.test.ts` | `fetchCursorPage` URL construction, `collectAllPages` multi-page, `streamNdjson` line parse + malformed skip, `subscribeToSse` error path |
+| `useStreamingList.test.ts` | Idle state, autoLoad, pagination, hasMore guard, error, reset, double-fetch prevention |
+
+### Integration / manual
+
+```bash
+# Start server
+npm run server:start
+
+# Confirm NDJSON stream
+curl --no-buffer "http://localhost:3001/subscriptions/stream?limit=10"
+
+# Watch memory metrics during stream
+curl "http://localhost:3001/metrics/memory"
+
+# Confirm SSE events
+curl -N "http://localhost:3001/exports/stream/exp_test?format=csv"
+
+# Download export
+curl --no-buffer "http://localhost:3001/exports/download/tok_test?format=csv" -o /tmp/test.csv
+```
+
+---
+
+## Future Work
+
+- [ ] Replace stub `getTransactionRecordsForMerchant()` in `server.ts` with real Postgres keyset query
+- [ ] Keyset cursor payload for Postgres: `{ id: lastId, createdAt: lastCreatedAt }`
+- [ ] Persistent export jobs with job ID (queue-backed, resumable SSE)
+- [ ] Backpressure: pause generator when the response write buffer is full
+- [ ] Rate-limit streaming connections separately from regular requests