From a5b80a3cc5fbc2283bfd07c226904cec60a378c4 Mon Sep 17 00:00:00 2001 From: Parv Ahuja <17094219+parvahuja@users.noreply.github.com> Date: Wed, 15 Jul 2026 14:56:23 -0700 Subject: [PATCH] feat(session): handle SSE responses in fetch --- .changeset/payment-aware-session-streams.md | 5 + src/client/internal/Fetch.test.ts | 29 +++++ src/client/internal/Fetch.ts | 57 +++++++++- src/client/internal/MethodResponse.ts | 39 +++++++ src/tempo/session/client/Session.test.ts | 84 +++++++++++++- src/tempo/session/client/Session.ts | 74 +++++++++++- src/tempo/session/client/SessionManager.ts | 2 + src/tempo/session/client/Transports.test.ts | 75 ++++++++++++ src/tempo/session/client/Transports.ts | 119 ++++++++++++++++++-- 9 files changed, 465 insertions(+), 19 deletions(-) create mode 100644 .changeset/payment-aware-session-streams.md create mode 100644 src/client/internal/MethodResponse.ts diff --git a/.changeset/payment-aware-session-streams.md b/.changeset/payment-aware-session-streams.md new file mode 100644 index 00000000..2dbb1670 --- /dev/null +++ b/.changeset/payment-aware-session-streams.md @@ -0,0 +1,5 @@ +--- +'mppx': patch +--- + +Added payment-aware Tempo session streams to client fetch responses. diff --git a/src/client/internal/Fetch.test.ts b/src/client/internal/Fetch.test.ts index 7f566109..434338e7 100644 --- a/src/client/internal/Fetch.test.ts +++ b/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)[] = [] diff --git a/src/client/internal/Fetch.ts b/src/client/internal/Fetch.ts index 7bb8a988..df7d73b6 100644 --- a/src/client/internal/Fetch.ts +++ b/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 boolean>() export type ClientEventMap< methods extends readonly Method.AnyClient[] = readonly Method.AnyClient[], @@ -174,7 +176,11 @@ export function from( // which can duplicate retries and make restore semantics fragile. const baseFetch = unwrapFetch(fetch) - const wrappedFetch = async (input: RequestInfo | URL, init?: from.RequestInit) => { + const request = async ( + input: RequestInfo | URL, + init: from.RequestInit | undefined, + canRefetch: boolean, + ): Promise => { 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 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( { 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( 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( throw error } } + const wrappedFetch = (input: RequestInfo | URL, init?: from.RequestInit) => + 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 = { 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( @@ -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]) diff --git a/src/client/internal/MethodResponse.ts b/src/client/internal/MethodResponse.ts new file mode 100644 index 00000000..67636ec1 --- /dev/null +++ b/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() + +/** 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) | undefined + response: Response + signal?: AbortSignal | undefined +} + +/** Internal client-method response adapter. */ +export type Handler = (parameters: HandlerParameters) => MaybePromise + +/** Registers an internal response adapter without changing the public method shape. */ +export function register( + 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 { + return Promise.resolve(handlers.get(method)?.(parameters) ?? parameters.response) +} diff --git a/src/tempo/session/client/Session.test.ts b/src/tempo/session/client/Session.test.ts index 344f713b..c727927e 100644 --- a/src/tempo/session/client/Session.test.ts +++ b/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( diff --git a/src/tempo/session/client/Session.ts b/src/tempo/session/client/Session.ts index 60099dd2..49036a87 100644 --- a/src/tempo/session/client/Session.ts +++ b/src/tempo/session/client/Session.ts @@ -1,7 +1,9 @@ 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 Credential from '../../../Credential.js' import * as Method from '../../../Method.js' import * as Account from '../../../viem/Account.js' import * as Client from '../../../viem/Client.js' @@ -11,6 +13,7 @@ import type { } from '../../client/ResolveAccount.js' import * as defaults from '../../internal/defaults.js' import * as Methods from '../../Methods.js' +import { isEventStream, requireSessionCredentialContext } from '../precompile/Protocol.js' import { serializeCredential, type ChannelEntry } from './ChannelOps.js' import { createChannelStore, type ChannelStore } from './ChannelStore.js' import { @@ -20,6 +23,15 @@ import { resolveRecoverContext, sessionContextSchema, } from './CredentialState.js' +import { assertWithinMaxDeposit } from './Runtime.js' +import { + handleSseNeedVoucher, + isTempoSessionChallenge, + managementInput, + postTopUp, + wrapSseResponse, + type SsePaymentDriver, +} from './Transports.js' export { sessionContextSchema, type SessionContext } from './CredentialState.js' @@ -52,7 +64,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 +110,66 @@ export function session(parameters: session.Parameters = {}) { return serializeCredential(challenge, payload, resolved.chainId, account) }, }) + + return MethodResponse.register( + method, + async ({ challenge, credential, fetch, headers, input, refetch, response, signal }) => { + if (!isTempoSessionChallenge(challenge)) return response + if (!isEventStream(response)) { + const credentialContext = requireSessionCredentialContext( + Credential.deserialize(credential).payload, + ) + if ( + credentialContext.action === 'open' && + headers.get('accept')?.toLowerCase().includes('text/event-stream') + ) + return (await refetch?.()) ?? response + return response + } + + const channelKey = ( + await resolveChallengeContext({ + challenge, + escrowOverride, + getClient, + }) + ).key + let channel = await store.get(channelKey) + const driver = { + assertVoucherWithinLocalLimit: (cumulativeAmount) => + assertWithinMaxDeposit(cumulativeAmount, maxDeposit), + createSessionCredential: (challenge, context) => + method.createCredential({ challenge, context }), + fetch, + getChannel: () => channel ?? null, + managementInput, + async topUpIfNeeded({ channelId, deposit, requiredCumulative }) { + if (requiredCumulative <= deposit || !channel) return + const additionalDeposit = requiredCumulative - deposit + await postTopUp({ + additionalDeposit, + challenge, + channel, + channelId, + createSessionCredential: (challenge, context) => + method.createCredential({ challenge, context }), + fetch, + input, + }) + channel.deposit += additionalDeposit + await store.set(channel) + sink.notifyUpdate(channel) + }, + } satisfies SsePaymentDriver + + return wrapSseResponse({ + onNeedVoucher: (event) => handleSseNeedVoucher({ challenge, driver, input }, event), + onReceipt() {}, + response, + signal, + }) + }, + ) } /** Type helpers for the low-level TIP-1034 session client method. */ diff --git a/src/tempo/session/client/SessionManager.ts b/src/tempo/session/client/SessionManager.ts index 1e2a76f3..50be43bf 100644 --- a/src/tempo/session/client/SessionManager.ts +++ b/src/tempo/session/client/SessionManager.ts @@ -3,6 +3,7 @@ import { parseUnits, type Address } from 'viem' import * as Challenge from '../../../Challenge.js' import * as Fetch from '../../../client/internal/Fetch.js' +import * as MethodResponse from '../../../client/internal/MethodResponse.js' import * as Constants from '../../../Constants.js' import type * as Account from '../../../viem/Account.js' import type * as Client from '../../../viem/Client.js' @@ -293,6 +294,7 @@ export function sessionManager(parameters: sessionManager.Parameters): SessionMa } }, }) + MethodResponse.unregister(method) const chargeMethod = chargePlugin({ account: parameters.account, getClient: parameters.client ? () => parameters.client! : parameters.getClient, diff --git a/src/tempo/session/client/Transports.test.ts b/src/tempo/session/client/Transports.test.ts index 75622645..ce5ece03 100644 --- a/src/tempo/session/client/Transports.test.ts +++ b/src/tempo/session/client/Transports.test.ts @@ -32,6 +32,7 @@ import { validateSocketCloseReadyReceipt, validateSocketPaymentReceipt, webSocketProbeUrl, + wrapSseResponse, type TempoSessionChallenge, type TopUpRequirement, } from './Transports.js' @@ -672,6 +673,80 @@ describe('SseDriver', () => { ]) expect(acceptReceipt).toHaveBeenCalledWith(receipt) }) + + test('wraps standard CRLF streams and preserves response metadata', async () => { + const receipt = createSessionReceipt({ + acceptedCumulative: 2n, + challengeId: 'challenge-1', + channelId: `0x${'01'.repeat(32)}`, + spent: 2n, + }) + const encoder = new TextEncoder() + const source = new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode(': keepalive\r\n\r')) + controller.enqueue( + encoder.encode( + [ + '\nevent: custom\r\ndata: hello\r\n\r\n', + `event: payment-receipt\r\ndata: ${JSON.stringify(receipt)}\r\n\r\n`, + ].join(''), + ), + ) + controller.close() + }, + }), + { + headers: { 'content-length': '100', 'content-type': 'text/event-stream' }, + status: 201, + statusText: 'Created', + }, + ) + Object.defineProperty(source, 'url', { value: 'https://example.com/stream' }) + + const response = wrapSseResponse({ + async onNeedVoucher() {}, + onReceipt() {}, + response: source, + }) + + expect(response.status).toBe(201) + expect(response.statusText).toBe('Created') + expect(response.url).toBe('https://example.com/stream') + expect(response.headers.get('content-type')).toBe('text/event-stream') + expect(response.headers.has('content-length')).toBe(false) + expect(await response.text()).toBe(': keepalive\r\n\r\nevent: custom\r\ndata: hello\r\n\r\n') + }) + + test.each(['consumer', 'signal'] as const)( + 'cancels the source stream from %s', + async (source) => { + const abortController = new AbortController() + const cancel = vi.fn() + const sourceResponse = new Response( + new ReadableStream({ + cancel, + start(controller) { + controller.enqueue(new TextEncoder().encode('data: first\n\n')) + }, + }), + ) + const response = wrapSseResponse({ + async onNeedVoucher() {}, + onReceipt() {}, + response: sourceResponse, + signal: abortController.signal, + }) + + const reader = response.body!.getReader() + await reader.read() + if (source === 'signal') abortController.abort() + else await reader.cancel() + + await vi.waitFor(() => expect(cancel).toHaveBeenCalledOnce()) + }, + ) }) describe('WsDriver', () => { diff --git a/src/tempo/session/client/Transports.ts b/src/tempo/session/client/Transports.ts index 47b52481..8dedbe91 100644 --- a/src/tempo/session/client/Transports.ts +++ b/src/tempo/session/client/Transports.ts @@ -748,6 +748,17 @@ export type OpenSseSessionParameters = { topUpIfNeeded(parameters: TopUpRequirement): Promise } +/** Session payment operations shared with fetch-backed streams. */ +export type SsePaymentDriver = Pick< + OpenSseSessionParameters, + | 'assertVoucherWithinLocalLimit' + | 'createSessionCredential' + | 'fetch' + | 'getChannel' + | 'managementInput' + | 'topUpIfNeeded' +> + /** * Opens an auto-driving paid SSE stream. * @@ -822,22 +833,32 @@ export async function* driveSseResponse( const reader = parameters.response.body.getReader() const decoder = new TextDecoder() let buffer = '' + let complete = false + const cancel = () => { + void reader.cancel(parameters.signal?.reason).catch(() => undefined) + } + parameters.signal?.addEventListener('abort', cancel, { once: true }) + if (parameters.signal?.aborted) cancel() try { while (true) { if (parameters.signal?.aborted) break const { done, value } = await reader.read() - if (done) break - - buffer += decoder.decode(value, { stream: true }) - const parts = buffer.split('\n\n') - buffer = parts.pop()! - - for (const part of parts) { + if (done) { + complete = true + buffer += decoder.decode() + } else buffer += decoder.decode(value, { stream: true }) + + while (true) { + const separator = findSseSeparator(buffer, done) + if (!separator) break + const end = separator.index + separator.length + const part = buffer.slice(0, separator.index) + const raw = buffer.slice(0, end) + buffer = buffer.slice(end) if (!part.trim()) continue - const event = parseEvent(part) - const raw = `${part}\n\n` + const event = parseEvent(part.replace(/\r\n|\r/g, '\n')) if (!event) { yield { raw } continue @@ -855,16 +876,92 @@ export async function* driveSseResponse( break } } + if (done) break } } finally { + parameters.signal?.removeEventListener('abort', cancel) + if (!complete) await reader.cancel().catch(() => undefined) reader.releaseLock() } } -async function handleSseNeedVoucher( +function findSseSeparator( + value: string, + complete: boolean, +): { index: number; length: number } | undefined { + for (let index = 0; index < value.length; index++) { + const first = lineEndingLength(value, index, complete) + if (!first) continue + const second = lineEndingLength(value, index + first, complete) + if (second) return { index, length: first + second } + index += first - 1 + } + return undefined +} + +function lineEndingLength(value: string, index: number, complete: boolean): number { + if (value[index] === '\n') return 1 + if (value[index] !== '\r') return 0 + if (value[index + 1] === '\n') return 2 + if (index + 1 === value.length && !complete) return 0 + return 1 +} + +/** Wraps an SSE response while handling session payment frames in-band. */ +export function wrapSseResponse(parameters: DriveSseResponseParameters): Response { + const abortController = new AbortController() + const abort = () => abortController.abort(parameters.signal?.reason) + parameters.signal?.addEventListener('abort', abort, { once: true }) + if (parameters.signal?.aborted) abort() + + const iterator = driveSseResponse({ ...parameters, signal: abortController.signal }) + const encoder = new TextEncoder() + const headers = new Headers(parameters.response.headers) + headers.delete('content-length') + const cleanup = () => parameters.signal?.removeEventListener('abort', abort) + + const response = new Response( + new ReadableStream({ + async pull(controller) { + try { + const frame = await iterator.next() + if (frame.done) { + cleanup() + return controller.close() + } + controller.enqueue(encoder.encode(frame.value.raw)) + } catch (error) { + cleanup() + throw error + } + }, + async cancel() { + cleanup() + abortController.abort() + await iterator.return(undefined) + }, + }), + { + headers, + status: parameters.response.status, + statusText: parameters.response.statusText, + }, + ) + + for (const property of ['redirected', 'type', 'url'] as const) + Object.defineProperty(response, property, { + configurable: true, + get: () => parameters.response[property], + }) + + return response +} + +/** Handles an in-band voucher request using the shared SSE payment driver. */ +export async function handleSseNeedVoucher( parameters: { challenge: TempoSessionChallenge | null - driver: OpenSseSessionParameters + driver: SsePaymentDriver input: RequestInfo | URL }, event: NeedVoucherEvent,