refactor(session): extract SSE response driver

.changeset/session-sse-driver.md

@@ -0,0 +1,5 @@
+---
+'mppx': patch
+---
+
+Extracted session SSE payment handling into a response-preserving driver.

src/tempo/PublicExports.test-d.ts

@@ -83,6 +83,7 @@ test('tempo session public barrel hides internal session drivers', () => {
   expectTypeOf<SessionPublic>().not.toHaveProperty('settle')
   expectTypeOf<SessionClientPublic>().not.toHaveProperty('ChannelOps')
   expectTypeOf<SessionClientPublic>().not.toHaveProperty('Chain')
+  expectTypeOf<SessionClientPublic>().not.toHaveProperty('driveSseResponse')
   expectTypeOf<SessionServerPublic>().not.toHaveProperty('ChannelOps')
   expectTypeOf<SessionServerPublic>().not.toHaveProperty('Chain')
 })

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 = {

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<void>
+  /** 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<string> {
-  const reader = parameters.response.body!.getReader()
+async function* iterateSseMessages(parameters: DriveSseResponseParameters): AsyncGenerator<string> {
+  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<SseResponseFrame> {
+  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()