feat: streaming client methods + types [AMB-3016]

packages/payments/src/client.ts

@@ -69,13 +69,13 @@ export class Payments extends AmbossClient {
 
   get wallets(): Wallets {
     this.requireServiceApiKey('payments.wallets');
-    this.#wallets ??= new Wallets(this.graphqlClient);
+    this.#wallets ??= new Wallets(this.graphqlClient, this.config);
     return this.#wallets;
   }
 
   get transactions(): Transactions {
     this.requireServiceApiKey('payments.transactions');
-    this.#transactions ??= new Transactions(this.graphqlClient);
+    this.#transactions ??= new Transactions(this.graphqlClient, this.config);
     return this.#transactions;
   }
 

packages/payments/src/index.ts

@@ -18,6 +18,7 @@ export type {
   SendProgress,
   SendResult,
 } from './resources/transactions.types.js';
+export type { WatchStreamOptions } from './resources/streaming.types.js';
 export type { NodePaymentResult, PaymentLifecycleStatus } from './node/types.js';
 
 // These methods' param/result types (e.g. `CreateReceiveTransactionInput`,

packages/payments/src/resources/sseParser.test.ts

@@ -0,0 +1,66 @@
+import assert from 'node:assert/strict';
+import { describe, it } from 'node:test';
+
+import { parseSseStream } from './sseParser.js';
+
+/** Turns SSE wire text into the chunked byte stream `parseSseStream` reads. */
+function streamOf(chunks: string[]): ReadableStream<Uint8Array> {
+  const encoder = new TextEncoder();
+  let i = 0;
+  return new ReadableStream<Uint8Array>({
+    pull(controller) {
+      if (i >= chunks.length) {
+        controller.close();
+        return;
+      }
+      controller.enqueue(encoder.encode(chunks[i]));
+      i += 1;
+    },
+  });
+}
+
+async function collect(stream: ReadableStream<Uint8Array>) {
+  const events = [];
+  for await (const event of parseSseStream(stream)) events.push(event);
+  return events;
+}
+
+describe('parseSseStream', () => {
+  it('parses id/event/data fields from a single dispatched event', async () => {
+    const events = await collect(
+      streamOf(['id: evt-1\nevent: payment.completed\ndata: {"a":1}\n\n']),
+    );
+    assert.deepEqual(events, [{ id: 'evt-1', event: 'payment.completed', data: '{"a":1}' }]);
+  });
+
+  it('joins multiple data: lines with newlines, per the SSE spec', async () => {
+    const events = await collect(streamOf(['data: line1\ndata: line2\n\n']));
+    assert.deepEqual(events, [{ id: undefined, event: undefined, data: 'line1\nline2' }]);
+  });
+
+  it('ignores comment lines used for heartbeats', async () => {
+    const events = await collect(streamOf([': keepalive\n\ndata: real\n\n']));
+    assert.deepEqual(events, [{ id: undefined, event: undefined, data: 'real' }]);
+  });
+
+  it('reassembles an event split across multiple chunks', async () => {
+    const events = await collect(streamOf(['event: pay', 'ment.pending\ndata: {"x":2}', '\n\n']));
+    assert.deepEqual(events, [{ id: undefined, event: 'payment.pending', data: '{"x":2}' }]);
+  });
+
+  it('dispatches multiple events from one stream in order', async () => {
+    const events = await collect(streamOf(['data: first\n\ndata: second\n\n']));
+    assert.deepEqual(
+      events.map((e) => e.data),
+      ['first', 'second'],
+    );
+  });
+
+  it('drops an eventless trailing buffer with no blank-line terminator', async () => {
+    const events = await collect(streamOf(['data: complete\n\ndata: incomplete']));
+    assert.deepEqual(
+      events.map((e) => e.data),
+      ['complete'],
+    );
+  });
+});

packages/payments/src/resources/sseParser.ts

@@ -0,0 +1,69 @@
+/** One dispatched Server-Sent Event, per the WHATWG SSE spec's field set (minus `retry`, which this SDK doesn't act on — see the note in `streaming.ts` about `Last-Event-ID`). */
+export interface ParsedSseEvent {
+  id?: string;
+  event?: string;
+  data: string;
+}
+
+/**
+ * Parses a `text/event-stream` body into dispatched events. Reads the stream
+ * manually via `getReader()`/`TextDecoder` (rather than `EventSource`, which
+ * doesn't exist in Node and can't set custom headers) so it works in both
+ * Node and browsers.
+ *
+ * Comment lines (`:...`, used by the server for heartbeats) are dropped
+ * silently. `retry:` and any other unrecognized field is ignored — this SDK
+ * never reconnects with `Last-Event-ID`, so there is nothing for `retry` to
+ * configure.
+ */
+export async function* parseSseStream(
+  body: ReadableStream<Uint8Array>,
+): AsyncGenerator<ParsedSseEvent, void, void> {
+  const reader = body.getReader();
+  const decoder = new TextDecoder('utf-8');
+  let buffer = '';
+  let id: string | undefined;
+  let event: string | undefined;
+  let dataLines: string[] = [];
+
+  const resetEvent = (): void => {
+    id = undefined;
+    event = undefined;
+    dataLines = [];
+  };
+
+  try {
+    for (;;) {
+      const { value, done } = await reader.read();
+      if (done) break;
+
+      buffer += decoder.decode(value, { stream: true });
+      const lines = buffer.split('\n');
+      buffer = lines.pop() ?? '';
+
+      for (const rawLine of lines) {
+        const line = rawLine.endsWith('\r') ? rawLine.slice(0, -1) : rawLine;
+
+        if (line === '') {
+          if (dataLines.length > 0) {
+            yield { id, event, data: dataLines.join('\n') };
+          }
+          resetEvent();
+          continue;
+        }
+        if (line.startsWith(':')) continue;
+
+        const colonIndex = line.indexOf(':');
+        const field = colonIndex === -1 ? line : line.slice(0, colonIndex);
+        let value2 = colonIndex === -1 ? '' : line.slice(colonIndex + 1);
+        if (value2.startsWith(' ')) value2 = value2.slice(1);
+
+        if (field === 'id') id = value2;
+        else if (field === 'event') event = value2;
+        else if (field === 'data') dataLines.push(value2);
+      }
+    }
+  } finally {
+    reader.releaseLock();
+  }
+}

packages/payments/src/resources/streamToken.ts

@@ -0,0 +1,51 @@
+import type { GraphQLClient } from 'graphql-request';
+
+import { AmbossClient } from '@ambosstech/core';
+
+import type {
+  MintStreamTokenMutation,
+  MintStreamTokenMutationVariables,
+  StreamTokenScope,
+} from './streamToken.types.js';
+
+/**
+ * Hand-authored — `payment.mutation.stream_token.mint` lands in
+ * amboss-rails-api#565 (unmerged, not yet deployed), so this SDK's schema
+ * snapshot (`packages/core/schema/rails.graphql`) doesn't have it yet and
+ * `pnpm --filter @ambosstech/payments run codegen` cannot generate a typed
+ * document for it. Written by hand against PR #565's schema in the meantime.
+ *
+ * TODO(AMB-3016): once #565 merges and deploys, run
+ * `pnpm --filter @ambosstech/core run refresh-schema && pnpm --filter @ambosstech/payments run codegen`
+ * and delete this file (and `streamToken.types.ts`) in favor of the generated
+ * `MintStreamToken` operation in `../generated/sdk.js`.
+ */
+const MintStreamTokenDocument = `
+  mutation MintStreamToken($input: MintStreamTokenInput!) {
+    payment {
+      stream_token {
+        mint(input: $input) {
+          token
+          expires_at
+        }
+      }
+    }
+  }
+`;
+
+/** Mints a short-lived JWT scoped to one transaction/wallet/environment, for use as the SSE stream's bearer token. */
+export async function mintStreamToken(
+  graphqlClient: GraphQLClient,
+  scope: StreamTokenScope,
+  id: string,
+): Promise<MintStreamTokenMutation['payment']['stream_token']['mint']> {
+  try {
+    const res = await graphqlClient.request<
+      MintStreamTokenMutation,
+      MintStreamTokenMutationVariables
+    >(MintStreamTokenDocument, { input: { scope, id } });
+    return res.payment.stream_token.mint;
+  } catch (err) {
+    throw AmbossClient.translateError(err);
+  }
+}

packages/payments/src/resources/streamToken.types.ts

@@ -0,0 +1,35 @@
+/**
+ * Hand-written GraphQL types for `payment.mutation.stream_token.mint`.
+ *
+ * TODO(AMB-3016): amboss-rails-api#565 (unmerged) adds this mutation to the
+ * live schema. Once it merges and deploys to production, run
+ * `pnpm --filter @ambosstech/core run refresh-schema && pnpm --filter @ambosstech/payments run codegen`
+ * to generate the real `MintStreamToken*` types/document in
+ * `../generated/sdk.js`, then delete this file and `streamToken.ts` in favor
+ * of the generated versions. Names here match PR #565's schema exactly so the
+ * swap is a rename, not a rewrite.
+ *
+ * (`WatchStreamOptions` lives in `streaming.types.ts`, not here — it's an SDK
+ * option, not part of the GraphQL schema, so it survives the codegen swap.)
+ */
+export type StreamTokenScope = 'TRANSACTION' | 'WALLET' | 'ENVIRONMENT';
+
+export interface MintStreamTokenInput {
+  scope: StreamTokenScope;
+  id: string;
+}
+
+export interface MintStreamTokenMutationVariables {
+  input: MintStreamTokenInput;
+}
+
+export interface MintStreamTokenMutation {
+  payment: {
+    stream_token: {
+      mint: {
+        token: string;
+        expires_at: string;
+      };
+    };
+  };
+}