diff --git a/src/client.ts b/src/client.ts index f109e0e..9b5bf99 100644 --- a/src/client.ts +++ b/src/client.ts @@ -38,6 +38,8 @@ import type { WatchOptions, EscrowRecord, AccountInfo, + VeriTixEvent, + StreamEventOptions, } from './types/index'; import { buildContractCall, simulateTransaction } from './utils/transaction'; import { DUMMY_PUBLIC_KEY, getMainnetConfig, getTestnetConfig } from './utils/network'; @@ -647,6 +649,183 @@ export class VeriTixClient extends EventEmitter { /** * Returns a proxy `SorobanRpc.Server` that throws a helpful error if * `connect()` has not been called yet. Modules hold a reference to this +/** + * Connects to Horizon Server-Sent Events (SSE) endpoint to stream contract events in real-time. + * Automatically reconnects with exponential backoff if the stream drops. Supports cancellation via AbortSignal. + * + * @param options - {@link StreamEventOptions} for stream configuration (signal, backoff settings, cursor) + * @returns AsyncIterableIterator that yields {@link VeriTixEvent} as they are received + * + * @example + * ```ts + * const controller = new AbortController(); + * for await (const event of client.streamEvents({ signal: controller.signal })) { + * console.log(`Received event: ${event.type} from ledger ${event.ledger}`); + * if (event.type === 'ticket_purchased') { + * console.log('New ticket sold!', event.data); + * } + * } + * ``` + */ + async *streamEvents(options?: StreamEventOptions): AsyncIterableIterator { + if (!this.connected || !this.server) { + throw new VeriTixError( + VeriTixErrorCode.ClientNotConnected, + 'VeriTixClient: call connect() before using streamEvents()' + ); + } + + // Get proper Horizon URL based on network + const TESTNET_HORIZON_URL = 'https://horizon-testnet.stellar.org'; + const MAINNET_HORIZON_URL = 'https://horizon.stellar.org'; + let baseHorizonUrl: string; + + if (this.config.network === 'testnet') { + baseHorizonUrl = TESTNET_HORIZON_URL; + } else { + baseHorizonUrl = MAINNET_HORIZON_URL; + } + + // If user provided a custom RPC URL that's not the default, try to derive Horizon URL from it + const isDefaultTestnetRpc = this.config.rpcUrl === 'https://soroban-testnet.stellar.org'; + const isDefaultMainnetRpc = this.config.rpcUrl === 'https://mainnet.stellar.validationcloud.io/v1/soroban/rpc'; + + if (!isDefaultTestnetRpc && !isDefaultMainnetRpc) { + // Try to extract Horizon URL from custom RPC URL + baseHorizonUrl = this.config.rpcUrl.replace(/\/soroban\/rpc$/, ''); + } + + if (!baseHorizonUrl.endsWith('/')) { + baseHorizonUrl += '/'; + } + + const opts = { + initialBackoffMs: options?.initialBackoffMs ?? 1000, + maxBackoffMs: options?.maxBackoffMs ?? 30000, + signal: options?.signal, + cursor: options?.cursor, + }; + + let currentBackoff = opts.initialBackoffMs; + let eventQueue: VeriTixEvent[] = []; + let queueResolver: (() => void) | null = null; + let eventSource: any = null; + let isAborted = false; + + // Dynamically import EventSource if in Node.js environment (browser has it globally) + let EventSourceImpl: typeof EventSource; + if (typeof EventSource === 'undefined') { + // Node.js environment - require eventsource package + try { + const { EventSource: NodeEventSource } = require('eventsource'); + EventSourceImpl = NodeEventSource; + } catch (err) { + throw new VeriTixError( + VeriTixErrorCode.InvalidConfig, + 'VeriTixClient: streamEvents() requires the "eventsource" package in Node.js environments. Please install it with npm install eventsource.' + ); + } + } else { + // Browser environment - use global EventSource + EventSourceImpl = EventSource; + } + + // Setup abort signal listener + if (opts.signal) { + opts.signal.addEventListener('abort', () => { + isAborted = true; + if (eventSource) { + eventSource.close(); + eventSource = null; + } + if (queueResolver) { + queueResolver(); + queueResolver = null; + } + }); + } + + // Function to create and connect EventSource + const connectStream = () => { + if (isAborted) return; + + // Horizon SSE endpoint for contract events: /contract//events + const streamUrl = new URL(`contract/${this.config.contractId}/events`, baseHorizonUrl); + if (opts.cursor) { + streamUrl.searchParams.set('cursor', opts.cursor); + } + + try { + eventSource = new EventSourceImpl(streamUrl.toString()); + + eventSource.onopen = () => { + // Reset backoff on successful connection + currentBackoff = opts.initialBackoffMs; + }; + + eventSource.onmessage = (event: any) => { + try { + const rawEvent = JSON.parse(event.data); + // Parse raw Horizon SSE event into VeriTixEvent (Horizon event format: https://developers.stellar.org/docs/data/horizon/api-reference/stream/contract-events) + const veriTixEvent: VeriTixEvent = { + type: rawEvent.topic?.[0] || 'unknown', + ledger: parseInt(rawEvent.ledger, 10), + timestamp: parseInt(rawEvent.created_at ? new Date(rawEvent.created_at).getTime() / 1000 : rawEvent.timestamp, 10), + topics: rawEvent.topic || [], + data: rawEvent.value, + }; + eventQueue.push(veriTixEvent); + if (queueResolver) { + queueResolver(); + queueResolver = null; + } + } catch (parseErr) { + // Skip invalid events + } + }; + + eventSource.onerror = () => { + if (eventSource) { + eventSource.close(); + eventSource = null; + } + // Schedule reconnection with exponential backoff if not aborted + if (!isAborted) { + setTimeout(() => { + currentBackoff = Math.min(currentBackoff * 2, opts.maxBackoffMs); + connectStream(); + }, currentBackoff); + } + }; + } catch (err) { + // Handle connection errors, schedule reconnection + if (!isAborted) { + setTimeout(() => { + currentBackoff = Math.min(currentBackoff * 2, opts.maxBackoffMs); + connectStream(); + }, currentBackoff); + } + } + }; + + // Start initial connection + connectStream(); + + // Yield events as they come in + while (!isAborted) { + if (eventQueue.length === 0) { + // Wait for new events + await new Promise((resolve) => { + queueResolver = resolve; + }); + } else { + const nextEvent = eventQueue.shift()!; + yield nextEvent; + } + } + } + + /** * proxy so they surface a clear message instead of a confusing crash. * * @internal diff --git a/src/index.ts b/src/index.ts index fc424d0..e9da6a6 100644 --- a/src/index.ts +++ b/src/index.ts @@ -45,6 +45,8 @@ export type { RecurringExecutionEntry, TransactionResult, WatchOptions, + VeriTixEvent, + StreamEventOptions, } from './types/index'; export { DisputeStatus } from './types/index'; diff --git a/src/types/index.ts b/src/types/index.ts index 8355fc8..c2bdd98 100644 --- a/src/types/index.ts +++ b/src/types/index.ts @@ -268,6 +268,70 @@ export interface WatchOptions { timeoutMs?: number; } +/** + * Represents a contract event emitted by the VeriTix contract. + */ +export interface VeriTixEvent { + /** Type of the event (contract-specific event name) */ + type: string; + /** Ledger sequence number where the event was emitted */ + ledger: number; + /** Unix timestamp (in seconds) when the ledger was closed */ + timestamp: number; + /** Array of topic strings associated with the event */ + topics: string[]; + /** Decoded event data from the contract */ + data: unknown; +} + +/** + * Options for {@link VeriTixClient.streamEvents}. + */ +export interface StreamEventOptions { + /** AbortSignal to cancel the event stream */ + signal?: AbortSignal; + /** Initial backoff delay in milliseconds for reconnections (default 1000) */ + initialBackoffMs?: number; + /** Maximum backoff delay in milliseconds (default 30000) */ + maxBackoffMs?: number; + /** Whether to include events from ledgers before the current one (default false) */ + history?: boolean; + /** Cursor to start streaming from a specific ledger sequence */ + cursor?: string; +} + +/** + * Represents a decoded Soroban contract event emitted by the VeriTix contract. + */ +export interface VeriTixEvent { + /** Event type/name as emitted by the contract */ + type: string; + /** Ledger sequence number when the event was emitted */ + ledger: number; + /** Unix timestamp (seconds since epoch) when the event was emitted */ + timestamp: number; + /** Array of topic values associated with the event */ + topics: string[]; + /** Decoded event data payload */ + data: unknown; +} + +/** + * Options for {@link VeriTixClient.streamEvents}. + */ +export interface StreamEventOptions { + /** AbortSignal to cancel the event stream */ + signal?: AbortSignal; + /** Initial backoff delay in milliseconds before first reconnection attempt (default 1000) */ + initialBackoffMs?: number; + /** Maximum backoff delay in milliseconds (default 30000) */ + maxBackoffMs?: number; + /** Whether to use exponential backoff for reconnections (default true) */ + useExponentialBackoff?: boolean; + /** Horizon URL to use for SSE events (defaults to network-specific Horizon) */ + horizonUrl?: string; +} + /** * Minimal representation of a submitted Stellar transaction result. */