From d9d40eac22060efedb4909f230d3fc52b2a49e64 Mon Sep 17 00:00:00 2001 From: diegocintron Date: Fri, 19 Jun 2026 11:42:12 -0600 Subject: [PATCH 1/2] feat(core): real-time escrow event subscription (SSE + reconnection) --- .../connect-core/src/escrow-events.test.ts | 217 +++++++++++++++ packages/connect-core/src/escrow-events.ts | 257 ++++++++++++++++++ packages/connect-core/src/index.ts | 63 +++++ packages/connect-core/src/sse.ts | 81 ++++++ packages/connect-react/src/index.ts | 6 + 5 files changed, 624 insertions(+) create mode 100644 packages/connect-core/src/escrow-events.test.ts create mode 100644 packages/connect-core/src/escrow-events.ts create mode 100644 packages/connect-core/src/sse.ts diff --git a/packages/connect-core/src/escrow-events.test.ts b/packages/connect-core/src/escrow-events.test.ts new file mode 100644 index 0000000..40fcac7 --- /dev/null +++ b/packages/connect-core/src/escrow-events.test.ts @@ -0,0 +1,217 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { EscrowEventSubscriber } from './escrow-events.js'; +import { parseSseBlock, readSseStream } from './sse.js'; + +function encodeSse(block: string): Uint8Array { + return new TextEncoder().encode(block); +} + +describe('sse parser', () => { + it('parses event blocks with id, event, and data', () => { + const message = parseSseBlock( + 'id: cursor-1\nevent: escrow.funded\ndata: {"escrowId":"esc_1","occurredAt":"2024-01-01T00:00:00.000Z"}', + ); + + expect(message).toEqual({ + id: 'cursor-1', + event: 'escrow.funded', + data: '{"escrowId":"esc_1","occurredAt":"2024-01-01T00:00:00.000Z"}', + }); + }); + + it('ignores heartbeat comments', () => { + expect(parseSseBlock(': heartbeat')).toBeNull(); + }); +}); + +describe('escrow event subscription', () => { + const sleep = vi.fn(async () => {}); + const gatewayUrl = 'https://gateway.example'; + const publishableKey = 'pk_test_123'; + const clientSecret = 'cs_session_1.signature'; + + beforeEach(() => { + vi.stubGlobal('fetch', vi.fn()); + sleep.mockClear(); + }); + + afterEach(() => { + vi.unstubAllGlobals(); + }); + + it('maps lifecycle events to escrow milestones', async () => { + const handler = vi.fn(); + const subscriber = new EscrowEventSubscriber({ + gatewayUrl, + publishableKey, + clientSecret, + sleep, + }); + + subscriber.on('escrow.funded', handler); + + const stream = new ReadableStream({ + start(controller) { + controller.enqueue( + encodeSse( + 'id: cursor-1\nevent: escrow.funded\ndata: {"escrowId":"esc_1","occurredAt":"2024-01-01T00:00:00.000Z"}\n\n', + ), + ); + controller.enqueue( + encodeSse( + 'id: cursor-2\nevent: fiat.reported\ndata: {"escrowId":"esc_1","occurredAt":"2024-01-01T00:05:00.000Z"}\n\n', + ), + ); + controller.close(); + }, + }); + + vi.mocked(fetch).mockResolvedValue({ + ok: true, + status: 200, + body: stream, + } as Response); + + await vi.waitFor(() => expect(handler).toHaveBeenCalledTimes(1)); + + expect(handler).toHaveBeenCalledWith({ + cursor: 'cursor-1', + type: 'escrow.funded', + escrowId: 'esc_1', + milestone: 'funded', + occurredAt: '2024-01-01T00:00:00.000Z', + }); + + subscriber.close(); + }); + + it('reconnects with cursor and replays missed events without duplicates', async () => { + const handler = vi.fn(); + const subscriber = new EscrowEventSubscriber({ + gatewayUrl, + publishableKey, + clientSecret, + sleep, + maxReconnectAttempts: 2, + }); + + subscriber.on('released', handler); + + const firstStream = new ReadableStream({ + start(controller) { + controller.enqueue( + encodeSse( + 'id: cursor-1\nevent: escrow.funded\ndata: {"escrowId":"esc_1","occurredAt":"2024-01-01T00:00:00.000Z"}\n\n', + ), + ); + controller.close(); + }, + }); + + const secondStream = new ReadableStream({ + start(controller) { + controller.enqueue( + encodeSse( + 'id: cursor-2\nevent: fiat.reported\ndata: {"escrowId":"esc_1","occurredAt":"2024-01-01T00:05:00.000Z"}\n\n', + ), + ); + controller.enqueue( + encodeSse( + 'id: cursor-3\nevent: released\ndata: {"escrowId":"esc_1","occurredAt":"2024-01-01T00:10:00.000Z"}\n\n', + ), + ); + controller.close(); + }, + }); + + vi.mocked(fetch) + .mockResolvedValueOnce({ + ok: true, + status: 200, + body: firstStream, + } as Response) + .mockResolvedValueOnce({ + ok: true, + status: 200, + body: secondStream, + } as Response); + + await vi.waitFor(() => expect(handler).toHaveBeenCalledTimes(1)); + await vi.waitFor(() => expect(vi.mocked(fetch).mock.calls.length).toBeGreaterThanOrEqual(2)); + + expect(handler).toHaveBeenCalledWith({ + cursor: 'cursor-3', + type: 'released', + escrowId: 'esc_1', + milestone: 'released', + occurredAt: '2024-01-01T00:10:00.000Z', + }); + + const reconnectUrl = vi.mocked(fetch).mock.calls.find( + (call, index) => index > 0 && String(call[0]).includes('cursor=cursor-1'), + ); + expect(reconnectUrl).toBeDefined(); + + subscriber.close(); + }); + + it('filters events by escrow id', async () => { + const handler = vi.fn(); + const subscriber = new EscrowEventSubscriber({ + gatewayUrl, + publishableKey, + clientSecret, + sleep, + }); + + subscriber.on('disputed', handler, { escrowId: 'esc_target' }); + + const stream = new ReadableStream({ + start(controller) { + controller.enqueue( + encodeSse( + 'id: cursor-1\nevent: disputed\ndata: {"escrowId":"esc_other","occurredAt":"2024-01-01T00:00:00.000Z"}\n\n', + ), + ); + controller.enqueue( + encodeSse( + 'id: cursor-2\nevent: disputed\ndata: {"escrowId":"esc_target","occurredAt":"2024-01-01T00:01:00.000Z"}\n\n', + ), + ); + controller.close(); + }, + }); + + vi.mocked(fetch).mockResolvedValue({ + ok: true, + status: 200, + body: stream, + } as Response); + + await vi.waitFor(() => expect(handler).toHaveBeenCalledTimes(1)); + + expect(handler).toHaveBeenCalledWith( + expect.objectContaining({ escrowId: 'esc_target', type: 'disputed' }), + ); + + subscriber.close(); + }); +}); + +describe('readSseStream', () => { + it('reads chunked sse payloads', async () => { + const messages: string[] = []; + const stream = new ReadableStream({ + start(controller) { + controller.enqueue(encodeSse('id: 1\nevent: ping\ndata: {}\n\n')); + controller.close(); + }, + }); + + await readSseStream(stream, (message) => { + messages.push(message.event ?? ''); + }); + + expect(messages).toEqual(['ping']); + }); +}); diff --git a/packages/connect-core/src/escrow-events.ts b/packages/connect-core/src/escrow-events.ts new file mode 100644 index 0000000..4711f2e --- /dev/null +++ b/packages/connect-core/src/escrow-events.ts @@ -0,0 +1,257 @@ +import { PUBLISHABLE_KEY_HEADER } from './http.js'; +import { readSseStream, type SseMessage } from './sse.js'; + +export const ESCROW_EVENT_NAMES = [ + 'escrow.funded', + 'fiat.reported', + 'released', + 'disputed', +] as const; + +export type EscrowEventName = (typeof ESCROW_EVENT_NAMES)[number]; + +/** Maps to Pacto P2P `escrow_milestones` lifecycle states. */ +export type EscrowMilestone = 'funded' | 'fiat_reported' | 'released' | 'disputed'; + +export interface EscrowEvent { + cursor: string; + type: EscrowEventName; + escrowId: string; + milestone: EscrowMilestone; + occurredAt: string; +} + +export type EscrowEventHandler = (event: EscrowEvent) => void; + +export interface EscrowSubscribeOptions { + escrowId?: string; +} + +export interface SessionConnectionConfig { + gatewayUrl: string; + publishableKey: string; + clientSecret: string; + origin?: string; + baseDelayMs?: number; + maxReconnectAttempts?: number; + sleep?: (ms: number) => Promise; +} + +const MILESTONE_BY_EVENT: Record = { + 'escrow.funded': 'funded', + 'fiat.reported': 'fiat_reported', + released: 'released', + disputed: 'disputed', +}; + +const DEFAULT_BASE_DELAY_MS = 500; +const DEFAULT_MAX_RECONNECT_ATTEMPTS = 5; + +function defaultSleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +function isEscrowEventName(value: string): value is EscrowEventName { + return (ESCROW_EVENT_NAMES as readonly string[]).includes(value); +} + +function getBackoffDelay(attempt: number, baseDelayMs: number): number { + const exponential = baseDelayMs * 2 ** attempt; + const jitter = Math.floor(Math.random() * baseDelayMs); + return exponential + jitter; +} + +function mapToEscrowEvent( + cursor: string | undefined, + type: EscrowEventName, + payload: Record, +): EscrowEvent | null { + const escrowId = typeof payload.escrowId === 'string' ? payload.escrowId : undefined; + const occurredAt = + typeof payload.occurredAt === 'string' + ? payload.occurredAt + : typeof payload.timestamp === 'string' + ? payload.timestamp + : undefined; + + if (!escrowId || !occurredAt || !cursor) { + return null; + } + + return { + cursor, + type, + escrowId, + milestone: MILESTONE_BY_EVENT[type], + occurredAt, + }; +} + +type HandlerEntry = { + handler: EscrowEventHandler; + escrowId?: string; +}; + +export class EscrowEventSubscriber { + private readonly handlers = new Map>(); + private lastCursor?: string; + private closed = false; + private connecting = false; + private reconnectAttempt = 0; + private readonly seenCursors = new Set(); + private readonly filterEscrowId?: string; + + constructor( + private readonly config: SessionConnectionConfig, + options?: EscrowSubscribeOptions, + ) { + this.filterEscrowId = options?.escrowId; + } + + on(event: EscrowEventName, handler: EscrowEventHandler, options?: EscrowSubscribeOptions): void { + const entries = this.handlers.get(event) ?? new Set(); + entries.add({ handler, escrowId: options?.escrowId ?? this.filterEscrowId }); + this.handlers.set(event, entries); + + if (!this.closed && !this.connecting) { + this.connectLoop(); + } + } + + off(event: EscrowEventName, handler: EscrowEventHandler): void { + const entries = this.handlers.get(event); + if (!entries) { + return; + } + + for (const entry of entries) { + if (entry.handler === handler) { + entries.delete(entry); + } + } + + if (entries.size === 0) { + this.handlers.delete(event); + } + } + + close(): void { + this.closed = true; + this.handlers.clear(); + } + + private hasHandlers(): boolean { + return this.handlers.size > 0; + } + + private async connectLoop(): Promise { + if (this.closed || !this.hasHandlers()) { + return; + } + + this.connecting = true; + + while (!this.closed && this.hasHandlers()) { + try { + await this.connectOnce(); + this.reconnectAttempt = 0; + } catch { + if (this.closed || !this.hasHandlers()) { + break; + } + + if (this.reconnectAttempt >= this.maxReconnectAttempts()) { + break; + } + + const delay = getBackoffDelay(this.reconnectAttempt, this.baseDelayMs()); + await this.sleep()(delay); + this.reconnectAttempt += 1; + } + } + + this.connecting = false; + } + + private async connectOnce(): Promise { + const url = new URL(`${this.config.gatewayUrl}/v1/escrows/events`); + if (this.filterEscrowId) { + url.searchParams.set('escrowId', this.filterEscrowId); + } + if (this.lastCursor) { + url.searchParams.set('cursor', this.lastCursor); + } + + const headers: Record = { + Authorization: `Bearer ${this.config.clientSecret}`, + [PUBLISHABLE_KEY_HEADER]: this.config.publishableKey, + Accept: 'text/event-stream', + }; + + if (this.config.origin) { + headers.Origin = this.config.origin; + } + + const response = await fetch(url.toString(), { method: 'GET', headers }); + + if (!response.ok || !response.body) { + throw new Error(`Escrow event stream failed with status ${response.status}`); + } + + await readSseStream(response.body, (message) => this.handleMessage(message)); + } + + private handleMessage(message: SseMessage): void { + if (message.id) { + if (this.seenCursors.has(message.id)) { + return; + } + this.seenCursors.add(message.id); + this.lastCursor = message.id; + } + + if (!message.event || !message.data) { + return; + } + + if (!isEscrowEventName(message.event)) { + return; + } + + let payload: Record; + try { + payload = JSON.parse(message.data) as Record; + } catch { + return; + } + + const escrowEvent = mapToEscrowEvent(message.id, message.event, payload); + if (!escrowEvent) { + return; + } + + const entries = this.handlers.get(message.event); + if (!entries) { + return; + } + + for (const entry of entries) { + if (entry.escrowId && entry.escrowId !== escrowEvent.escrowId) { + continue; + } + entry.handler(escrowEvent); + } + } + + private baseDelayMs(): number { + return this.config.baseDelayMs ?? DEFAULT_BASE_DELAY_MS; + } + + private maxReconnectAttempts(): number { + return this.config.maxReconnectAttempts ?? DEFAULT_MAX_RECONNECT_ATTEMPTS; + } + + private sleep(): (ms: number) => Promise { + return this.config.sleep ?? defaultSleep; + } +} diff --git a/packages/connect-core/src/index.ts b/packages/connect-core/src/index.ts index 313a20d..bf13dd7 100644 --- a/packages/connect-core/src/index.ts +++ b/packages/connect-core/src/index.ts @@ -10,6 +10,12 @@ import { PactoError, PactoSessionError, } from './errors.js'; +import { + type EscrowEventHandler, + type EscrowEventName, + EscrowEventSubscriber, + type EscrowSubscribeOptions, +} from './escrow-events.js'; import { PUBLISHABLE_KEY_HEADER } from './http.js'; import { createApiClient, type PactoApiClient } from './resources.js'; @@ -21,6 +27,14 @@ export { PactoRateLimitError, PactoSessionError, } from './errors.js'; +export { + ESCROW_EVENT_NAMES, + type EscrowEvent, + type EscrowEventHandler, + type EscrowEventName, + type EscrowMilestone, + type EscrowSubscribeOptions, +} from './escrow-events.js'; export type { CreateEscrowParams, CreateQuoteParams, @@ -47,6 +61,8 @@ export interface PactoInitOptions { maxRetries?: number; /** Base delay in milliseconds for exponential backoff. */ baseDelayMs?: number; + /** Maximum reconnect attempts for escrow event streams. */ + maxReconnectAttempts?: number; } export type CreateCheckoutSessionParams = @@ -74,6 +90,15 @@ interface GatewaySessionResponse { mode: CheckoutMode; } +interface SessionRuntimeConfig { + gatewayUrl: string; + publishableKey: string; + origin?: string; + baseDelayMs?: number; + maxRetries?: number; + maxReconnectAttempts?: number; +} + const DEFAULT_GATEWAY_URL = 'https://connect.pacto.example'; function isCheckoutMode(value: string): value is CheckoutMode { @@ -86,6 +111,8 @@ export class PactoSession { readonly expiresAt: Date; readonly mode: CheckoutMode; + private subscriber?: EscrowEventSubscriber; + constructor( private readonly client: InternalPactoClient, data: PactoSessionData, @@ -104,9 +131,34 @@ export class PactoSession { const data = await this.client.refreshSession(this.clientSecret); return new PactoSession(this.client, data); } + + on(event: EscrowEventName, handler: EscrowEventHandler, options?: EscrowSubscribeOptions): void { + if (!this.subscriber) { + this.subscriber = new EscrowEventSubscriber({ + gatewayUrl: this.client.runtime.gatewayUrl, + publishableKey: this.client.runtime.publishableKey, + clientSecret: this.clientSecret, + origin: this.client.runtime.origin, + baseDelayMs: this.client.runtime.baseDelayMs, + maxReconnectAttempts: this.client.runtime.maxReconnectAttempts, + }); + } + + this.subscriber.on(event, handler, options); + } + + off(event: EscrowEventName, handler: EscrowEventHandler): void { + this.subscriber?.off(event, handler); + } + + closeEvents(): void { + this.subscriber?.close(); + this.subscriber = undefined; + } } interface InternalPactoClient extends PactoClient { + readonly runtime: SessionRuntimeConfig; refreshSession(clientSecret: string): Promise; } @@ -116,6 +168,16 @@ function createGatewayClient(options: PactoInitOptions): InternalPactoClient { const origin = options.origin; const maxRetries = options.maxRetries; const baseDelayMs = options.baseDelayMs; + const maxReconnectAttempts = options.maxReconnectAttempts; + + const runtime: SessionRuntimeConfig = { + gatewayUrl, + publishableKey, + origin, + baseDelayMs, + maxRetries, + maxReconnectAttempts, + }; async function requestSession( path: string, @@ -166,6 +228,7 @@ function createGatewayClient(options: PactoInitOptions): InternalPactoClient { return { publishableKey, gatewayUrl, + runtime, async createCheckoutSession(params: CreateCheckoutSessionParams): Promise { const data = await requestSession('/v1/session', params); return new PactoSession(this, data); diff --git a/packages/connect-core/src/sse.ts b/packages/connect-core/src/sse.ts new file mode 100644 index 0000000..bd4f7cb --- /dev/null +++ b/packages/connect-core/src/sse.ts @@ -0,0 +1,81 @@ +export interface SseMessage { + id?: string; + event?: string; + data: string; +} + +export function parseSseBlock(block: string): SseMessage | null { + const lines = block.split('\n'); + let id: string | undefined; + let event: string | undefined; + const dataLines: string[] = []; + + for (const line of lines) { + if (line.startsWith(':')) { + continue; + } + + if (line.startsWith('id:')) { + id = line.slice(3).trim(); + continue; + } + + if (line.startsWith('event:')) { + event = line.slice(6).trim(); + continue; + } + + if (line.startsWith('data:')) { + dataLines.push(line.slice(5).trimStart()); + } + } + + if (dataLines.length === 0 && !event) { + return null; + } + + return { + id, + event, + data: dataLines.join('\n'), + }; +} + +export async function readSseStream( + body: ReadableStream, + onMessage: (message: SseMessage) => void, +): Promise { + const reader = body.getReader(); + const decoder = new TextDecoder(); + let buffer = ''; + + while (true) { + 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) { + const trimmed = part.trim(); + if (!trimmed) { + continue; + } + + const message = parseSseBlock(trimmed); + if (message) { + onMessage(message); + } + } + } + + if (buffer.trim()) { + const message = parseSseBlock(buffer.trim()); + if (message) { + onMessage(message); + } + } +} diff --git a/packages/connect-react/src/index.ts b/packages/connect-react/src/index.ts index 9b1add5..0e4ff3c 100644 --- a/packages/connect-react/src/index.ts +++ b/packages/connect-react/src/index.ts @@ -10,8 +10,13 @@ export type { CreateEscrowParams, CreateQuoteParams, Escrow, + EscrowEvent, + EscrowEventHandler, + EscrowEventName, + EscrowMilestone, EscrowStatus, EscrowStatusResponse, + EscrowSubscribeOptions, Listing, PactoApiClient, PactoClient, @@ -20,6 +25,7 @@ export type { Quote, } from '@pacto-connect/core'; export { + ESCROW_EVENT_NAMES, PactoApiError, PactoAuthError, PactoError, From 4e0da98ae1312496db5f1e6ad84a45fe35befd3c Mon Sep 17 00:00:00 2001 From: diegocintron Date: Fri, 19 Jun 2026 11:46:53 -0600 Subject: [PATCH 2/2] style(core): fix biome format in escrow-events test --- packages/connect-core/src/escrow-events.test.ts | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/packages/connect-core/src/escrow-events.test.ts b/packages/connect-core/src/escrow-events.test.ts index 40fcac7..937c6a1 100644 --- a/packages/connect-core/src/escrow-events.test.ts +++ b/packages/connect-core/src/escrow-events.test.ts @@ -147,9 +147,9 @@ describe('escrow event subscription', () => { occurredAt: '2024-01-01T00:10:00.000Z', }); - const reconnectUrl = vi.mocked(fetch).mock.calls.find( - (call, index) => index > 0 && String(call[0]).includes('cursor=cursor-1'), - ); + const reconnectUrl = vi + .mocked(fetch) + .mock.calls.find((call, index) => index > 0 && String(call[0]).includes('cursor=cursor-1')); expect(reconnectUrl).toBeDefined(); subscriber.close();