feat(session): handle SSE responses in fetch

.changeset/payment-aware-session-streams.md

@@ -0,0 +1,5 @@
+---
+'mppx': patch
+---
+
+Added payment-aware Tempo session streams to client 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'
@@ -391,6 +392,34 @@ function encodeRawPaymentRequiredHeader(value: unknown): string {
   return Base64.fromString(JSON.stringify(value))
 }
 
+describe('Fetch.from: method responses', () => {
+  test('handles successful paid responses without touching free responses', async () => {
+    const method = { ...noopMethod }
+    const handle = vi.fn(
+      async ({ response }: MethodResponse.HandlerParameters) =>
+        new Response(`handled:${await response.text()}`, response),
+    )
+    MethodResponse.register(method, handle)
+
+    let paidResponse: Response | undefined
+    const mockFetch: typeof globalThis.fetch = async (input, init) => {
+      if (input.toString().endsWith('/free')) return new Response('free')
+      if (!new Headers(init?.headers).has('Authorization')) return make402()
+      paidResponse = new Response('paid')
+      vi.spyOn(paidResponse, 'clone')
+      return paidResponse
+    }
+    const fetch = Fetch.from({ fetch: mockFetch, methods: [method] })
+
+    expect(await (await fetch('https://example.com/free')).text()).toBe('free')
+    expect(handle).not.toHaveBeenCalled()
+    expect(await (await fetch('https://example.com/paid')).text()).toBe('handled:paid')
+    expect(handle).toHaveBeenCalledOnce()
+    expect(handle).toHaveBeenCalledWith(expect.objectContaining({ credential: 'credential' }))
+    expect(paidResponse?.clone).not.toHaveBeenCalled()
+  })
+})
+
 describe('Fetch.from: init passthrough (non-402)', () => {
   test('preserves init object identity while adding Accept-Payment', async () => {
     const receivedInits: (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
@@ -18,6 +19,7 @@ type WrappedFetch = typeof globalThis.fetch & {
 }
 
 let originalFetch: typeof globalThis.fetch | undefined
+const eventObserverChecks = new WeakMap<object, (name: keyof ClientEventMap) => boolean>()
 
 export type ClientEventMap<
   methods extends readonly Method.AnyClient[] = readonly Method.AnyClient[],
@@ -174,7 +176,11 @@ export function from<const methods extends readonly Method.AnyClient[]>(
   // which can duplicate retries and make restore semantics fragile.
   const baseFetch = unwrapFetch(fetch)
 
-  const wrappedFetch = async (input: RequestInfo | URL, init?: from.RequestInit<methods>) => {
+  const request = async (
+    input: RequestInfo | URL,
+    init: from.RequestInit<methods> | undefined,
+    canRefetch: boolean,
+  ): Promise<Response> => {
     const callerHeaders = getCallerHeaders(input, init?.headers)
     const hasExplicitAcceptPayment = callerHeaders.has(Constants.Headers.acceptPayment)
     const paymentPreferences = resolvePaymentPreferences(callerHeaders, resolvedAcceptPayment)
@@ -272,8 +278,13 @@ export function from<const methods extends readonly Method.AnyClient[]>(
           }),
         )
 
+        const paymentInput = resolvePaymentRetryInput(
+          response,
+          initialRequest.input,
+          initialRequest.input,
+        )
         response = await baseFetch(
-          resolvePaymentRetryInput(response, initialRequest.input, initialRequest.input),
+          paymentInput,
           transport.setCredential(
             {
               ...fetchInit,
@@ -283,7 +294,8 @@ export function from<const methods extends readonly Method.AnyClient[]>(
             { challenge: selectedChallenge },
           ),
         )
-        if (response.ok)
+        // Cloning an unobserved stream would tee and buffer it indefinitely.
+        if (response.ok && hasEventObservers(events, 'payment.response'))
           await events.emit(
             'payment.response',
             createPaymentResponsePayload({
@@ -295,8 +307,23 @@ export function from<const methods extends readonly Method.AnyClient[]>(
               response,
             }),
           )
-        if (!(await transport.isPaymentRequired(response, transportRequest as never)))
-          return response
+        if (!(await transport.isPaymentRequired(response, transportRequest as never))) {
+          if (!response.ok) return response
+          return MethodResponse.handle(selected.method, {
+            challenge: selectedChallenge,
+            credential,
+            fetch: baseFetch,
+            headers: initialRequest.headers,
+            input: paymentInput,
+            ...(canRefetch
+              ? {
+                  refetch: () => request(cloneRequestInput(initialRequest.input), init, false),
+                }
+              : {}),
+            response,
+            signal: resolveRequestSignal(input, init),
+          })
+        }
       }
       return response
     } catch (error) {
@@ -315,6 +342,8 @@ export function from<const methods extends readonly Method.AnyClient[]>(
       throw error
     }
   }
+  const wrappedFetch = (input: RequestInfo | URL, init?: from.RequestInit<methods>) =>
+    request(input, init, true)
 
   // Record the wrapped target so future polyfill() / restore() calls can detect origin
   // and safely unwrap only mppx-installed wrappers.
@@ -491,7 +520,7 @@ export function createEventDispatcher<
     }
   }
 
-  return {
+  const dispatcher: ClientEventDispatcher<methods, response> = {
     async emit(name, payload) {
       switch (name) {
         case 'challenge.received': {
@@ -525,6 +554,15 @@ export function createEventDispatcher<
     },
     on,
   }
+  eventObserverChecks.set(
+    dispatcher,
+    (name) => handlers[name as keyof typeof handlers].size > 0 || handlers['*'].size > 0,
+  )
+  return dispatcher
+}
+
+function hasEventObservers(dispatcher: ClientEventDispatcher, name: keyof ClientEventMap): boolean {
+  return eventObserverChecks.get(dispatcher)?.(name) ?? true
 }
 
 async function emitChallengeReceived<methods extends readonly Method.AnyClient[], response>(
@@ -812,6 +850,13 @@ function cloneRequestInput(input: RequestInfo | URL): RequestInfo | URL {
   }
 }
 
+function resolveRequestSignal(
+  input: RequestInfo | URL,
+  init: RequestInit | undefined,
+): AbortSignal | undefined {
+  return init?.signal ?? (input instanceof Request ? input.signal : undefined)
+}
+
 /** @internal */
 function isWrappedFetch(fetch: typeof globalThis.fetch): fetch is WrappedFetch {
   return Boolean((fetch as WrappedFetch)[MPPX_FETCH_WRAPPER])

src/client/internal/MethodResponse.ts

@@ -0,0 +1,39 @@
+import type * as Challenge from '../../Challenge.js'
+import type { MaybePromise } from '../../internal/types.js'
+import type * as Method from '../../Method.js'
+
+const handlers = new WeakMap<Method.AnyClient, Handler>()
+
+/** Inputs available when a client method handles a successful paid response. */
+export type HandlerParameters = {
+  challenge: Challenge.Challenge
+  credential: string
+  fetch: typeof globalThis.fetch
+  headers: Headers
+  input: RequestInfo | URL
+  refetch?: (() => Promise<Response>) | undefined
+  response: Response
+  signal?: AbortSignal | undefined
+}
+
+/** Internal client-method response adapter. */
+export type Handler = (parameters: HandlerParameters) => MaybePromise<Response>
+
+/** Registers an internal response adapter without changing the public method shape. */
+export function register<const method extends Method.AnyClient>(
+  method: method,
+  handler: Handler,
+): method {
+  handlers.set(method, handler)
+  return method
+}
+
+/** Removes response handling from a method whose caller owns the response lifecycle. */
+export function unregister(method: Method.AnyClient): void {
+  handlers.delete(method)
+}
+
+/** Lets the selected client method handle a successful paid response. */
+export function handle(method: Method.AnyClient, parameters: HandlerParameters): Promise<Response> {
+  return Promise.resolve(handlers.get(method)?.(parameters) ?? parameters.response)
+}

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

@@ -3,7 +3,8 @@ import { privateKeyToAccount } from 'viem/accounts'
 import { Account as TempoAccount, Secp256k1, Transaction } from 'viem/tempo'
 import { describe, expect, test } from 'vp/test'
 
-import type { Challenge } from '../../../Challenge.js'
+import { serialize as serializeChallenge, type Challenge } from '../../../Challenge.js'
+import * as Fetch from '../../../client/internal/Fetch.js'
 import * as Constants from '../../../Constants.js'
 import * as Credential from '../../../Credential.js'
 import * as z from '../../../zod.js'
@@ -13,6 +14,7 @@ import { escrowAbi } from '../precompile/escrow.abi.js'
 import { tip20ChannelEscrow } from '../precompile/Protocol.js'
 import * as Types from '../precompile/Protocol.js'
 import * as Voucher from '../precompile/Voucher.js'
+import { createChannelStore } from './ChannelStore.js'
 import { session } from './Session.js'
 
 const account = privateKeyToAccount(
@@ -126,6 +128,86 @@ describe('precompile client session', () => {
     ).toBe(false)
   })
 
+  test('drives paid SSE responses with a supplied credential', async () => {
+    const challenge = makeChallenge({
+      suggestedDeposit: '100',
+      methodDetails: {
+        chainId,
+        escrowContract: tip20ChannelEscrow,
+        sessionProtocol: Constants.SessionProtocols.v2,
+      },
+    })
+    const channelStore = createChannelStore()
+    const parameters = {
+      account,
+      channelStore,
+      decimals: 0,
+      getClient: () => client,
+      maxDeposit: '300',
+    } as const
+    // The response handler must follow the credential selected by `onChallenge`.
+    const externalCredential = await session(parameters).createCredential({
+      challenge,
+      context: {},
+    })
+    let useExternalCredential = true
+    const actions: Types.SessionCredentialPayload['action'][] = []
+    const rawFetch: typeof globalThis.fetch = async (_input, init) => {
+      const authorization = new Headers(init?.headers).get(Constants.Headers.authorization)
+      if (!authorization)
+        return new Response(null, {
+          status: 402,
+          headers: { [Constants.Headers.wwwAuthenticate]: serializeChallenge(challenge) },
+        })
+
+      const payload = deserialize(authorization)
+      actions.push(payload.action)
+      if (init?.method === 'POST' || payload.action === 'open')
+        return new Response(null, { status: 204 })
+      if (payload.action !== 'voucher') throw new Error('expected voucher')
+
+      const receipt = Types.createSessionReceipt({
+        acceptedCumulative: 200n,
+        challengeId: challenge.id,
+        channelId: payload.channelId,
+        spent: 200n,
+      })
+      return new Response(
+        [
+          Types.formatMessageEvent('first'),
+          Types.formatNeedVoucherEvent({
+            acceptedCumulative: '100',
+            channelId: payload.channelId,
+            deposit: '100',
+            requiredCumulative: '200',
+          }),
+          Types.formatMessageEvent('second'),
+          Types.formatReceiptEvent(receipt),
+        ].join(''),
+        { headers: { 'content-type': 'text/event-stream' } },
+      )
+    }
+    const fetch = Fetch.from({
+      fetch: rawFetch,
+      methods: [session(parameters)],
+      async onChallenge() {
+        if (!useExternalCredential) return undefined
+        useExternalCredential = false
+        return externalCredential
+      },
+    })
+
+    const response = await fetch('https://example.com/stream', {
+      headers: { accept: 'text/event-stream' },
+    })
+
+    expect(response.headers.get('content-type')).toBe('text/event-stream')
+    expect(await response.text()).toBe(
+      `${Types.formatMessageEvent('first')}${Types.formatMessageEvent('second')}`,
+    )
+    expect(actions).toEqual(['open', 'voucher', 'topUp', 'voucher'])
+  })
+
   test('opens for the current amount without client deposit configuration', async () => {
     const method = session({ account, getClient: () => client })
     const payload = deserialize(