Handle paid Tempo session SSE fetches

.changeset/payment-aware-sse-session-fetch.md

@@ -0,0 +1,5 @@
+---
+'mppx': patch
+---
+
+Added paid Tempo session SSE handling to payment-aware fetch responses.

src/client/internal/Fetch.test.ts

@@ -10,6 +10,7 @@ import { rpcUrl } from '~test/tempo/rpc.js'
 import { accounts, asset, chain, client, http } from '~test/tempo/viem.js'
 
 import * as Fetch from './Fetch.js'
+import * as MethodResponse from './MethodResponse.js'
 
 const realm = 'api.example.com'
 const secretKey = 'test-secret-key-test-secret-key-32'
@@ -710,6 +711,26 @@ describe('Fetch.from: 402 retry path', () => {
     expect(headers.get('Authorization')).toBe('credential')
   })
 
+  test('lets methods wrap post-payment responses', async () => {
+    let callCount = 0
+    const wrapResponse: MethodResponse.Handler = async ({ response }) =>
+      new Response(`wrapped:${await response.text()}`, response)
+    const method = MethodResponse.set({ ...noopMethod }, wrapResponse)
+    const mockFetch: typeof globalThis.fetch = async () => {
+      callCount++
+      if (callCount === 1) return make402()
+      return new Response('OK', { status: 200 })
+    }
+    const fetch = Fetch.from({
+      fetch: mockFetch,
+      methods: [method],
+    })
+
+    await expect(
+      fetch('https://example.com/api').then((response) => response.text()),
+    ).resolves.toBe('wrapped:OK')
+  })
+
   test('chooses native Payment-auth from combined MPP and x402 HTTP 402 offers', async () => {
     let callCount = 0
     const calls: { init: RequestInit | undefined }[] = []

src/client/internal/Fetch.ts

@@ -6,6 +6,7 @@ import type { MaybePromise } from '../../internal/types.js'
 import type * as Method from '../../Method.js'
 import type * as z from '../../zod.js'
 import * as Transport from '../Transport.js'
+import * as MethodResponse from './MethodResponse.js'
 
 // We tag wrappers with a global symbol so we can recognize wrappers created by mppx,
 // even across multiple module instances/bundles. This lets restore() avoid clobbering
@@ -235,10 +236,9 @@ export function from<const methods extends readonly Method.AnyClient[]>(
         mi = selected.method
         if (challenge.expires) Expires.assert(challenge.expires, challenge.id)
 
-        const createCredential = memoizeCreateCredential(
-          (overrideContext?: AnyContextFor<methods>) =>
-            resolveCredential(selectedChallenge, selected.method, overrideContext ?? context),
-        )
+        const createCredentialForContext = (overrideContext?: AnyContextFor<methods>) =>
+          resolveCredential(selectedChallenge, selected.method, overrideContext ?? context)
+        const createCredential = memoizeCreateCredential(createCredentialForContext)
         const eventCredential = await events.emit(
           'challenge.received',
           createChallengeReceivedPayload({
@@ -283,6 +283,15 @@ export function from<const methods extends readonly Method.AnyClient[]>(
             { challenge: selectedChallenge },
           ),
         )
+        response = await handleMethodResponse({
+          challenge: selectedChallenge,
+          createCredential: createCredentialForContext as (context?: unknown) => Promise<string>,
+          credential,
+          fetch: baseFetch,
+          input: initialRequest.input,
+          method: selected.method,
+          response,
+        })
         if (response.ok)
           await events.emit(
             'payment.response',
@@ -322,6 +331,27 @@ export function from<const methods extends readonly Method.AnyClient[]>(
   return wrappedFetch as from.Fetch<methods>
 }
 
+async function handleMethodResponse(parameters: {
+  challenge: Challenge.Challenge
+  createCredential(context?: unknown): Promise<string>
+  credential: string
+  fetch: typeof globalThis.fetch
+  input: RequestInfo | URL
+  method: Method.AnyClient
+  response: Response
+}): Promise<Response> {
+  const handler = MethodResponse.get(parameters.method)
+  if (!handler) return parameters.response
+  return handler({
+    challenge: parameters.challenge,
+    createCredential: parameters.createCredential,
+    credential: parameters.credential,
+    fetch: parameters.fetch,
+    input: parameters.input,
+    response: parameters.response,
+  })
+}
+
 /** Union of all context types from all methods that have context schemas. */
 type AnyContextFor<methods extends readonly Method.AnyClient[]> = {
   [K in keyof methods]: NonNullable<methods[K]['context']> extends infer ctx

src/client/internal/MethodResponse.ts

@@ -0,0 +1,28 @@
+import type * as Challenge from '../../Challenge.js'
+import type * as Method from '../../Method.js'
+
+const handler = Symbol.for('mppx.client.method.response')
+
+export type Handler = (parameters: {
+  challenge: Challenge.Challenge
+  createCredential(context?: unknown): Promise<string>
+  credential: string
+  fetch: typeof globalThis.fetch
+  input: RequestInfo | URL
+  response: Response
+}) => Promise<Response> | Response
+
+type MethodWithResponseHandler = Method.AnyClient & {
+  [handler]?: Handler | undefined
+}
+
+/** Returns a method's internal response hook, when one is registered. */
+export function get(method: Method.AnyClient): Handler | undefined {
+  return (method as MethodWithResponseHandler)[handler]
+}
+
+/** Registers an internal response hook without changing the public method type. */
+export function set<const method extends Method.AnyClient>(method: method, value: Handler): method {
+  Object.defineProperty(method, handler, { configurable: true, value })
+  return method
+}

src/tempo/session/client/Session.ts

@@ -1,6 +1,7 @@
 import { type Address, parseUnits } from 'viem'
 import { tempo as tempo_chain } from 'viem/chains'
 
+import * as MethodResponse from '../../../client/internal/MethodResponse.js'
 import * as Constants from '../../../Constants.js'
 import * as Method from '../../../Method.js'
 import * as Account from '../../../viem/Account.js'
@@ -20,6 +21,7 @@ import {
   resolveRecoverContext,
   sessionContextSchema,
 } from './CredentialState.js'
+import { wrapSseFetchResponse } from './Transports.js'
 
 export { sessionContextSchema, type SessionContext } from './CredentialState.js'
 
@@ -52,7 +54,7 @@ export function session(parameters: session.Parameters = {}) {
   const store = channelStore ?? createChannelStore()
   const sink = { store, notifyUpdate: (entry: ChannelEntry) => onChannelUpdate?.(entry) }
 
-  return Method.toClient(Methods.session, {
+  const method = Method.toClient(Methods.session, {
     canHandleChallenge({ challenge }) {
       return (
         Constants.getMethodDetail(
@@ -98,6 +100,8 @@ export function session(parameters: session.Parameters = {}) {
       return serializeCredential(challenge, payload, resolved.chainId, account)
     },
   })
+  MethodResponse.set(method, wrapSseFetchResponse)
+  return method
 }
 
 /** Type helpers for the low-level TIP-1034 session client method. */

src/tempo/session/client/Transports.test.ts

@@ -9,9 +9,11 @@ import type { ChannelEntry } from '../client/ChannelOps.js'
 import type { SessionContext } from '../client/CredentialState.js'
 import {
   createSessionReceipt,
+  formatNeedVoucherEvent,
   serializeSessionReceipt,
   tip20ChannelEscrow,
   type ChannelDescriptor,
+  type SessionCredentialPayload,
 } from '../precompile/Protocol.js'
 import type { SessionSnapshot } from '../Snapshot.js'
 import { initialState, type SessionState } from './Runtime.js'
@@ -31,6 +33,7 @@ import {
   validateSocketCloseReadyReceipt,
   validateSocketPaymentReceipt,
   webSocketProbeUrl,
+  wrapSseFetchResponse,
   type TempoSessionChallenge,
   type TopUpRequirement,
 } from './Transports.js'
@@ -266,6 +269,91 @@ describe('HttpManagement', () => {
       expect(restoreCumulative).toHaveBeenCalledWith(channelId, 5n)
     })
 
+    test('wrapSseFetchResponse posts vouchers and hides session control events', async () => {
+      const paymentChallenge = challenge()
+      const openCredential = Credential.serialize({
+        challenge: paymentChallenge,
+        payload: {
+          action: 'open',
+          type: 'transaction',
+          channelId,
+          transaction: '0x01',
+          signature: '0x02',
+          descriptor,
+          cumulativeAmount: '5',
+        },
+      })
+      const createdContexts: SessionContext[] = []
+      const createCredential = vi.fn(async (context?: SessionContext) => {
+        if (!context?.action) throw new Error('expected manual context')
+        createdContexts.push(context)
+        const payload =
+          context.action === 'topUp'
+            ? {
+                action: 'topUp' as const,
+                type: 'transaction' as const,
+                channelId,
+                transaction: '0x03' as Hex.Hex,
+                descriptor,
+                additionalDeposit: context.additionalDepositRaw!,
+              }
+            : {
+                action: 'voucher' as const,
+                channelId,
+                descriptor,
+                cumulativeAmount: context.cumulativeAmountRaw!,
+                signature: '0x04' as Hex.Hex,
+              }
+        return Credential.serialize({ challenge: paymentChallenge, payload })
+      })
+      const postedPayloads: SessionCredentialPayload[] = []
+      const fetch = vi.fn(async (_input: RequestInfo | URL, init?: RequestInit) => {
+        postedPayloads.push(
+          Credential.deserialize<SessionCredentialPayload>(authorizationHeader(init)!).payload,
+        )
+        return new Response(null, { status: 204 })
+      })
+      const receipt = createSessionReceipt({
+        acceptedCumulative: 8n,
+        challengeId: paymentChallenge.id,
+        channelId,
+        spent: 8n,
+      })
+      const response = wrapSseFetchResponse({
+        challenge: paymentChallenge,
+        createCredential,
+        credential: openCredential,
+        fetch,
+        input: 'https://example.test/stream?cursor=1',
+        response: new Response(
+          [
+            'event: message\ndata: chunk-1\n\n',
+            formatNeedVoucherEvent({
+              acceptedCumulative: '5',
+              channelId,
+              deposit: '6',
+              requiredCumulative: '8',
+            }),
+            `event: payment-receipt\ndata: ${JSON.stringify(receipt)}\n\n`,
+            'event: message\ndata: chunk-2\n\n',
+          ].join(''),
+          { headers: { 'Content-Type': 'text/event-stream' } },
+        ),
+      })
+
+      await expect(response.text()).resolves.toBe(
+        'event: message\ndata: chunk-1\n\nevent: message\ndata: chunk-2\n\n',
+      )
+      expect(createdContexts.map((context) => context.action)).toEqual(['topUp', 'voucher'])
+      expect(createdContexts[0]).toMatchObject({ additionalDepositRaw: '2' })
+      expect(createdContexts[1]).toMatchObject({ cumulativeAmountRaw: '8' })
+      expect(fetch.mock.calls.map(([input]) => input.toString())).toEqual([
+        'https://example.test/stream',
+        'https://example.test/stream',
+      ])
+      expect(postedPayloads.map((payload) => payload.action)).toEqual(['topUp', 'voucher'])
+    })
+
     test('closeHttpSession posts a close credential and parses the receipt', async () => {
       const entry = channel()
       const createSessionCredential = vi.fn(async (_challenge, context: SessionContext) => {