From 56e59c6ef7a8fe084955e57264bb4223f8ba28be Mon Sep 17 00:00:00 2001 From: Parv Ahuja <17094219+parvahuja@users.noreply.github.com> Date: Wed, 15 Jul 2026 13:23:33 -0700 Subject: [PATCH] refactor(session): extract SSE response driver --- .changeset/session-sse-driver.md | 5 ++ src/tempo/PublicExports.test-d.ts | 1 + src/tempo/session/client/Transports.test.ts | 34 +++++++++++ src/tempo/session/client/Transports.ts | 67 +++++++++++++++------ 4 files changed, 87 insertions(+), 20 deletions(-) create mode 100644 .changeset/session-sse-driver.md diff --git a/.changeset/session-sse-driver.md b/.changeset/session-sse-driver.md new file mode 100644 index 00000000..fbc0cff5 --- /dev/null +++ b/.changeset/session-sse-driver.md @@ -0,0 +1,5 @@ +--- +'mppx': patch +--- + +Extracted session SSE payment handling into a response-preserving driver. diff --git a/src/tempo/PublicExports.test-d.ts b/src/tempo/PublicExports.test-d.ts index 7042b3ad..204515b1 100644 --- a/src/tempo/PublicExports.test-d.ts +++ b/src/tempo/PublicExports.test-d.ts @@ -83,6 +83,7 @@ test('tempo session public barrel hides internal session drivers', () => { expectTypeOf().not.toHaveProperty('settle') expectTypeOf().not.toHaveProperty('ChannelOps') expectTypeOf().not.toHaveProperty('Chain') + expectTypeOf().not.toHaveProperty('driveSseResponse') expectTypeOf().not.toHaveProperty('ChannelOps') expectTypeOf().not.toHaveProperty('Chain') }) diff --git a/src/tempo/session/client/Transports.test.ts b/src/tempo/session/client/Transports.test.ts index 99834c35..75622645 100644 --- a/src/tempo/session/client/Transports.test.ts +++ b/src/tempo/session/client/Transports.test.ts @@ -19,6 +19,7 @@ import { applyTopUpResult, closeHttpSession, createActiveSocketSession, + driveSseResponse, isExpectedSocketReceipt, managementInput, postTopUp, @@ -640,6 +641,39 @@ describe('VoucherManagement', () => { }) }) +describe('SseDriver', () => { + test('preserves application frames and consumes payment receipts', async () => { + const receipt = createSessionReceipt({ + acceptedCumulative: 2n, + challengeId: 'challenge-1', + channelId: `0x${'01'.repeat(32)}`, + spent: 2n, + }) + const acceptReceipt = vi.fn() + const response = new Response( + [ + ': keepalive\n\n', + 'event: custom\ndata: hello\n\n', + `event: payment-receipt\ndata: ${JSON.stringify(receipt)}\n\n`, + ].join(''), + ) + + const frames = [] + for await (const frame of driveSseResponse({ + async onNeedVoucher() {}, + onReceipt: acceptReceipt, + response, + })) + frames.push(frame) + + expect(frames).toEqual([ + { raw: ': keepalive\n\n' }, + { data: 'hello', raw: 'event: custom\ndata: hello\n\n' }, + ]) + expect(acceptReceipt).toHaveBeenCalledWith(receipt) + }) +}) + describe('WsDriver', () => { const channelId = '0x0000000000000000000000000000000000000000000000000000000000000001' as Hex.Hex const challenge: TempoSessionChallenge = { diff --git a/src/tempo/session/client/Transports.ts b/src/tempo/session/client/Transports.ts index cd0eb811..47b52481 100644 --- a/src/tempo/session/client/Transports.ts +++ b/src/tempo/session/client/Transports.ts @@ -777,29 +777,49 @@ export async function openSseSession( if (!isEventStream(response)) throw new Error('SSE response is not an event stream.') if (!response.body) throw new Error('Response has no body.') - return iterateSseResponse({ - challenge, - driver, - input, - onReceipt, + return iterateSseMessages({ + onNeedVoucher: (event) => handleSseNeedVoucher({ challenge, driver, input }, event), + onReceipt(receipt) { + driver.acceptReceipt(receipt) + onReceipt?.(receipt) + }, response, signal, }) } -type IterateSseResponseParameters = { - challenge: TempoSessionChallenge | null - driver: OpenSseSessionParameters - input: RequestInfo | URL - onReceipt?: ((receipt: SessionReceipt) => void) | undefined +/** Application SSE frame emitted after session payment events are handled. */ +export type SseResponseFrame = { + /** Parsed application data, when present. */ + data?: string | undefined + /** Original frame text, including its separator. */ + raw: string +} + +/** Inputs for driving an open paid SSE response. */ +export type DriveSseResponseParameters = { + /** Handles a request for the stream's next cumulative voucher. */ + onNeedVoucher(event: NeedVoucherEvent): Promise + /** Handles a payment receipt emitted by the stream. */ + onReceipt(receipt: SessionReceipt): void + /** Open SSE response to drive. */ response: Response + /** Abort signal used to stop reading the response. */ signal?: AbortSignal | undefined } -async function* iterateSseResponse( - parameters: IterateSseResponseParameters, -): AsyncGenerator { - const reader = parameters.response.body!.getReader() +async function* iterateSseMessages(parameters: DriveSseResponseParameters): AsyncGenerator { + for await (const frame of driveSseResponse(parameters)) { + if (frame.data !== undefined) yield frame.data + } +} + +/** Handles session payment events and yields application SSE frames unchanged. */ +export async function* driveSseResponse( + parameters: DriveSseResponseParameters, +): AsyncGenerator { + if (!parameters.response.body) throw new Error('Response has no body.') + const reader = parameters.response.body.getReader() const decoder = new TextDecoder() let buffer = '' @@ -817,18 +837,21 @@ async function* iterateSseResponse( for (const part of parts) { if (!part.trim()) continue const event = parseEvent(part) - if (!event) continue + const raw = `${part}\n\n` + if (!event) { + yield { raw } + continue + } switch (event.type) { case 'message': - yield event.data + yield { data: event.data, raw } break case 'payment-need-voucher': - await handleSseNeedVoucher(parameters, event.data) + await parameters.onNeedVoucher(event.data) break case 'payment-receipt': - parameters.driver.acceptReceipt(event.data) - parameters.onReceipt?.(event.data) + parameters.onReceipt(event.data) break } } @@ -839,7 +862,11 @@ async function* iterateSseResponse( } async function handleSseNeedVoucher( - parameters: IterateSseResponseParameters, + parameters: { + challenge: TempoSessionChallenge | null + driver: OpenSseSessionParameters + input: RequestInfo | URL + }, event: NeedVoucherEvent, ) { const channel = parameters.driver.getChannel()