From 63b80ef90b10a4f5d05b7fb67ec6e5ff78a8538c Mon Sep 17 00:00:00 2001 From: Bufo Date: Fri, 21 Aug 2026 11:57:54 +0200 Subject: [PATCH] feat: add streaming client methods for transactions and wallets [AMB-3016] --- packages/payments/src/client.ts | 4 +- packages/payments/src/index.ts | 1 + .../payments/src/resources/sseParser.test.ts | 66 ++++++++ packages/payments/src/resources/sseParser.ts | 69 ++++++++ .../payments/src/resources/streamToken.ts | 51 ++++++ .../src/resources/streamToken.types.ts | 35 ++++ .../payments/src/resources/streaming.test.ts | 152 ++++++++++++++++++ packages/payments/src/resources/streaming.ts | 94 +++++++++++ .../payments/src/resources/streaming.types.ts | 5 + .../src/resources/transactions.send.test.ts | 35 ++-- .../payments/src/resources/transactions.ts | 41 ++++- packages/payments/src/resources/wallets.ts | 39 ++++- 12 files changed, 577 insertions(+), 15 deletions(-) create mode 100644 packages/payments/src/resources/sseParser.test.ts create mode 100644 packages/payments/src/resources/sseParser.ts create mode 100644 packages/payments/src/resources/streamToken.ts create mode 100644 packages/payments/src/resources/streamToken.types.ts create mode 100644 packages/payments/src/resources/streaming.test.ts create mode 100644 packages/payments/src/resources/streaming.ts create mode 100644 packages/payments/src/resources/streaming.types.ts diff --git a/packages/payments/src/client.ts b/packages/payments/src/client.ts index df88a99..e83412d 100644 --- a/packages/payments/src/client.ts +++ b/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; } diff --git a/packages/payments/src/index.ts b/packages/payments/src/index.ts index f2e317b..33a2b1d 100644 --- a/packages/payments/src/index.ts +++ b/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`, diff --git a/packages/payments/src/resources/sseParser.test.ts b/packages/payments/src/resources/sseParser.test.ts new file mode 100644 index 0000000..0fc682c --- /dev/null +++ b/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 { + const encoder = new TextEncoder(); + let i = 0; + return new ReadableStream({ + pull(controller) { + if (i >= chunks.length) { + controller.close(); + return; + } + controller.enqueue(encoder.encode(chunks[i])); + i += 1; + }, + }); +} + +async function collect(stream: ReadableStream) { + 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'], + ); + }); +}); diff --git a/packages/payments/src/resources/sseParser.ts b/packages/payments/src/resources/sseParser.ts new file mode 100644 index 0000000..f38d327 --- /dev/null +++ b/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, +): AsyncGenerator { + 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(); + } +} diff --git a/packages/payments/src/resources/streamToken.ts b/packages/payments/src/resources/streamToken.ts new file mode 100644 index 0000000..177ad16 --- /dev/null +++ b/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 { + try { + const res = await graphqlClient.request< + MintStreamTokenMutation, + MintStreamTokenMutationVariables + >(MintStreamTokenDocument, { input: { scope, id } }); + return res.payment.stream_token.mint; + } catch (err) { + throw AmbossClient.translateError(err); + } +} diff --git a/packages/payments/src/resources/streamToken.types.ts b/packages/payments/src/resources/streamToken.types.ts new file mode 100644 index 0000000..09fe946 --- /dev/null +++ b/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; + }; + }; + }; +} diff --git a/packages/payments/src/resources/streaming.test.ts b/packages/payments/src/resources/streaming.test.ts new file mode 100644 index 0000000..38d5905 --- /dev/null +++ b/packages/payments/src/resources/streaming.test.ts @@ -0,0 +1,152 @@ +import assert from 'node:assert/strict'; +import { describe, it } from 'node:test'; + +import { ApiError, NetworkError } from '@ambosstech/core'; + +import { Payments } from '../client.js'; +import type { PaymentEvent } from '../types/webhooks.js'; + +const MINT_RESPONSE = { + payment: { + stream_token: { mint: { token: 'stream-tok-abc', expires_at: '2099-01-01T00:00:00Z' } }, + }, +}; + +/** + * Fake `fetch` shared by the mint (GraphQL POST) and stream (plain GET) legs + * of `watch`/`watchEvents`. Routes on the URL so both calls can be answered + * from one client, and records the stream request's headers for assertions. + */ +function buildFetch(opts: { + streamBody: string | ReadableStream; + streamStatus?: number; + streamHeaders?: Record; + onStreamRequest?: (headers: Headers) => void; +}): typeof fetch { + return (async (input: string | URL | Request, init?: RequestInit) => { + const url = typeof input === 'string' || input instanceof URL ? String(input) : input.url; + if (url.includes('/payments/stream/')) { + if (init?.signal?.aborted) { + throw new DOMException('The operation was aborted.', 'AbortError'); + } + opts.onStreamRequest?.(new Headers(init?.headers)); + return new Response(opts.streamBody, { + status: opts.streamStatus ?? 200, + headers: { 'content-type': 'text/event-stream', ...opts.streamHeaders }, + }); + } + return new Response(JSON.stringify({ data: MINT_RESPONSE }), { + status: 200, + headers: { 'content-type': 'application/json' }, + }); + }) as typeof fetch; +} + +async function collect(iterable: AsyncIterable): Promise { + const items: T[] = []; + for await (const item of iterable) items.push(item); + return items; +} + +const SAMPLE_EVENT: PaymentEvent = { + id: 'payment.completed:tx1', + event_type: 'payment.completed', + environment: 'production', + environment_id: 'env_1', + wallet_id: 'wal_1', + node_id: null, + data: { + id: 'tx1', + fee: null, + amount: { amount: '1000', asset_id: 'btc', precision: 0, asset_symbol: 'BTC' }, + status: 'completed', + metadata: null, + direction: 'receive', + expires_at: null, + settled_at: '2026-01-01T00:00:00Z', + description: null, + exchange_rate: null, + settle_amount: { amount: '1000', asset_id: 'btc', precision: 0, asset_symbol: 'BTC' }, + payment_details: { payment_hash: 'hash1', payment_type: 'bolt11', payment_request: 'lnbc1' }, + }, +}; + +describe('Transactions.watch', () => { + it('mints a stream token and yields each PaymentEvent from the SSE body', async () => { + let sentAuth: string | null = null; + const sse = `id: ${SAMPLE_EVENT.id}\nevent: ${SAMPLE_EVENT.event_type}\ndata: ${JSON.stringify(SAMPLE_EVENT)}\n\n`; + const payments = new Payments({ + serviceApiKey: 'amb_live_test', + fetch: buildFetch({ + streamBody: sse, + onStreamRequest: (headers) => { + sentAuth = headers.get('authorization'); + }, + }), + }); + + const events = await collect(payments.transactions.watch('tx1')); + + assert.deepEqual(events, [SAMPLE_EVENT]); + assert.equal(sentAuth, 'Bearer stream-tok-abc'); + }); + + it('stops without yielding a PaymentEvent for a stream_closed control event', async () => { + const sse = `data: ${JSON.stringify(SAMPLE_EVENT)}\n\n` + `event: stream_closed\ndata: {}\n\n`; + const payments = new Payments({ + serviceApiKey: 'amb_live_test', + fetch: buildFetch({ streamBody: sse }), + }); + + const events = await collect(payments.transactions.watch('tx1')); + + assert.deepEqual(events, [SAMPLE_EVENT]); + }); + + it('throws ApiError when the stream endpoint rejects before opening (401/404)', async () => { + const payments = new Payments({ + serviceApiKey: 'amb_live_test', + fetch: buildFetch({ + streamBody: JSON.stringify({ error: 'unauthorized' }), + streamStatus: 401, + streamHeaders: { 'content-type': 'application/json' }, + }), + }); + + await assert.rejects( + () => collect(payments.transactions.watch('tx1')), + (err: unknown) => + err instanceof ApiError && err.status === 401 && err.message === 'unauthorized', + ); + }); +}); + +describe('Transactions.watch — abort', () => { + it('rejects with NetworkError when options.signal aborts before the stream opens', async () => { + const controller = new AbortController(); + controller.abort(); + const payments = new Payments({ + serviceApiKey: 'amb_live_test', + fetch: buildFetch({ streamBody: '' }), + }); + + await assert.rejects( + () => collect(payments.transactions.watch('tx1', { signal: controller.signal })), + (err: unknown) => err instanceof NetworkError, + ); + }); +}); + +describe('Wallets.watchEvents', () => { + it('mints a stream token and yields each PaymentEvent from the SSE body', async () => { + const sse = `data: ${JSON.stringify(SAMPLE_EVENT)}\n\n`; + const payments = new Payments({ + serviceApiKey: 'amb_live_test', + fetch: buildFetch({ streamBody: sse }), + }); + + const events = await collect(payments.wallets.watchEvents('wal_1')); + + assert.deepEqual(events, [SAMPLE_EVENT]); + }); +}); diff --git a/packages/payments/src/resources/streaming.ts b/packages/payments/src/resources/streaming.ts new file mode 100644 index 0000000..071e2aa --- /dev/null +++ b/packages/payments/src/resources/streaming.ts @@ -0,0 +1,94 @@ +import type { GraphQLClient } from 'graphql-request'; + +import { AmbossClient, ApiError, NetworkError, type ResolvedClientConfig } from '@ambosstech/core'; + +import type { PaymentEvent } from '../types/webhooks.js'; +import { parseSseStream } from './sseParser.js'; +import { mintStreamToken } from './streamToken.js'; +import type { StreamTokenScope } from './streamToken.types.js'; +import type { WatchStreamOptions } from './streaming.types.js'; + +/** + * Sent by the server (per the SSE design doc) as the final event before it + * closes a stream for a reason other than transaction-terminal state (e.g. + * the 30-minute max stream lifetime). It carries no `PaymentEvent` payload, + * so it's a signal to stop iterating rather than something to yield. + */ +const STREAM_CLOSED_EVENT = 'stream_closed'; + +/** + * Mints a scoped stream token, opens the SSE connection, and yields each + * `PaymentEvent` the server sends. Shared by `Transactions.watch` and + * `Wallets.watchEvents` — same mechanism, different scope/endpoint. + * + * Ends (the generator returns) when the server closes the connection — + * either after a transaction reaches a terminal status or after the max + * stream lifetime. An aborted `options.signal`, a pre-stream HTTP rejection + * (401/404), or a transport failure all reject instead, as `ApiError` / + * `NetworkError` — the same error types every other resource method throws. + */ +export async function* watchPaymentEventStream( + graphqlClient: GraphQLClient, + config: Pick, + scope: StreamTokenScope, + streamPath: string, + id: string, + options?: WatchStreamOptions, +): AsyncGenerator { + const { token } = await mintStreamToken(graphqlClient, scope, id); + + // The stream endpoints are plain HTTP routes on the same origin as the + // GraphQL endpoint (e.g. `.../graphql` -> `.../payments/stream/...`), not + // part of the GraphQL schema. + const url = new URL(streamPath, new URL(config.baseUrl).origin); + + let response: Response; + try { + response = await config.fetch(url, { + // A fetch-based client can set headers (unlike browser `EventSource`), + // so prefer the header over `?token=` — the design doc calls this out + // as the lower-exposure option (query strings land in access/proxy + // logs; the header doesn't). + headers: { authorization: `Bearer ${token}`, accept: 'text/event-stream' }, + signal: options?.signal, + }); + } catch (err) { + throw AmbossClient.translateError(err); + } + + if (!response.ok) { + // Rejected before the stream opens (401 missing/invalid token, 404 + // wrong scope/id/environment) — per the design doc, both come back as a + // plain JSON body, not an SSE event. + const body: unknown = await response.json().catch(() => undefined); + const message = + (body as { error?: string } | undefined)?.error ?? + `Stream request failed with HTTP ${response.status}`; + throw new ApiError({ message, status: response.status, response: body }); + } + if (!response.body) { + throw new NetworkError('Stream response has no body', undefined); + } + + try { + for await (const parsedEvent of parseSseStream(response.body)) { + if (parsedEvent.event === STREAM_CLOSED_EVENT) return; + if (!parsedEvent.data) continue; + + let paymentEvent: PaymentEvent; + try { + paymentEvent = JSON.parse(parsedEvent.data) as PaymentEvent; + } catch (err) { + throw new NetworkError('Received malformed SSE payment event payload', err); + } + yield paymentEvent; + } + } catch (err) { + // Covers a mid-stream transport failure and an aborted `options.signal` + // (the fetch body read rejects with `AbortError`) — translated the same + // way as the pre-stream fetch above, so every failure mode of `watch`/ + // `watchEvents` surfaces as `ApiError`/`NetworkError`. + if (err instanceof ApiError || err instanceof NetworkError) throw err; + throw AmbossClient.translateError(err); + } +} diff --git a/packages/payments/src/resources/streaming.types.ts b/packages/payments/src/resources/streaming.types.ts new file mode 100644 index 0000000..75fa718 --- /dev/null +++ b/packages/payments/src/resources/streaming.types.ts @@ -0,0 +1,5 @@ +/** Options shared by `Transactions.watch` and `Wallets.watchEvents`. */ +export interface WatchStreamOptions { + /** Aborts the stream connection. */ + signal?: AbortSignal; +} diff --git a/packages/payments/src/resources/transactions.send.test.ts b/packages/payments/src/resources/transactions.send.test.ts index d7e27de..c53d382 100644 --- a/packages/payments/src/resources/transactions.send.test.ts +++ b/packages/payments/src/resources/transactions.send.test.ts @@ -6,6 +6,8 @@ import { argon2id } from '@noble/hashes/argon2'; import { bytesToHex } from '@noble/hashes/utils'; import type { GraphQLClient } from 'graphql-request'; +import type { ResolvedClientConfig } from '@ambosstech/core'; + import { nip44Encrypt } from '../crypto/nip44.js'; import { Transactions } from './transactions.js'; @@ -14,6 +16,15 @@ const TEAM_ID = '11111111-1111-1111-1111-111111111111'; const MACAROON_HEX = '0201036c6e6402240a'; const SYMMETRIC_KEY = bytesToHex(new Uint8Array(64).map((_, i) => (i * 5 + 1) & 0xff)); +/** `Transactions` only needs this for `watch()`, which none of these send() tests exercise. */ +const FAKE_CONFIG: ResolvedClientConfig = { + apiKey: undefined, + serviceApiKey: undefined, + baseUrl: 'https://rails.amboss.tech/graphql', + fetch, + timeoutMs: 30_000, +}; + let server: Server | undefined; let lastBody: unknown; afterEach(async () => { @@ -138,7 +149,7 @@ async function prepareThenFailWrongPassword(): Promise<{ { result: { status: 'SUCCEEDED', payment_hash: 'ph', fee_sat: '1' } }, ]); const { client, ops } = withCallLog(fakeClient(host)); - const transactions = new Transactions(client); + const transactions = new Transactions(client, FAKE_CONFIG); await transactions.prepareSend({ walletId: 'w1', password: PASSWORD }); ops.length = 0; @@ -161,7 +172,7 @@ describe('Transactions.send', () => { { result: { status: 'IN_FLIGHT' } }, { result: { status: 'SUCCEEDED', payment_hash: 'ph', fee_sat: '1' } }, ]); - const transactions = new Transactions(fakeClient(host)); + const transactions = new Transactions(fakeClient(host), FAKE_CONFIG); const statuses: string[] = []; const result = await transactions.send({ @@ -182,7 +193,7 @@ describe('Transactions.send', () => { it('creates a sandbox send without a password and returns payment: null', async () => { // No node should be contacted for sandbox — point at an unroutable host // so any accidental node call would fail the test. - const transactions = new Transactions(fakeClient('http://127.0.0.1:1', 'SANDBOX')); + const transactions = new Transactions(fakeClient('http://127.0.0.1:1', 'SANDBOX'), FAKE_CONFIG); const result = await transactions.send({ walletId: 'w1', @@ -197,7 +208,7 @@ describe('Transactions.send', () => { const host = await startNode([ { result: { status: 'SUCCEEDED', payment_hash: 'ph', fee_sat: '1' } }, ]); - const transactions = new Transactions(fakeClient(host)); + const transactions = new Transactions(fakeClient(host), FAKE_CONFIG); const result = await transactions.send({ walletId: 'w1', @@ -212,7 +223,7 @@ describe('Transactions.send', () => { it('surfaces a wrong password as a DecryptionError before paying', async () => { const host = await startNode([{ result: { status: 'SUCCEEDED' } }]); - const transactions = new Transactions(fakeClient(host)); + const transactions = new Transactions(fakeClient(host), FAKE_CONFIG); await assert.rejects( transactions.send({ @@ -234,6 +245,7 @@ describe('Transactions.send', () => { fee: '3', payment_request: 'lnbc1xyz', }), + FAKE_CONFIG, ); const result = await transactions.send({ @@ -256,6 +268,7 @@ describe('Transactions.send', () => { ]); const transactions = new Transactions( fakeClient(host, 'LIVE', { id: 'tx1', status: 'PENDING', payment_request: 'lnbc1xyz' }), + FAKE_CONFIG, ); const result = await transactions.send({ @@ -276,7 +289,7 @@ describe('Transactions.prepareSend', () => { { result: { status: 'SUCCEEDED', payment_hash: 'ph', fee_sat: '1' } }, ]); const { client, ops } = withCallLog(fakeClient(host)); - const transactions = new Transactions(client); + const transactions = new Transactions(client, FAKE_CONFIG); await transactions.prepareSend({ walletId: 'w1', password: PASSWORD }); assert.equal(countOf(ops, 'GetWalletSendContext'), 1); @@ -298,7 +311,7 @@ describe('Transactions.prepareSend', () => { }); it('reports isSendReady false while preparing and true once resolved', async () => { - const transactions = new Transactions(fakeClient('http://127.0.0.1:1')); + const transactions = new Transactions(fakeClient('http://127.0.0.1:1'), FAKE_CONFIG); const pending = transactions.prepareSend({ walletId: 'w1', password: PASSWORD }); assert.equal(transactions.isSendReady('w1'), false, 'not ready while Argon2 is still running'); @@ -321,7 +334,7 @@ describe('Transactions.prepareSend', () => { }); it('marks a sandbox wallet ready without a password', async () => { - const transactions = new Transactions(fakeClient('http://127.0.0.1:1', 'SANDBOX')); + const transactions = new Transactions(fakeClient('http://127.0.0.1:1', 'SANDBOX'), FAKE_CONFIG); await transactions.prepareSend({ walletId: 'w1' }); @@ -353,7 +366,7 @@ describe('Transactions.prepareSend', () => { { result: { status: 'SUCCEEDED', payment_hash: 'ph', fee_sat: '1' } }, ]); const { client, ops } = withCallLog(fakeClient(host)); - const transactions = new Transactions(client); + const transactions = new Transactions(client, FAKE_CONFIG); await transactions.prepareSend({ walletId: 'w1', password: PASSWORD }); ops.length = 0; @@ -373,7 +386,7 @@ describe('Transactions.prepareSend', () => { it('keeps a concurrent successful preparation when another send has bad credentials', async () => { const host = await startNode([{ result: { status: 'SUCCEEDED' } }]); - const transactions = new Transactions(fakeClient(host)); + const transactions = new Transactions(fakeClient(host), FAKE_CONFIG); // Both derivations are in flight at once: the good one is started first, // then the bad one, which must not displace it. @@ -400,7 +413,7 @@ describe('Transactions.prepareSend', () => { { result: { status: 'SUCCEEDED', payment_hash: 'ph', fee_sat: '1' } }, ]); const { client, ops } = withCallLog(fakeClient(host, 'LIVE', undefined, WALLET_TEAM_ID)); - const transactions = new Transactions(client); + const transactions = new Transactions(client, FAKE_CONFIG); // Only the override's salt decrypts this wallet, so preparing succeeding // at all proves the override was used. diff --git a/packages/payments/src/resources/transactions.ts b/packages/payments/src/resources/transactions.ts index 5ca7414..b7ffeed 100644 --- a/packages/payments/src/resources/transactions.ts +++ b/packages/payments/src/resources/transactions.ts @@ -1,5 +1,7 @@ import type { GraphQLClient } from 'graphql-request'; +import type { ResolvedClientConfig } from '@ambosstech/core'; + import { createMasterPasswordHash } from '../crypto/argon2.js'; import { decryptAdminMacaroonWithMasterKey } from '../crypto/decryptAdminMacaroon.js'; import { PaymentSendError } from '../errors.js'; @@ -12,8 +14,11 @@ import { import { sendAssetPayment } from '../node/lit.js'; import { sendLndPayment } from '../node/lnd.js'; import type { PaymentLifecycleStatus } from '../node/types.js'; +import type { PaymentEvent } from '../types/webhooks.js'; import { translateSdkErrors } from './sdkErrors.js'; import { selectSendNode } from './sendNode.js'; +import { watchPaymentEventStream } from './streaming.js'; +import type { WatchStreamOptions } from './streaming.types.js'; import type { PreparedSend, PrepareSendParams, @@ -68,6 +73,8 @@ function lndAmountSats(destination: SendDestination): string | undefined { export class Transactions { readonly #sdk: ReturnType; + readonly #graphqlClient: GraphQLClient; + readonly #config: ResolvedClientConfig; /** * Macaroons prepared by {@link prepareSend}, keyed by wallet id. Only a * *successful* preparation lands here, and only `prepareSend` ever writes: @@ -79,8 +86,10 @@ export class Transactions { /** Preparations still running, so concurrent `prepareSend` calls share one Argon2 pass. */ readonly #pending = new Map>(); - constructor(graphqlClient: GraphQLClient) { + constructor(graphqlClient: GraphQLClient, config: ResolvedClientConfig) { this.#sdk = getSdk(graphqlClient, translateSdkErrors); + this.#graphqlClient = graphqlClient; + this.#config = config; } /** @@ -251,6 +260,36 @@ export class Transactions { return { transaction, payment }; } + /** + * Streams live `PaymentEvent`s for a single transaction over SSE — the + * live-status alternative to configuring a webhook endpoint. Mints a + * short-lived stream token via GraphQL, then reads + * `GET /payments/stream/transactions/:id` as `text/event-stream` (a plain + * `fetch`, not `EventSource`, so this works in Node and browsers alike). + * + * Per the SSE design doc: the first event is a snapshot of the transaction's + * current state; the stream then emits one event per status transition and + * closes once the transaction reaches a terminal status + * (`COMPLETED`/`FAILED`/`EXPIRED`) or the token's max stream lifetime (30 + * minutes) elapses — the returned iterable simply ends. Call `watch` again + * (which mints a fresh token) to reconnect. + * + * Rejects with `ApiError` (401 missing/invalid token, 404 unknown + * transaction) before the stream opens, or `NetworkError` for a transport + * failure or an aborted `options.signal` — the same error types every other + * resource method throws. + */ + watch(id: string, options?: WatchStreamOptions): AsyncIterable { + return watchPaymentEventStream( + this.#graphqlClient, + this.#config, + 'TRANSACTION', + `/payments/stream/transactions/${id}`, + id, + options, + ); + } + /** * The context `send()` will pay with. Passing a `password` means "use these * credentials", so it always derives; omitting one means "use what was diff --git a/packages/payments/src/resources/wallets.ts b/packages/payments/src/resources/wallets.ts index 50949be..3f0bdee 100644 --- a/packages/payments/src/resources/wallets.ts +++ b/packages/payments/src/resources/wallets.ts @@ -1,18 +1,27 @@ import type { GraphQLClient } from 'graphql-request'; +import type { ResolvedClientConfig } from '@ambosstech/core'; + import { getSdk, type CreatePaymentsWalletInput, type PaymentsWalletFieldsFragment, type SimplePaymentsWalletFieldsFragment, } from '../generated/sdk.js'; +import type { PaymentEvent } from '../types/webhooks.js'; import { translateSdkErrors } from './sdkErrors.js'; +import { watchPaymentEventStream } from './streaming.js'; +import type { WatchStreamOptions } from './streaming.types.js'; export class Wallets { readonly #sdk: ReturnType; + readonly #graphqlClient: GraphQLClient; + readonly #config: ResolvedClientConfig; - constructor(graphqlClient: GraphQLClient) { + constructor(graphqlClient: GraphQLClient, config: ResolvedClientConfig) { this.#sdk = getSdk(graphqlClient, translateSdkErrors); + this.#graphqlClient = graphqlClient; + this.#config = config; } async list(params: { environmentId: string }): Promise { @@ -34,4 +43,32 @@ export class Wallets { const res = await this.#sdk.DeleteWallet({ id }); return res.payment.wallet.delete; } + + /** + * Streams live `PaymentEvent`s for every transaction on a wallet over SSE — + * the wallet/environment-scoped counterpart to `Transactions.watch`. Mints a + * short-lived stream token via GraphQL, then reads + * `GET /payments/stream/wallets/:id` as `text/event-stream` (plain `fetch`, + * not `EventSource`, so it works in Node and browsers alike). + * + * Unlike a transaction stream, a wallet has no terminal state: per the SSE + * design doc this stream stays open until the token's max stream lifetime + * (30 minutes) elapses — there is no other natural end. Call `watchEvents` + * again (which mints a fresh token) to reconnect. + * + * Rejects with `ApiError` (401 missing/invalid token, 404 unknown wallet) + * before the stream opens, or `NetworkError` for a transport failure or an + * aborted `options.signal` — the same error types every other resource + * method throws. + */ + watchEvents(id: string, options?: WatchStreamOptions): AsyncIterable { + return watchPaymentEventStream( + this.#graphqlClient, + this.#config, + 'WALLET', + `/payments/stream/wallets/${id}`, + id, + options, + ); + } }