From c4ffae0cb874277ed75067981c59c1bde76cd39e Mon Sep 17 00:00:00 2001 From: wavyboy-build Date: Sat, 17 Oct 2026 12:42:47 +0100 Subject: [PATCH] feat(ingestion): add schema versioning, decoder registry, raw fallbacks, and RPC startup checks --- .../schema/contract-version-registry.test.ts | 401 ++++++++++++++++ .../src/schema/contract-version-registry.ts | 256 ++++++++++ .../services/event-decoder-registry.test.ts | 390 +++++++++++++++ .../src/services/event-decoder-registry.ts | 319 +++++++++++++ .../services/rpc-capability-checker.test.ts | 373 +++++++++++++++ .../src/services/rpc-capability-checker.ts | 444 ++++++++++++++++++ 6 files changed, 2183 insertions(+) create mode 100644 listener/src/schema/contract-version-registry.test.ts create mode 100644 listener/src/schema/contract-version-registry.ts create mode 100644 listener/src/services/event-decoder-registry.test.ts create mode 100644 listener/src/services/event-decoder-registry.ts create mode 100644 listener/src/services/rpc-capability-checker.test.ts create mode 100644 listener/src/services/rpc-capability-checker.ts diff --git a/listener/src/schema/contract-version-registry.test.ts b/listener/src/schema/contract-version-registry.test.ts new file mode 100644 index 00000000..940fe1d4 --- /dev/null +++ b/listener/src/schema/contract-version-registry.test.ts @@ -0,0 +1,401 @@ +/** + * Tests for ContractVersionRegistry (Task 1) + * + * Coverage: + * - Schema registration and overwrite + * - resolveSchema: found / not_found results + * - diagnose: compatible / incompatible / unsupported + * - Structured diagnostics carry correct fields + * - Singleton helpers (get / set / reset) + * - Input validation (bad version, empty events) + */ + +import { + ContractVersionRegistry, + getContractVersionRegistry, + resetContractVersionRegistry, + setContractVersionRegistry, + type VersionedSchemaSnapshot, +} from './contract-version-registry'; +import type { ContractEventSchema, OffConsumerEventSchema } from './compatibility-check'; + +// --------------------------------------------------------------------------- +// Fixtures +// --------------------------------------------------------------------------- + +/** Minimal valid consumer event definition used across tests. */ +const EVENT_NOTIFICATION_SCHEDULED: OffConsumerEventSchema = { + eventName: 'NotificationScheduled', + expectedTopics: [ + { name: 'creator', type: 'Address' }, + { name: 'category', type: 'NotificationCategory' }, + { name: 'priority', type: 'NotificationPriority' }, + ], + expectedFields: [ + { name: 'notification_id', type: 'BytesN<32>' }, + { name: 'payload_version', type: 'u32' }, + ], +}; + +/** A second minimal event definition. */ +const EVENT_CONTRACT_PAUSED: OffConsumerEventSchema = { + eventName: 'ContractPaused', + expectedTopics: [ + { name: 'admin', type: 'Address' }, + { name: 'category', type: 'NotificationCategory' }, + { name: 'priority', type: 'NotificationPriority' }, + ], + expectedFields: [], +}; + +/** Contract-side representation of NotificationScheduled (matching the consumer). */ +function makeContractEvent( + name: string, + topics: Array<[string, string]>, + dataFields: Array<[string, string]>, +): ContractEventSchema { + return { + name, + topics: topics.map(([n, t]) => ({ name: n, type: t })), + dataFields: dataFields.map(([n, t]) => ({ name: n, type: t })), + }; +} + +const CONTRACT_EVENT_NOTIFICATION_SCHEDULED = makeContractEvent( + 'NotificationScheduled', + [ + ['creator', 'Address'], + ['category', 'NotificationCategory'], + ['priority', 'NotificationPriority'], + ], + [ + ['notification_id', 'BytesN<32>'], + ['payload_version', 'u32'], + ], +); + +const CONTRACT_EVENT_CONTRACT_PAUSED = makeContractEvent( + 'ContractPaused', + [ + ['admin', 'Address'], + ['category', 'NotificationCategory'], + ['priority', 'NotificationPriority'], + ], + [], +); + +// A broken contract event that renamed `payload_version` → `schema_version` +const CONTRACT_EVENT_NOTIFICATION_SCHEDULED_BROKEN = makeContractEvent( + 'NotificationScheduled', + [ + ['creator', 'Address'], + ['category', 'NotificationCategory'], + ['priority', 'NotificationPriority'], + ], + [ + ['notification_id', 'BytesN<32>'], + ['schema_version', 'u32'], // renamed — breaking change + ], +); + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +function makeContractEvents( + ...events: ContractEventSchema[] +): Map { + const map = new Map(); + for (const e of events) map.set(e.name, e); + return map; +} + +// --------------------------------------------------------------------------- +// Registration tests +// --------------------------------------------------------------------------- + +describe('ContractVersionRegistry — registration', () => { + let registry: ContractVersionRegistry; + + beforeEach(() => { + registry = new ContractVersionRegistry(); + }); + + it('registers a snapshot and confirms with has()', () => { + registry.register(1, [EVENT_NOTIFICATION_SCHEDULED]); + expect(registry.has(1)).toBe(true); + }); + + it('lists registered versions in ascending order', () => { + registry.register(3, [EVENT_CONTRACT_PAUSED]); + registry.register(1, [EVENT_NOTIFICATION_SCHEDULED]); + registry.register(2, [EVENT_NOTIFICATION_SCHEDULED, EVENT_CONTRACT_PAUSED]); + expect(registry.registeredVersions()).toEqual([1, 2, 3]); + }); + + it('overwrites an existing snapshot', () => { + registry.register(1, [EVENT_NOTIFICATION_SCHEDULED]); + registry.register(1, [EVENT_CONTRACT_PAUSED]); // overwrite + const result = registry.resolveSchema(1); + expect(result.status).toBe('found'); + if (result.status === 'found') { + expect(result.snapshot.events).toHaveLength(1); + expect(result.snapshot.events[0].eventName).toBe('ContractPaused'); + } + }); + + it('deregisters a snapshot', () => { + registry.register(1, [EVENT_NOTIFICATION_SCHEDULED]); + expect(registry.deregister(1)).toBe(true); + expect(registry.has(1)).toBe(false); + }); + + it('returns false when deregistering an unregistered version', () => { + expect(registry.deregister(99)).toBe(false); + }); + + it('throws for non-positive contract version', () => { + expect(() => registry.register(0, [EVENT_NOTIFICATION_SCHEDULED])).toThrow( + /positive integer/, + ); + expect(() => registry.register(-1, [EVENT_NOTIFICATION_SCHEDULED])).toThrow( + /positive integer/, + ); + }); + + it('throws for non-integer contract version', () => { + expect(() => registry.register(1.5, [EVENT_NOTIFICATION_SCHEDULED])).toThrow( + /positive integer/, + ); + }); + + it('throws when events array is empty', () => { + expect(() => registry.register(1, [])).toThrow(/non-empty array/); + }); + + it('clears all snapshots', () => { + registry.register(1, [EVENT_NOTIFICATION_SCHEDULED]); + registry.register(2, [EVENT_CONTRACT_PAUSED]); + registry.clear(); + expect(registry.registeredVersions()).toHaveLength(0); + }); +}); + +// --------------------------------------------------------------------------- +// resolveSchema tests +// --------------------------------------------------------------------------- + +describe('ContractVersionRegistry — resolveSchema', () => { + let registry: ContractVersionRegistry; + + beforeEach(() => { + registry = new ContractVersionRegistry(); + registry.register(1, [EVENT_NOTIFICATION_SCHEDULED]); + }); + + it('returns found status for a registered version', () => { + const result = registry.resolveSchema(1); + expect(result.status).toBe('found'); + }); + + it('returns the correct snapshot when found', () => { + const result = registry.resolveSchema(1); + if (result.status !== 'found') throw new Error('Expected found'); + expect(result.snapshot.contractVersion).toBe(1); + expect(result.snapshot.events).toHaveLength(1); + expect(result.snapshot.events[0].eventName).toBe('NotificationScheduled'); + }); + + it('returns not_found for an unregistered version', () => { + const result = registry.resolveSchema(42); + expect(result.status).toBe('not_found'); + }); + + it('not_found result includes the requested version and registered versions', () => { + const result = registry.resolveSchema(42); + if (result.status !== 'not_found') throw new Error('Expected not_found'); + expect(result.requestedVersion).toBe(42); + expect(result.registeredVersions).toContain(1); + expect(result.message).toMatch(/42/); + expect(result.message).toMatch(/register/i); + }); + + it('not_found message is informative when no versions are registered', () => { + const empty = new ContractVersionRegistry(); + const result = empty.resolveSchema(5); + if (result.status !== 'not_found') throw new Error('Expected not_found'); + expect(result.registeredVersions).toHaveLength(0); + expect(result.message).toMatch(/none/i); + }); +}); + +// --------------------------------------------------------------------------- +// diagnose — compatible +// --------------------------------------------------------------------------- + +describe('ContractVersionRegistry — diagnose (compatible)', () => { + let registry: ContractVersionRegistry; + + beforeEach(() => { + registry = new ContractVersionRegistry(); + registry.register(1, [EVENT_NOTIFICATION_SCHEDULED, EVENT_CONTRACT_PAUSED]); + }); + + it('returns compatible when contract events match the snapshot', () => { + const contractEvents = makeContractEvents( + CONTRACT_EVENT_NOTIFICATION_SCHEDULED, + CONTRACT_EVENT_CONTRACT_PAUSED, + ); + const diag = registry.diagnose(contractEvents, 1); + expect(diag.status).toBe('compatible'); + }); + + it('compatible diagnostic has no breaking changes', () => { + const contractEvents = makeContractEvents( + CONTRACT_EVENT_NOTIFICATION_SCHEDULED, + CONTRACT_EVENT_CONTRACT_PAUSED, + ); + const diag = registry.diagnose(contractEvents, 1); + expect(diag.breakingChanges).toHaveLength(0); + }); + + it('compatible diagnostic includes registeredVersions', () => { + const contractEvents = makeContractEvents(CONTRACT_EVENT_NOTIFICATION_SCHEDULED); + registry.register(2, [EVENT_NOTIFICATION_SCHEDULED]); + const diag = registry.diagnose(contractEvents, 1); + expect(diag.registeredVersions).toContain(1); + expect(diag.registeredVersions).toContain(2); + }); + + it('safe additions are surfaced when the contract has extra events', () => { + // The snapshot only knows about NotificationScheduled + // The contract also has ContractPaused — that is a safe addition + const contractEvents = makeContractEvents( + CONTRACT_EVENT_NOTIFICATION_SCHEDULED, + CONTRACT_EVENT_CONTRACT_PAUSED, + ); + registry.clear(); + registry.register(1, [EVENT_NOTIFICATION_SCHEDULED]); + const diag = registry.diagnose(contractEvents, 1); + expect(diag.status).toBe('compatible'); + expect(diag.safeAdditions.length).toBeGreaterThan(0); + expect(diag.safeAdditions.some((s) => s.toLowerCase().includes('contractpaused'))).toBe(true); + }); +}); + +// --------------------------------------------------------------------------- +// diagnose — incompatible +// --------------------------------------------------------------------------- + +describe('ContractVersionRegistry — diagnose (incompatible)', () => { + let registry: ContractVersionRegistry; + + beforeEach(() => { + registry = new ContractVersionRegistry(); + registry.register(1, [EVENT_NOTIFICATION_SCHEDULED]); + }); + + it('returns incompatible when a data field is renamed', () => { + const contractEvents = makeContractEvents(CONTRACT_EVENT_NOTIFICATION_SCHEDULED_BROKEN); + const diag = registry.diagnose(contractEvents, 1); + expect(diag.status).toBe('incompatible'); + }); + + it('incompatible diagnostic lists breaking changes', () => { + const contractEvents = makeContractEvents(CONTRACT_EVENT_NOTIFICATION_SCHEDULED_BROKEN); + const diag = registry.diagnose(contractEvents, 1); + expect(diag.breakingChanges.length).toBeGreaterThan(0); + }); + + it('incompatible message mentions the version and breaking change count', () => { + const contractEvents = makeContractEvents(CONTRACT_EVENT_NOTIFICATION_SCHEDULED_BROKEN); + const diag = registry.diagnose(contractEvents, 1); + expect(diag.message).toMatch(/1/); + expect(diag.message).toMatch(/breaking/i); + }); + + it('returns incompatible when a consumer-expected event is missing from contract', () => { + // Snapshot requires NotificationScheduled but the contract map is empty + const diag = registry.diagnose(new Map(), 1); + expect(diag.status).toBe('incompatible'); + expect(diag.breakingChanges.some((c) => c.toLowerCase().includes('missing'))).toBe(true); + }); +}); + +// --------------------------------------------------------------------------- +// diagnose — unsupported +// --------------------------------------------------------------------------- + +describe('ContractVersionRegistry — diagnose (unsupported)', () => { + let registry: ContractVersionRegistry; + + beforeEach(() => { + registry = new ContractVersionRegistry(); + registry.register(1, [EVENT_NOTIFICATION_SCHEDULED]); + }); + + it('returns unsupported for an unregistered version', () => { + const contractEvents = makeContractEvents(CONTRACT_EVENT_NOTIFICATION_SCHEDULED); + const diag = registry.diagnose(contractEvents, 99); + expect(diag.status).toBe('unsupported'); + }); + + it('unsupported diagnostic message is actionable', () => { + const diag = registry.diagnose(new Map(), 99); + expect(diag.message).toMatch(/99/); + expect(diag.message).toMatch(/register/i); + expect(diag.message).toMatch(/supported/i); + }); + + it('unsupported diagnostic has no breaking changes', () => { + const diag = registry.diagnose(new Map(), 99); + expect(diag.breakingChanges).toHaveLength(0); + expect(diag.safeAdditions).toHaveLength(0); + }); + + it('unsupported diagnostic surfaces registered versions', () => { + const diag = registry.diagnose(new Map(), 99); + expect(diag.registeredVersions).toContain(1); + }); + + it('unsupported when registry is completely empty', () => { + const empty = new ContractVersionRegistry(); + const diag = empty.diagnose(new Map(), 1); + expect(diag.status).toBe('unsupported'); + expect(diag.registeredVersions).toHaveLength(0); + }); +}); + +// --------------------------------------------------------------------------- +// Singleton helpers +// --------------------------------------------------------------------------- + +describe('ContractVersionRegistry — singleton helpers', () => { + afterEach(() => { + resetContractVersionRegistry(); + }); + + it('getContractVersionRegistry returns a ContractVersionRegistry', () => { + const reg = getContractVersionRegistry(); + expect(reg).toBeInstanceOf(ContractVersionRegistry); + }); + + it('getContractVersionRegistry returns the same instance on repeated calls', () => { + const a = getContractVersionRegistry(); + const b = getContractVersionRegistry(); + expect(a).toBe(b); + }); + + it('setContractVersionRegistry replaces the singleton', () => { + const custom = new ContractVersionRegistry(); + setContractVersionRegistry(custom); + expect(getContractVersionRegistry()).toBe(custom); + }); + + it('resetContractVersionRegistry forces a fresh instance on next get', () => { + const first = getContractVersionRegistry(); + resetContractVersionRegistry(); + const second = getContractVersionRegistry(); + expect(second).not.toBe(first); + }); +}); diff --git a/listener/src/schema/contract-version-registry.ts b/listener/src/schema/contract-version-registry.ts new file mode 100644 index 00000000..8e268a5f --- /dev/null +++ b/listener/src/schema/contract-version-registry.ts @@ -0,0 +1,256 @@ +/** + * Contract Version Registry (Task 1) + * + * Associates supported event schemas with specific contract versions so that + * maintainers can detect incompatible event changes at startup or during CI. + * + * Concepts + * -------- + * - A *contract version* is the integer value returned by the on-chain + * `version()` entry point (currently `1`). + * - A *schema snapshot* is the set of event definitions that the off-chain + * consumer expects for that contract version. It re-uses the existing + * `OffConsumerEventSchema` shape from `compatibility-check.ts`. + * - A *version diagnostic* is a structured object describing the result of a + * compatibility check: either `compatible`, `unsupported` (the contract + * version has no snapshot registered), or `incompatible` (a registered + * snapshot was found but field-level mismatches exist). + * + * Usage + * ----- + * ```ts + * const registry = new ContractVersionRegistry(); + * registry.register(1, snapshot_v1); + * + * const result = registry.resolveSchema(1); + * // → { status: 'found', snapshot: snapshot_v1 } + * + * const diag = registry.diagnose(contractEvents, 1); + * // → { status: 'compatible', contractVersion: 1, breakingChanges: [], ... } + * ``` + */ + +import type { ContractEventSchema, OffConsumerEventSchema } from './compatibility-check'; +import { checkSchemaCompatibility } from './compatibility-check'; + +// --------------------------------------------------------------------------- +// Public types +// --------------------------------------------------------------------------- + +/** A complete schema snapshot keyed by event name for one contract version. */ +export interface VersionedSchemaSnapshot { + /** The contract version this snapshot covers (matches on-chain `version()`). */ + contractVersion: number; + /** Full event definitions expected for this version. */ + events: OffConsumerEventSchema[]; +} + +/** Result of resolving a schema for a given contract version. */ +export type SchemaResolutionResult = + | { status: 'found'; snapshot: VersionedSchemaSnapshot } + | { + status: 'not_found'; + requestedVersion: number; + registeredVersions: number[]; + message: string; + }; + +/** Possible outcomes of a version-aware compatibility check. */ +export type VersionDiagnosticStatus = 'compatible' | 'incompatible' | 'unsupported'; + +export interface VersionDiagnostic { + /** Overall outcome. */ + status: VersionDiagnosticStatus; + /** The contract version that was evaluated. */ + contractVersion: number; + /** Human-readable summary of the outcome. */ + message: string; + /** Field-level breaking changes (empty when `compatible` or `unsupported`). */ + breakingChanges: string[]; + /** Non-breaking additions (empty when `unsupported`). */ + safeAdditions: string[]; + /** All contract versions that have a registered snapshot. */ + registeredVersions: number[]; +} + +// --------------------------------------------------------------------------- +// Registry implementation +// --------------------------------------------------------------------------- + +/** + * Maintains a mapping of `contractVersion → VersionedSchemaSnapshot` and + * exposes version-aware schema resolution and diagnostic generation. + */ +export class ContractVersionRegistry { + private readonly snapshots = new Map(); + + // ------------------------------------------------------------------------- + // Registration + // ------------------------------------------------------------------------- + + /** + * Register a schema snapshot for a specific contract version. + * + * Calling this with an already-registered version overwrites the previous + * snapshot — useful in tests and when updating schemas in-place. + * + * @throws {Error} if `contractVersion` is not a positive integer. + */ + register(contractVersion: number, events: OffConsumerEventSchema[]): this { + if (!Number.isInteger(contractVersion) || contractVersion < 1) { + throw new Error( + `ContractVersionRegistry: contractVersion must be a positive integer, got ${contractVersion}`, + ); + } + if (!Array.isArray(events) || events.length === 0) { + throw new Error( + `ContractVersionRegistry: events must be a non-empty array for version ${contractVersion}`, + ); + } + this.snapshots.set(contractVersion, { contractVersion, events }); + return this; + } + + /** + * Remove the snapshot for a specific contract version. + * Returns `true` if a snapshot was found and removed, `false` otherwise. + */ + deregister(contractVersion: number): boolean { + return this.snapshots.delete(contractVersion); + } + + // ------------------------------------------------------------------------- + // Query + // ------------------------------------------------------------------------- + + /** + * Returns all registered contract versions in ascending order. + */ + registeredVersions(): number[] { + return [...this.snapshots.keys()].sort((a, b) => a - b); + } + + /** + * Returns `true` when a snapshot exists for `contractVersion`. + */ + has(contractVersion: number): boolean { + return this.snapshots.has(contractVersion); + } + + /** + * Resolve the schema snapshot for a given contract version. + * + * Returns a discriminated union — callers must handle both `found` and + * `not_found` branches. + */ + resolveSchema(contractVersion: number): SchemaResolutionResult { + const snapshot = this.snapshots.get(contractVersion); + if (snapshot) { + return { status: 'found', snapshot }; + } + const registered = this.registeredVersions(); + return { + status: 'not_found', + requestedVersion: contractVersion, + registeredVersions: registered, + message: + `No schema snapshot registered for contract version ${contractVersion}. ` + + `Registered versions: [${registered.join(', ') || 'none'}]. ` + + `Register a snapshot with ContractVersionRegistry.register(${contractVersion}, events).`, + }; + } + + // ------------------------------------------------------------------------- + // Diagnostics + // ------------------------------------------------------------------------- + + /** + * Run a full compatibility check between live contract events and the + * registered snapshot for `contractVersion`. + * + * Returns a `VersionDiagnostic` with status: + * - `unsupported` — no snapshot is registered for `contractVersion`. + * - `compatible` — all consumer-expected fields match the contract events. + * - `incompatible`— one or more breaking changes were detected. + */ + diagnose( + contractEvents: Map, + contractVersion: number, + ): VersionDiagnostic { + const registered = this.registeredVersions(); + const resolution = this.resolveSchema(contractVersion); + + if (resolution.status === 'not_found') { + return { + status: 'unsupported', + contractVersion, + message: + `Contract version ${contractVersion} is not supported. ` + + `Supported versions: [${registered.join(', ') || 'none'}]. ` + + `Add a snapshot via ContractVersionRegistry.register(${contractVersion}, events) ` + + `to enable compatibility checking for this version.`, + breakingChanges: [], + safeAdditions: [], + registeredVersions: registered, + }; + } + + const { snapshot } = resolution; + const consumerSchema = { events: snapshot.events }; + const result = checkSchemaCompatibility(contractEvents, consumerSchema); + + if (result.compatible) { + return { + status: 'compatible', + contractVersion, + message: `Contract version ${contractVersion} is fully compatible with the registered schema.`, + breakingChanges: [], + safeAdditions: result.safeAdditions, + registeredVersions: registered, + }; + } + + return { + status: 'incompatible', + contractVersion, + message: + `Contract version ${contractVersion} is incompatible with the registered schema. ` + + `Found ${result.breakingChanges.length} breaking change(s). ` + + `Review the breakingChanges field for details.`, + breakingChanges: result.breakingChanges, + safeAdditions: result.safeAdditions, + registeredVersions: registered, + }; + } + + /** + * Clear all registered snapshots. Primarily useful in test teardowns. + */ + clear(): void { + this.snapshots.clear(); + } +} + +// --------------------------------------------------------------------------- +// Module-level singleton +// --------------------------------------------------------------------------- + +let _registry: ContractVersionRegistry | null = null; + +/** Returns the module-level singleton registry, creating it on first call. */ +export function getContractVersionRegistry(): ContractVersionRegistry { + if (!_registry) { + _registry = new ContractVersionRegistry(); + } + return _registry; +} + +/** Replace the singleton. Use in tests or during bootstrap. */ +export function setContractVersionRegistry(registry: ContractVersionRegistry): void { + _registry = registry; +} + +/** Reset the singleton to `null`. Use in test teardowns. */ +export function resetContractVersionRegistry(): void { + _registry = null; +} diff --git a/listener/src/services/event-decoder-registry.test.ts b/listener/src/services/event-decoder-registry.test.ts new file mode 100644 index 00000000..b5791c5b --- /dev/null +++ b/listener/src/services/event-decoder-registry.test.ts @@ -0,0 +1,390 @@ +/** + * Tests for EventDecoderRegistry (Tasks 2 & 3) + * + * Coverage: + * - Decoder registration, lookup, overwrite, removal + * - Successful decoding produces a DecodedEvent + * - Unknown event name → RawUndecodedEvent with reason 'no_decoder' + * - Decoder returning null → RawUndecodedEvent with reason 'decode_failed' + * - Decoder throwing → RawUndecodedEvent with reason 'decode_failed' + * - Raw events always preserve source identifiers (Task 2) + * - API distinguishes decoded vs raw via `kind` discriminant (Task 2) + * - Singleton helpers (get / set / reset) + */ + +import { + EventDecoderRegistry, + getEventDecoderRegistry, + resetEventDecoderRegistry, + setEventDecoderRegistry, + type DecoderInput, + type DecodedEvent, + type EventDecoder, + type RawUndecodedEvent, +} from './event-decoder-registry'; +import type { NormalizedEvent } from './event-normalization'; + +// --------------------------------------------------------------------------- +// Fixtures +// --------------------------------------------------------------------------- + +/** Minimal NormalizedEvent factory. */ +function makeNormalized(overrides: Partial = {}): NormalizedEvent { + return { + eventId: 'event-001', + contractAddress: 'CAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAWHF', + eventName: 'notification_scheduled', + eventType: 'contract', + ledger: 1000, + topic: ['notification_scheduled', 'GCREATOR', 'Notification', 'Medium'], + value: '{"notification_id":"abc","payload_version":1}', + txHash: 'tx-hash-001', + payloadVersion: 1, + ...overrides, + }; +} + +/** Typed payload for a decoded notification_scheduled event. */ +interface NotificationScheduledPayload { + notificationId: string; + payloadVersion: number; +} + +/** + * A simple well-behaved decoder for `notification_scheduled`. + */ +const notificationScheduledDecoder: EventDecoder = { + eventName: 'notification_scheduled', + decode(input: DecoderInput): NotificationScheduledPayload | null { + if (!input.payloadVersion) return null; + return { + notificationId: input.topic[1] ?? 'unknown', + payloadVersion: input.payloadVersion, + }; + }, +}; + +/** A decoder that always returns null (simulating a parse failure). */ +const alwaysNullDecoder: EventDecoder = { + eventName: 'bad_event', + decode(): null { + return null; + }, +}; + +/** A decoder that always throws. */ +const alwaysThrowsDecoder: EventDecoder = { + eventName: 'throwing_event', + decode(): never { + throw new Error('Decoder internal error'); + }, +}; + +// --------------------------------------------------------------------------- +// Registration tests +// --------------------------------------------------------------------------- + +describe('EventDecoderRegistry — registration', () => { + let registry: EventDecoderRegistry; + + beforeEach(() => { + registry = new EventDecoderRegistry(); + }); + + it('registers a decoder and confirms with has()', () => { + registry.register(notificationScheduledDecoder); + expect(registry.has('notification_scheduled')).toBe(true); + }); + + it('listRegistered includes the registered event name', () => { + registry.register(notificationScheduledDecoder); + expect(registry.listRegistered()).toContain('notification_scheduled'); + }); + + it('can register multiple decoders', () => { + registry.register(notificationScheduledDecoder); + registry.register(alwaysNullDecoder); + expect(registry.listRegistered()).toHaveLength(2); + }); + + it('overwrites an existing decoder without throwing', () => { + registry.register(notificationScheduledDecoder); + const replacement: EventDecoder<{ replaced: true }> = { + eventName: 'notification_scheduled', + decode: () => ({ replaced: true }), + }; + expect(() => registry.register(replacement)).not.toThrow(); + expect(registry.has('notification_scheduled')).toBe(true); + }); + + it('unregister removes a registered decoder', () => { + registry.register(notificationScheduledDecoder); + expect(registry.unregister('notification_scheduled')).toBe(true); + expect(registry.has('notification_scheduled')).toBe(false); + }); + + it('unregister returns false for an unknown event name', () => { + expect(registry.unregister('does_not_exist')).toBe(false); + }); + + it('throws when decoder eventName is empty', () => { + const bad: EventDecoder = { eventName: '', decode: () => null }; + expect(() => registry.register(bad)).toThrow(/non-empty string/); + }); + + it('clear removes all decoders', () => { + registry.register(notificationScheduledDecoder); + registry.register(alwaysNullDecoder); + registry.clear(); + expect(registry.listRegistered()).toHaveLength(0); + }); +}); + +// --------------------------------------------------------------------------- +// Decoding — success path +// --------------------------------------------------------------------------- + +describe('EventDecoderRegistry — decode (success)', () => { + let registry: EventDecoderRegistry; + + beforeEach(() => { + registry = new EventDecoderRegistry(); + registry.register(notificationScheduledDecoder); + }); + + it('returns a DecodedEvent with kind "decoded"', () => { + const event = makeNormalized(); + const result = registry.decode(event); + expect(result.kind).toBe('decoded'); + }); + + it('decoded event carries source identifiers', () => { + const event = makeNormalized({ eventId: 'my-event', contractAddress: 'CA123' }); + const result = registry.decode(event); + expect(result.eventId).toBe('my-event'); + expect(result.contractAddress).toBe('CA123'); + }); + + it('decoded event carries ledger and txHash', () => { + const event = makeNormalized({ ledger: 9999, txHash: 'tx-xyz' }); + const result = registry.decode(event); + expect(result.ledger).toBe(9999); + expect(result.txHash).toBe('tx-xyz'); + }); + + it('decoded event payload has typed fields', () => { + const event = makeNormalized({ payloadVersion: 2 }); + const result = registry.decode(event); + if (result.kind !== 'decoded') throw new Error('Expected decoded'); + expect(result.payload.payloadVersion).toBe(2); + }); +}); + +// --------------------------------------------------------------------------- +// Decoding — unknown event (Task 2 + 3) +// --------------------------------------------------------------------------- + +describe('EventDecoderRegistry — decode (unknown event / fallback)', () => { + let registry: EventDecoderRegistry; + + beforeEach(() => { + registry = new EventDecoderRegistry(); + // Intentionally do NOT register a decoder for 'unknown_event' + }); + + it('returns RawUndecodedEvent with kind "raw" for unregistered event', () => { + const event = makeNormalized({ eventName: 'unknown_event' }); + const result = registry.decode(event); + expect(result.kind).toBe('raw'); + }); + + it('raw result has reason "no_decoder"', () => { + const event = makeNormalized({ eventName: 'unknown_event' }); + const result = registry.decode(event); + if (result.kind !== 'raw') throw new Error('Expected raw'); + expect(result.reason).toBe('no_decoder'); + }); + + it('raw result preserves the source eventId (Task 2 — source identifiers)', () => { + const event = makeNormalized({ eventId: 'preserve-me', eventName: 'unknown_event' }); + const result = registry.decode(event); + expect(result.eventId).toBe('preserve-me'); + }); + + it('raw result preserves contractAddress (Task 2)', () => { + const event = makeNormalized({ + contractAddress: 'CA-UNKNOWN-CONTRACT', + eventName: 'unknown_event', + }); + const result = registry.decode(event); + expect(result.contractAddress).toBe('CA-UNKNOWN-CONTRACT'); + }); + + it('raw result preserves the raw topic array (Task 2)', () => { + const event = makeNormalized({ + eventName: 'unknown_event', + topic: ['unknown_event', 'raw_payload_data'], + }); + const result = registry.decode(event); + if (result.kind !== 'raw') throw new Error('Expected raw'); + expect(result.rawTopic).toEqual(['unknown_event', 'raw_payload_data']); + }); + + it('raw result preserves the raw value string (Task 2)', () => { + const event = makeNormalized({ + eventName: 'unknown_event', + value: '{"some":"raw","data":true}', + }); + const result = registry.decode(event); + if (result.kind !== 'raw') throw new Error('Expected raw'); + expect(result.rawValue).toBe('{"some":"raw","data":true}'); + }); + + it('handles null eventName gracefully (completely unrecognized contract)', () => { + const event = makeNormalized({ eventName: null as unknown as string }); + const result = registry.decode(event); + expect(result.kind).toBe('raw'); + if (result.kind !== 'raw') throw new Error('Expected raw'); + expect(result.reason).toBe('no_decoder'); + }); +}); + +// --------------------------------------------------------------------------- +// Decoding — decoder returns null (parse failure) +// --------------------------------------------------------------------------- + +describe('EventDecoderRegistry — decode (decoder returns null)', () => { + let registry: EventDecoderRegistry; + + beforeEach(() => { + registry = new EventDecoderRegistry(); + registry.register(alwaysNullDecoder); + }); + + it('returns RawUndecodedEvent when decoder returns null', () => { + const event = makeNormalized({ eventName: 'bad_event' }); + const result = registry.decode(event); + expect(result.kind).toBe('raw'); + }); + + it('reason is "decode_failed" when decoder returns null', () => { + const event = makeNormalized({ eventName: 'bad_event' }); + const result = registry.decode(event); + if (result.kind !== 'raw') throw new Error('Expected raw'); + expect(result.reason).toBe('decode_failed'); + }); + + it('raw result still preserves source identifiers when decoder fails', () => { + const event = makeNormalized({ eventId: 'fail-event-id', eventName: 'bad_event' }); + const result = registry.decode(event); + expect(result.eventId).toBe('fail-event-id'); + }); +}); + +// --------------------------------------------------------------------------- +// Decoding — decoder throws (pipeline robustness, Task 3) +// --------------------------------------------------------------------------- + +describe('EventDecoderRegistry — decode (decoder throws)', () => { + let registry: EventDecoderRegistry; + + beforeEach(() => { + registry = new EventDecoderRegistry(); + registry.register(alwaysThrowsDecoder); + }); + + it('does not propagate decoder exceptions', () => { + const event = makeNormalized({ eventName: 'throwing_event' }); + expect(() => registry.decode(event)).not.toThrow(); + }); + + it('returns RawUndecodedEvent when decoder throws', () => { + const event = makeNormalized({ eventName: 'throwing_event' }); + const result = registry.decode(event); + expect(result.kind).toBe('raw'); + }); + + it('reason is "decode_failed" when decoder throws', () => { + const event = makeNormalized({ eventName: 'throwing_event' }); + const result = registry.decode(event); + if (result.kind !== 'raw') throw new Error('Expected raw'); + expect(result.reason).toBe('decode_failed'); + }); +}); + +// --------------------------------------------------------------------------- +// API distinguishes decoded vs raw (Task 2) +// --------------------------------------------------------------------------- + +describe('EventDecoderRegistry — API discriminates decoded vs raw (Task 2)', () => { + it('kind field is the sole discriminant — no instanceof checks needed', () => { + const registry = new EventDecoderRegistry(); + registry.register(notificationScheduledDecoder); + + const decoded = registry.decode(makeNormalized()); + const raw = registry.decode(makeNormalized({ eventName: 'unfamiliar_event' })); + + // Callers branch on `kind` only: + expect(decoded.kind).toBe('decoded'); + expect(raw.kind).toBe('raw'); + + // TypeScript narrowing: these properties exist on each branch + if (decoded.kind === 'decoded') { + expect(decoded.payload).toBeDefined(); + } + if (raw.kind === 'raw') { + expect(raw.rawTopic).toBeDefined(); + expect(raw.rawValue).toBeDefined(); + expect(raw.reason).toBeDefined(); + } + }); + + it('decoded result does not expose rawTopic or rawValue', () => { + const registry = new EventDecoderRegistry(); + registry.register(notificationScheduledDecoder); + + const result = registry.decode(makeNormalized()); + expect('rawTopic' in result).toBe(false); + expect('rawValue' in result).toBe(false); + }); + + it('raw result does not expose a payload', () => { + const registry = new EventDecoderRegistry(); + const result = registry.decode(makeNormalized({ eventName: 'unknown' })); + expect('payload' in result).toBe(false); + }); +}); + +// --------------------------------------------------------------------------- +// Singleton helpers +// --------------------------------------------------------------------------- + +describe('EventDecoderRegistry — singleton helpers', () => { + afterEach(() => { + resetEventDecoderRegistry(); + }); + + it('getEventDecoderRegistry returns an EventDecoderRegistry', () => { + const reg = getEventDecoderRegistry(); + expect(reg).toBeInstanceOf(EventDecoderRegistry); + }); + + it('getEventDecoderRegistry returns the same instance on repeated calls', () => { + const a = getEventDecoderRegistry(); + const b = getEventDecoderRegistry(); + expect(a).toBe(b); + }); + + it('setEventDecoderRegistry replaces the singleton', () => { + const custom = new EventDecoderRegistry(); + setEventDecoderRegistry(custom); + expect(getEventDecoderRegistry()).toBe(custom); + }); + + it('resetEventDecoderRegistry forces a fresh instance on next get', () => { + const first = getEventDecoderRegistry(); + resetEventDecoderRegistry(); + const second = getEventDecoderRegistry(); + expect(second).not.toBe(first); + }); +}); diff --git a/listener/src/services/event-decoder-registry.ts b/listener/src/services/event-decoder-registry.ts new file mode 100644 index 00000000..173e44ba --- /dev/null +++ b/listener/src/services/event-decoder-registry.ts @@ -0,0 +1,319 @@ +/** + * Event Decoder Registry (Task 3) + * + * Replaces scattered `if/else` or `switch` chains for event decoding with a + * clean, extensible registry pattern. Decoders are registered independently + * (plugin style) and unknown event types fall back to a `RawUndecodedEvent` + * without crashing the ingestion pipeline. + * + * Concepts + * -------- + * - An *EventDecoder* knows how to parse the raw topic/value strings of one + * specific event name into a strongly-typed `T`. + * - `DecodedEvent` is the successful result: source metadata + the typed + * payload. + * - `RawUndecodedEvent` is the fallback result: source identifiers and the + * verbatim raw payload are preserved so the event remains inspectable even + * when no decoder is registered (Task 2 requirement is satisfied here too). + * - `EventDecodeResult` is a discriminated union over the two above — the + * `kind` field lets callers branch without `instanceof` checks. + * + * Usage + * ----- + * ```ts + * // Register a decoder (e.g. in bootstrap): + * getEventDecoderRegistry().register('notification_scheduled', notificationScheduledDecoder); + * + * // In the ingestion pipeline: + * const result = getEventDecoderRegistry().decode(normalizedEvent); + * if (result.kind === 'decoded') { + * // result.payload has the typed fields + * } else { + * // result.rawTopic / result.rawValue preserved + * } + * ``` + */ + +import type { NormalizedEvent } from './event-normalization'; +import logger from '../utils/logger'; + +// --------------------------------------------------------------------------- +// Core types +// --------------------------------------------------------------------------- + +/** + * The input passed to every decoder: the source event name plus the already- + * formatted topic array and value string (from `normalizeSorobanEvent`). + */ +export interface DecoderInput { + /** Normalised event name extracted from topic[0]. */ + eventName: string; + /** Formatted topic strings — `topic[0]` is the event name symbol. */ + topic: string[]; + /** Formatted value string. */ + value: string; + /** Raw ledger number for ordering context. */ + ledger: number; + /** Transaction hash, or `null` when unavailable. */ + txHash: string | null; + /** Payload version tag from the event data, or `null` when absent. */ + payloadVersion: number | null; +} + +/** + * A registered decoder for a single event type `T`. + * + * `decode` must not throw — it should return `null` on any parse failure so + * the registry can fall back to `RawUndecodedEvent` without disrupting the + * pipeline. + */ +export interface EventDecoder { + /** + * Unique name that matches the on-chain Symbol emitted as topic[0] + * (e.g. `"notification_scheduled"`). + */ + readonly eventName: string; + /** + * Parse the input into a typed payload. Return `null` when the input + * cannot be fully parsed (the registry treats `null` as a decode failure + * and falls back to the raw event). + */ + decode(input: DecoderInput): T | null; +} + +/** + * A successfully decoded event with a typed payload. + * The `kind` field discriminates this from `RawUndecodedEvent`. + */ +export interface DecodedEvent { + kind: 'decoded'; + /** Unique event ID from the RPC response. */ + eventId: string; + /** Contract address that emitted the event. */ + contractAddress: string; + /** Decoded event name. */ + eventName: string; + /** Ledger number. */ + ledger: number; + /** Transaction hash. */ + txHash: string | null; + /** The typed, decoder-specific payload. */ + payload: T; +} + +/** + * An event whose raw payload is preserved verbatim because no decoder is + * registered for its event name, or the registered decoder returned `null`. + * + * The API intentionally distinguishes this from `DecodedEvent` so callers can + * route, log, or forward raw events without losing their source identifiers. + * (Satisfies the Task 2 inspectability requirement.) + */ +export interface RawUndecodedEvent { + kind: 'raw'; + /** Unique event ID from the RPC response — always preserved. */ + eventId: string; + /** Contract address that emitted the event — always preserved. */ + contractAddress: string; + /** + * Event name extracted from the first topic Symbol, or `null` when the + * topic could not be parsed (e.g. completely unknown encoding). + */ + eventName: string | null; + /** Ledger number. */ + ledger: number; + /** Transaction hash. */ + txHash: string | null; + /** Verbatim topic array as received from the normalization layer. */ + rawTopic: string[]; + /** Verbatim formatted value string. */ + rawValue: string; + /** Reason the event was not decoded. */ + reason: 'no_decoder' | 'decode_failed'; +} + +/** Discriminated union returned by `EventDecoderRegistry.decode`. */ +export type EventDecodeResult = DecodedEvent | RawUndecodedEvent; + +// --------------------------------------------------------------------------- +// Registry +// --------------------------------------------------------------------------- + +/** + * Central registry that maps event names to their decoders. + * + * - Decoders are registered with `register()` — any number may be registered + * at any time (plugin / lazy-load friendly). + * - `decode()` dispatches to the right decoder by `normalizedEvent.eventName`. + * - Unknown event names (or decoders that return `null`) produce a + * `RawUndecodedEvent` so the ingestion pipeline never crashes. + */ +export class EventDecoderRegistry { + // Use `unknown` internally; the generic flows through at the call site. + private readonly decoders = new Map>(); + + // ------------------------------------------------------------------------- + // Registration + // ------------------------------------------------------------------------- + + /** + * Register a decoder. Overwrites any previously registered decoder for the + * same `eventName`. + */ + register(decoder: EventDecoder): this { + const name = decoder.eventName; + if (!name || typeof name !== 'string') { + throw new Error('EventDecoderRegistry: decoder.eventName must be a non-empty string'); + } + if (this.decoders.has(name)) { + logger.warn('EventDecoderRegistry: overwriting existing decoder', { eventName: name }); + } + this.decoders.set(name, decoder as EventDecoder); + logger.info('EventDecoderRegistry: decoder registered', { eventName: name }); + return this; + } + + /** + * Remove the decoder for `eventName`. + * Returns `true` if a decoder was found and removed. + */ + unregister(eventName: string): boolean { + const removed = this.decoders.delete(eventName); + if (removed) { + logger.info('EventDecoderRegistry: decoder removed', { eventName }); + } + return removed; + } + + /** + * Returns `true` when a decoder is registered for `eventName`. + */ + has(eventName: string): boolean { + return this.decoders.has(eventName); + } + + /** + * List all registered event names. + */ + listRegistered(): string[] { + return [...this.decoders.keys()]; + } + + // ------------------------------------------------------------------------- + // Decoding + // ------------------------------------------------------------------------- + + /** + * Dispatch a normalised event to its registered decoder. + * + * - When no decoder exists for `normalizedEvent.eventName`, returns a + * `RawUndecodedEvent` with `reason: 'no_decoder'`. + * - When the decoder returns `null` (parse failure), returns a + * `RawUndecodedEvent` with `reason: 'decode_failed'`. + * - On success, returns a `DecodedEvent` with the typed payload. + * + * This method never throws. + */ + decode(normalizedEvent: NormalizedEvent): EventDecodeResult { + const { + eventId, + contractAddress, + eventName, + ledger, + txHash, + topic, + value, + payloadVersion, + } = normalizedEvent; + + const raw: Omit = { + kind: 'raw', + eventId, + contractAddress, + eventName, + ledger, + txHash, + rawTopic: topic, + rawValue: value, + }; + + if (!eventName || !this.decoders.has(eventName)) { + logger.debug('EventDecoderRegistry: no decoder registered for event', { + eventId, + contractAddress, + eventName, + }); + return { ...raw, reason: 'no_decoder' } as RawUndecodedEvent; + } + + const decoder = this.decoders.get(eventName)!; + const input: DecoderInput = { + eventName, + topic, + value, + ledger, + txHash, + payloadVersion, + }; + + try { + const payload = decoder.decode(input); + if (payload === null) { + logger.warn('EventDecoderRegistry: decoder returned null (decode failed)', { + eventId, + contractAddress, + eventName, + }); + return { ...raw, reason: 'decode_failed' } as RawUndecodedEvent; + } + return { + kind: 'decoded', + eventId, + contractAddress, + eventName, + ledger, + txHash, + payload: payload as T, + }; + } catch (err) { + logger.error('EventDecoderRegistry: decoder threw an error, falling back to raw', { + eventId, + contractAddress, + eventName, + error: err instanceof Error ? err.message : String(err), + }); + return { ...raw, reason: 'decode_failed' } as RawUndecodedEvent; + } + } + + /** + * Remove all registered decoders. Primarily for test teardowns. + */ + clear(): void { + this.decoders.clear(); + } +} + +// --------------------------------------------------------------------------- +// Module-level singleton +// --------------------------------------------------------------------------- + +let _registry: EventDecoderRegistry | null = null; + +/** Returns the module-level singleton decoder registry. */ +export function getEventDecoderRegistry(): EventDecoderRegistry { + if (!_registry) { + _registry = new EventDecoderRegistry(); + } + return _registry; +} + +/** Replace the singleton. Use in tests or during bootstrap. */ +export function setEventDecoderRegistry(registry: EventDecoderRegistry): void { + _registry = registry; +} + +/** Reset the singleton to `null`. Use in test teardowns. */ +export function resetEventDecoderRegistry(): void { + _registry = null; +} diff --git a/listener/src/services/rpc-capability-checker.test.ts b/listener/src/services/rpc-capability-checker.test.ts new file mode 100644 index 00000000..bf79c848 --- /dev/null +++ b/listener/src/services/rpc-capability-checker.test.ts @@ -0,0 +1,373 @@ +/** + * Tests for RpcCapabilityChecker (Task 4) + * + * Coverage: + * - All probes pass → CapabilityCheckPassed + * - One probe fails → CapabilityCheckFailed with actionable gaps + * - Multiple probes fail → all gaps are aggregated + * - METHOD_NOT_FOUND, RESPONSE_MALFORMED, RESPONSE_MISSING_FIELD codes + * - Probe timeout → gap with UNEXPECTED_ERROR code + * - Probe that throws → gap with UNEXPECTED_ERROR code + * - Custom probes (for unit-testing individual capabilities) + * - createCapabilityChecker factory smoke test + * - Failure message is non-empty and actionable + */ + +import * as StellarSDK from '@stellar/stellar-sdk'; +import { + RpcCapabilityChecker, + createCapabilityChecker, + type CapabilityCheckFailed, + type CapabilityCheckPassed, + type CapabilityGap, +} from './rpc-capability-checker'; + +// --------------------------------------------------------------------------- +// Mock server factory +// --------------------------------------------------------------------------- + +/** + * Build a minimal mock of `StellarSDK.rpc.Server`. + * Only the methods exercised by the checker need to be present. + */ +function makeMockServer(overrides: { + getLatestLedger?: () => Promise; + getEvents?: (req: unknown) => Promise; + getFeeStats?: () => Promise; +}): StellarSDK.rpc.Server { + return { + getLatestLedger: overrides.getLatestLedger ?? (() => Promise.resolve({ sequence: 5000 })), + getEvents: + overrides.getEvents ?? + (() => Promise.resolve({ events: [], latestLedger: 5000, cursor: '' })), + getFeeStats: + overrides.getFeeStats ?? (() => Promise.resolve({ sorobanInclusionFee: { min: '100' } })), + } as unknown as StellarSDK.rpc.Server; +} + +// --------------------------------------------------------------------------- +// All-pass scenarios +// --------------------------------------------------------------------------- + +describe('RpcCapabilityChecker — all probes pass', () => { + it('returns passed: true when all methods respond correctly', async () => { + const server = makeMockServer({}); + const checker = new RpcCapabilityChecker(server); + const result = await checker.check(); + expect(result.passed).toBe(true); + }); + + it('reports the correct number of probes run', async () => { + const server = makeMockServer({}); + const checker = new RpcCapabilityChecker(server); + const result = await checker.check(); + if (!result.passed) throw new Error('Expected passed'); + // Default probes: getLatestLedger, getEvents, getFeeStats + expect(result.probesRun).toBe(3); + }); + + it('includes a durationMs in the passed result', async () => { + const server = makeMockServer({}); + const checker = new RpcCapabilityChecker(server); + const result = await checker.check(); + expect(result.durationMs).toBeGreaterThanOrEqual(0); + }); +}); + +// --------------------------------------------------------------------------- +// getLatestLedger failures +// --------------------------------------------------------------------------- + +describe('RpcCapabilityChecker — getLatestLedger failures', () => { + it('reports METHOD_NOT_FOUND when getLatestLedger throws with -32601', async () => { + const server = makeMockServer({ + getLatestLedger: () => Promise.reject(new Error('method not found -32601')), + }); + const checker = new RpcCapabilityChecker(server); + const result = await checker.check(); + expect(result.passed).toBe(false); + if (result.passed) throw new Error('Expected failed'); + const gap = result.gaps.find((g) => g.method === 'getLatestLedger'); + expect(gap).toBeDefined(); + expect(gap!.code).toBe('METHOD_NOT_FOUND'); + expect(gap!.action).toBeTruthy(); + }); + + it('reports RESPONSE_MALFORMED when getLatestLedger returns null', async () => { + const server = makeMockServer({ + getLatestLedger: () => Promise.resolve(null), + }); + const checker = new RpcCapabilityChecker(server); + const result = await checker.check(); + expect(result.passed).toBe(false); + if (result.passed) throw new Error('Expected failed'); + const gap = result.gaps.find((g) => g.method === 'getLatestLedger'); + expect(gap!.code).toBe('RESPONSE_MALFORMED'); + }); + + it('reports RESPONSE_MISSING_FIELD when sequence is absent', async () => { + const server = makeMockServer({ + getLatestLedger: () => Promise.resolve({ status: 'ok' }), // no sequence + }); + const checker = new RpcCapabilityChecker(server); + const result = await checker.check(); + expect(result.passed).toBe(false); + if (result.passed) throw new Error('Expected failed'); + const gap = result.gaps.find((g) => g.method === 'getLatestLedger'); + expect(gap!.code).toBe('RESPONSE_MISSING_FIELD'); + }); + + it('accepts the `ledger` field as an alternative to `sequence`', async () => { + const server = makeMockServer({ + getLatestLedger: () => Promise.resolve({ ledger: 1234 }), + }); + const checker = new RpcCapabilityChecker(server); + const result = await checker.check(); + // getLatestLedger should pass — the other probes may still produce gaps + const gap = result.passed ? null : result.gaps.find((g) => g.method === 'getLatestLedger'); + expect(gap).toBeUndefined(); + }); +}); + +// --------------------------------------------------------------------------- +// getEvents failures +// --------------------------------------------------------------------------- + +describe('RpcCapabilityChecker — getEvents failures', () => { + it('reports METHOD_NOT_FOUND when getEvents throws with -32601', async () => { + const server = makeMockServer({ + getEvents: () => Promise.reject(new Error('Method not found: -32601')), + }); + const checker = new RpcCapabilityChecker(server); + const result = await checker.check(); + expect(result.passed).toBe(false); + if (result.passed) throw new Error('Expected failed'); + const gap = result.gaps.find((g) => g.method === 'getEvents'); + expect(gap!.code).toBe('METHOD_NOT_FOUND'); + expect(gap!.action).toMatch(/soroban-rpc/i); + }); + + it('ignores "start ledger too old" errors (method exists, node pruned early ledgers)', async () => { + const server = makeMockServer({ + getEvents: () => + Promise.reject(new Error('Start ledger is too old and not available on this node')), + }); + const checker = new RpcCapabilityChecker(server); + const result = await checker.check(); + // getEvents probe should pass; other probes may or may not pass + const gap = result.passed ? null : result.gaps.find((g) => g.method === 'getEvents'); + expect(gap).toBeUndefined(); + }); + + it('reports RESPONSE_MISSING_FIELD when events array is absent', async () => { + const server = makeMockServer({ + getEvents: () => Promise.resolve({ latestLedger: 100 }), // no `events` + }); + const checker = new RpcCapabilityChecker(server); + const result = await checker.check(); + expect(result.passed).toBe(false); + if (result.passed) throw new Error('Expected failed'); + const gap = result.gaps.find((g) => g.method === 'getEvents'); + expect(gap!.code).toBe('RESPONSE_MISSING_FIELD'); + }); +}); + +// --------------------------------------------------------------------------- +// Multiple probes fail — all gaps aggregated +// --------------------------------------------------------------------------- + +describe('RpcCapabilityChecker — multiple failures aggregated', () => { + it('collects gaps from all failing probes', async () => { + const server = makeMockServer({ + getLatestLedger: () => Promise.reject(new Error('method not found')), + getEvents: () => Promise.reject(new Error('method not found')), + }); + const checker = new RpcCapabilityChecker(server); + const result = await checker.check(); + expect(result.passed).toBe(false); + if (result.passed) throw new Error('Expected failed'); + expect(result.gaps.length).toBeGreaterThanOrEqual(2); + }); + + it('failure message lists all missing methods', async () => { + const server = makeMockServer({ + getLatestLedger: () => Promise.reject(new Error('method not found -32601')), + getEvents: () => Promise.reject(new Error('method not found -32601')), + }); + const checker = new RpcCapabilityChecker(server); + const result = (await checker.check()) as CapabilityCheckFailed; + expect(result.message).toMatch(/getLatestLedger/); + expect(result.message).toMatch(/getEvents/); + }); + + it('failure message is non-empty and contains action guidance', async () => { + const server = makeMockServer({ + getLatestLedger: () => Promise.reject(new Error('method not found -32601')), + }); + const checker = new RpcCapabilityChecker(server); + const result = (await checker.check()) as CapabilityCheckFailed; + expect(result.message.length).toBeGreaterThan(50); + expect(result.message).toMatch(/action/i); + }); +}); + +// --------------------------------------------------------------------------- +// Probe-level timeout +// --------------------------------------------------------------------------- + +describe('RpcCapabilityChecker — probe timeout', () => { + it('returns a gap with UNEXPECTED_ERROR code when a probe times out', async () => { + const neverResolves: CapabilityProbeType = { + name: 'slowProbe', + run: () => new Promise(() => {}), // never resolves + }; + const server = makeMockServer({}); + const checker = new RpcCapabilityChecker(server, { + timeoutMs: 10, // 10 ms — tiny on purpose + probes: [neverResolves], + }); + const result = await checker.check(); + expect(result.passed).toBe(false); + if (result.passed) throw new Error('Expected failed'); + const gap = result.gaps.find((g) => g.method === 'slowProbe'); + expect(gap).toBeDefined(); + expect(gap!.code).toBe('UNEXPECTED_ERROR'); + expect(gap!.detail).toMatch(/timed out/i); + }); +}); + +// --------------------------------------------------------------------------- +// Probe that throws +// --------------------------------------------------------------------------- + +describe('RpcCapabilityChecker — probe throws', () => { + it('captures thrown probe errors as UNEXPECTED_ERROR gaps', async () => { + const throwingProbe: CapabilityProbeType = { + name: 'throwingProbe', + run: () => Promise.reject(new Error('probe blew up')), + }; + const server = makeMockServer({}); + const checker = new RpcCapabilityChecker(server, { + probes: [throwingProbe], + }); + const result = await checker.check(); + expect(result.passed).toBe(false); + if (result.passed) throw new Error('Expected failed'); + expect(result.gaps[0].code).toBe('UNEXPECTED_ERROR'); + }); +}); + +// --------------------------------------------------------------------------- +// Custom probes (extensibility) +// --------------------------------------------------------------------------- + +type CapabilityProbeType = { + name: string; + run(server: StellarSDK.rpc.Server): Promise; +}; + +describe('RpcCapabilityChecker — custom probes', () => { + it('accepts an array of custom probes and runs only those', async () => { + let called = false; + const customProbe: CapabilityProbeType = { + name: 'customMethod', + run: async () => { + called = true; + return null; // pass + }, + }; + const server = makeMockServer({}); + const checker = new RpcCapabilityChecker(server, { probes: [customProbe] }); + const result = await checker.check(); + expect(called).toBe(true); + expect(result.passed).toBe(true); + if (!result.passed) throw new Error('Expected passed'); + expect(result.probesRun).toBe(1); + }); + + it('custom probe that fails is captured as a gap', async () => { + const failingProbe: CapabilityProbeType = { + name: 'missingFeature', + run: async () => ({ + method: 'missingFeature', + code: 'METHOD_NOT_FOUND' as const, + detail: 'missingFeature is not exposed by this endpoint.', + action: 'Upgrade your RPC node.', + }), + }; + const server = makeMockServer({}); + const checker = new RpcCapabilityChecker(server, { probes: [failingProbe] }); + const result = await checker.check(); + expect(result.passed).toBe(false); + if (result.passed) throw new Error('Expected failed'); + expect(result.gaps[0].method).toBe('missingFeature'); + }); +}); + +// --------------------------------------------------------------------------- +// createCapabilityChecker factory smoke test +// --------------------------------------------------------------------------- + +describe('createCapabilityChecker factory', () => { + it('returns an RpcCapabilityChecker instance', () => { + const checker = createCapabilityChecker('https://soroban-rpc.example.com'); + expect(checker).toBeInstanceOf(RpcCapabilityChecker); + }); +}); + +// --------------------------------------------------------------------------- +// Actionable error messages (core acceptance criterion) +// --------------------------------------------------------------------------- + +describe('RpcCapabilityChecker — actionable error messages', () => { + it('each gap has a non-empty action string', async () => { + const server = makeMockServer({ + getLatestLedger: () => Promise.reject(new Error('method not found -32601')), + getEvents: () => Promise.reject(new Error('method not found -32601')), + }); + const checker = new RpcCapabilityChecker(server); + const result = await checker.check(); + if (result.passed) throw new Error('Expected failed'); + for (const gap of result.gaps) { + expect(gap.action.trim().length).toBeGreaterThan(0); + } + }); + + it('each gap has a non-empty detail string', async () => { + const server = makeMockServer({ + getLatestLedger: () => Promise.reject(new Error('method not found -32601')), + }); + const checker = new RpcCapabilityChecker(server); + const result = await checker.check(); + if (result.passed) throw new Error('Expected failed'); + for (const gap of result.gaps) { + expect(gap.detail.trim().length).toBeGreaterThan(0); + } + }); + + it('failure message includes the number of checks that ran', async () => { + const server = makeMockServer({ + getLatestLedger: () => Promise.reject(new Error('method not found -32601')), + }); + const checker = new RpcCapabilityChecker(server); + const result = (await checker.check()) as CapabilityCheckFailed; + // message should be informative enough to drive operator action + expect(result.message).toMatch(/check/i); + }); + + it('system does not silently pass when a required method is absent', async () => { + // Scenario: endpoint is a bare Horizon node with no soroban-rpc methods + const horizonLikeServer = makeMockServer({ + getLatestLedger: () => + Promise.reject(new Error('Method not found -32601: getLatestLedger')), + getEvents: () => + Promise.reject(new Error('Method not found -32601: getEvents')), + getFeeStats: () => + Promise.reject(new Error('Method not found -32601: getFeeStats')), + }); + const checker = new RpcCapabilityChecker(horizonLikeServer); + const result = await checker.check(); + // Must FAIL — not silently pass + expect(result.passed).toBe(false); + }); +}); diff --git a/listener/src/services/rpc-capability-checker.ts b/listener/src/services/rpc-capability-checker.ts new file mode 100644 index 00000000..ba3156ca --- /dev/null +++ b/listener/src/services/rpc-capability-checker.ts @@ -0,0 +1,444 @@ +/** + * RPC Startup Capability Checker (Task 4) + * + * Probes an RPC endpoint at startup to verify that all required methods are + * available and return well-formed responses. If any requirement is missing + * or returns an unexpected shape, the checker produces a clear, actionable + * `CapabilityCheckFailure` listing exactly what is wrong — and the caller + * (typically `index.ts`) must treat this as a fatal startup error so the + * system never starts in a degraded, unsupported state. + * + * Design + * ------ + * Each *capability probe* is a small async function that tests one RPC method + * and returns either `null` (pass) or a `CapabilityGap` describing the problem. + * The checker runs all probes and aggregates results. + * + * Adding a new requirement is as simple as pushing a new `CapabilityProbe` + * into the `probes` array — no branching logic elsewhere needs changing. + * + * Usage + * ----- + * ```ts + * const checker = new RpcCapabilityChecker(server); + * const result = await checker.check(); + * if (!result.passed) { + * logger.error('RPC capability checks failed', result); + * process.exit(1); + * } + * ``` + */ + +import * as StellarSDK from '@stellar/stellar-sdk'; +import logger from '../utils/logger'; + +// --------------------------------------------------------------------------- +// Types +// --------------------------------------------------------------------------- + +/** + * A single unmet capability — describes what method is missing/broken and + * what the operator should do to resolve it. + */ +export interface CapabilityGap { + /** The RPC method name that was probed (e.g. `"getLatestLedger"`). */ + method: string; + /** Machine-readable code for programmatic filtering. */ + code: + | 'METHOD_NOT_FOUND' + | 'RESPONSE_MALFORMED' + | 'RESPONSE_MISSING_FIELD' + | 'UNEXPECTED_ERROR'; + /** Human-readable description of the problem. */ + detail: string; + /** Suggested action for the operator. */ + action: string; +} + +/** Successful outcome — all probes passed. */ +export interface CapabilityCheckPassed { + passed: true; + /** Total number of probes that ran. */ + probesRun: number; + /** Latency of the overall check in milliseconds. */ + durationMs: number; +} + +/** Failed outcome — one or more probes found gaps. */ +export interface CapabilityCheckFailed { + passed: false; + /** All capability gaps detected. */ + gaps: CapabilityGap[]; + /** Total number of probes that ran. */ + probesRun: number; + /** Latency of the overall check in milliseconds. */ + durationMs: number; + /** Consolidated, human-readable error message suitable for logs / process.exit output. */ + message: string; +} + +export type CapabilityCheckResult = CapabilityCheckPassed | CapabilityCheckFailed; + +/** + * Internal probe interface. Each probe tests one RPC requirement and returns + * `null` on success or a `CapabilityGap` on failure. + */ +interface CapabilityProbe { + name: string; + run(server: StellarSDK.rpc.Server): Promise; +} + +// --------------------------------------------------------------------------- +// Probe implementations +// --------------------------------------------------------------------------- + +/** + * Probe: `getLatestLedger` must return an object with a numeric `sequence`. + */ +const probeGetLatestLedger: CapabilityProbe = { + name: 'getLatestLedger', + async run(server) { + let response: unknown; + try { + response = await (server as any).getLatestLedger(); + } catch (err) { + const msg = err instanceof Error ? err.message : String(err); + if ( + msg.toLowerCase().includes('method not found') || + msg.toLowerCase().includes('-32601') + ) { + return { + method: 'getLatestLedger', + code: 'METHOD_NOT_FOUND', + detail: `RPC endpoint does not expose getLatestLedger: ${msg}`, + action: + 'Ensure the RPC endpoint is running a Stellar RPC node (soroban-rpc) and not a bare Horizon instance.', + }; + } + return { + method: 'getLatestLedger', + code: 'UNEXPECTED_ERROR', + detail: `getLatestLedger threw an unexpected error: ${msg}`, + action: 'Check RPC endpoint connectivity and logs.', + }; + } + + if (typeof response !== 'object' || response === null) { + return { + method: 'getLatestLedger', + code: 'RESPONSE_MALFORMED', + detail: 'getLatestLedger returned a non-object response.', + action: 'Verify the RPC endpoint is compatible with soroban-rpc v0.19+.', + }; + } + + const r = response as Record; + const sequence = + typeof r.sequence === 'number' + ? r.sequence + : typeof r.ledger === 'number' + ? r.ledger + : typeof r.latestLedger === 'number' + ? r.latestLedger + : null; + + if (sequence === null || !Number.isInteger(sequence) || sequence < 1) { + return { + method: 'getLatestLedger', + code: 'RESPONSE_MISSING_FIELD', + detail: + 'getLatestLedger response does not contain a valid `sequence` (or `ledger` / `latestLedger`) field.', + action: + 'Upgrade the RPC endpoint to a version that returns a numeric ledger sequence in getLatestLedger.', + }; + } + + return null; // pass + }, +}; + +/** + * Probe: `getEvents` must be callable with a minimal valid request and must + * return an object with an `events` array. + * + * We pass `startLedger: 1` which may return 0 events — that is fine; we are + * only checking the method exists and returns the expected shape. + */ +const probeGetEvents: CapabilityProbe = { + name: 'getEvents', + async run(server) { + let response: unknown; + try { + response = await server.getEvents({ + filters: [], + startLedger: 1, + limit: 1, + } as StellarSDK.rpc.Api.GetEventsRequest); + } catch (err) { + const msg = err instanceof Error ? err.message : String(err); + // A "start ledger too old" error means the method exists but the node + // has already pruned ledger 1. That is acceptable — the method is available. + if ( + msg.toLowerCase().includes('start ledger') || + msg.toLowerCase().includes('ledger is not') || + msg.toLowerCase().includes('too old') + ) { + return null; // method exists, node just pruned early ledgers + } + if ( + msg.toLowerCase().includes('method not found') || + msg.toLowerCase().includes('-32601') + ) { + return { + method: 'getEvents', + code: 'METHOD_NOT_FOUND', + detail: `RPC endpoint does not expose getEvents: ${msg}`, + action: + 'The listener requires soroban-rpc, which exposes getEvents. ' + + 'Horizon does not support this method.', + }; + } + return { + method: 'getEvents', + code: 'UNEXPECTED_ERROR', + detail: `getEvents threw an unexpected error during startup probe: ${msg}`, + action: 'Check RPC endpoint connectivity and logs.', + }; + } + + if (typeof response !== 'object' || response === null) { + return { + method: 'getEvents', + code: 'RESPONSE_MALFORMED', + detail: 'getEvents returned a non-object response.', + action: 'Verify the RPC endpoint is compatible with soroban-rpc v0.19+.', + }; + } + + const r = response as Record; + if (!Array.isArray(r.events)) { + return { + method: 'getEvents', + code: 'RESPONSE_MISSING_FIELD', + detail: 'getEvents response does not contain an `events` array field.', + action: + 'Upgrade the RPC endpoint to a version that includes an `events` array in getEvents responses.', + }; + } + + return null; // pass + }, +}; + +/** + * Probe: `getFeeStats` must be callable and return an object. + * + * This verifies the node is a modern soroban-rpc that exposes fee information, + * which the scheduler uses to cap transaction fees. A missing getFeeStats + * means the node may be too old or misconfigured. + */ +const probeGetFeeStats: CapabilityProbe = { + name: 'getFeeStats', + async run(server) { + try { + const response = await (server as any).getFeeStats?.(); + if (response === undefined) { + // getFeeStats is not exposed as a method on this SDK version — treat + // as unsupported rather than crashing. + return { + method: 'getFeeStats', + code: 'METHOD_NOT_FOUND', + detail: 'getFeeStats is not available on this RPC server or SDK version.', + action: + 'Upgrade the RPC endpoint to soroban-rpc v0.24+ to enable fee-aware transaction submission.', + }; + } + if (typeof response !== 'object' || response === null) { + return { + method: 'getFeeStats', + code: 'RESPONSE_MALFORMED', + detail: 'getFeeStats returned a non-object response.', + action: 'Verify the RPC endpoint version supports getFeeStats.', + }; + } + return null; // pass + } catch (err) { + const msg = err instanceof Error ? err.message : String(err); + if ( + msg.toLowerCase().includes('method not found') || + msg.toLowerCase().includes('-32601') + ) { + return { + method: 'getFeeStats', + code: 'METHOD_NOT_FOUND', + detail: `RPC endpoint does not expose getFeeStats: ${msg}`, + action: 'Upgrade to soroban-rpc v0.24+ to use fee-aware transaction handling.', + }; + } + // getFeeStats failing for any other reason is treated as a soft failure — + // return null to avoid blocking startup for non-critical capability. + return null; + } + }, +}; + +// --------------------------------------------------------------------------- +// Checker +// --------------------------------------------------------------------------- + +/** Options for `RpcCapabilityChecker`. */ +export interface RpcCapabilityCheckerOptions { + /** + * Timeout (ms) for the entire capability check. Individual probes that + * exceed this limit will be treated as failures. Default: 10 000 ms. + */ + timeoutMs?: number; + /** + * Override the set of probes to run. Useful in tests to isolate specific + * capabilities without stubbing every method. + * + * When omitted the default set (`probeGetLatestLedger`, `probeGetEvents`, + * `probeGetFeeStats`) is used. + */ + probes?: CapabilityProbe[]; +} + +/** + * Runs a set of capability probes against a Stellar RPC endpoint and + * aggregates the results into a single `CapabilityCheckResult`. + * + * All probes run concurrently (via `Promise.allSettled`) so a slow or broken + * method does not block other probes. + */ +export class RpcCapabilityChecker { + private readonly server: StellarSDK.rpc.Server; + private readonly timeoutMs: number; + private readonly probes: CapabilityProbe[]; + + constructor(server: StellarSDK.rpc.Server, options: RpcCapabilityCheckerOptions = {}) { + this.server = server; + this.timeoutMs = options.timeoutMs ?? 10_000; + this.probes = options.probes ?? [probeGetLatestLedger, probeGetEvents, probeGetFeeStats]; + } + + /** + * Run all probes and return the aggregated result. + * + * Does not throw — failures are captured in the returned `CapabilityCheckFailed`. + */ + async check(): Promise { + const start = Date.now(); + + logger.info('RpcCapabilityChecker: starting capability checks', { + probeCount: this.probes.length, + probes: this.probes.map((p) => p.name), + timeoutMs: this.timeoutMs, + }); + + const settled = await Promise.allSettled( + this.probes.map((probe) => this.runWithTimeout(probe)), + ); + + const gaps: CapabilityGap[] = []; + for (let i = 0; i < settled.length; i++) { + const outcome = settled[i]; + const probeName = this.probes[i].name; + + if (outcome.status === 'rejected') { + gaps.push({ + method: probeName, + code: 'UNEXPECTED_ERROR', + detail: `Probe "${probeName}" threw during execution: ${String(outcome.reason)}`, + action: 'Check RPC endpoint connectivity and listener logs.', + }); + } else if (outcome.value !== null) { + gaps.push(outcome.value); + } + } + + const durationMs = Date.now() - start; + + if (gaps.length === 0) { + logger.info('RpcCapabilityChecker: all checks passed', { + probesRun: this.probes.length, + durationMs, + }); + return { passed: true, probesRun: this.probes.length, durationMs }; + } + + const message = this.buildFailureMessage(gaps); + logger.error('RpcCapabilityChecker: capability checks failed', { + gapCount: gaps.length, + gaps, + probesRun: this.probes.length, + durationMs, + }); + + return { + passed: false, + gaps, + probesRun: this.probes.length, + durationMs, + message, + }; + } + + // ------------------------------------------------------------------------- + // Private helpers + // ------------------------------------------------------------------------- + + private async runWithTimeout(probe: CapabilityProbe): Promise { + return Promise.race([ + probe.run(this.server), + new Promise((resolve) => { + const timer = setTimeout(() => { + resolve({ + method: probe.name, + code: 'UNEXPECTED_ERROR', + detail: `Probe "${probe.name}" timed out after ${this.timeoutMs}ms.`, + action: `Check RPC endpoint responsiveness. The "${probe.name}" method did not respond within ${this.timeoutMs}ms.`, + }); + }, this.timeoutMs); + if (timer && typeof (timer as any).unref === 'function') { + (timer as any).unref(); + } + }), + ]); + } + + private buildFailureMessage(gaps: CapabilityGap[]): string { + const lines = [ + `RPC endpoint failed ${gaps.length} capability check(s). ` + + 'The listener will not start in a degraded state. ' + + 'Address the following issues:', + '', + ...gaps.map( + (g, i) => + ` [${i + 1}] Method: ${g.method}\n` + + ` Code: ${g.code}\n` + + ` Detail: ${g.detail}\n` + + ` Action: ${g.action}`, + ), + ]; + return lines.join('\n'); + } +} + +// --------------------------------------------------------------------------- +// Convenience factory +// --------------------------------------------------------------------------- + +/** + * Creates a `RpcCapabilityChecker` pre-wired to the given RPC URL. + * + * ```ts + * const result = await createCapabilityChecker(config.stellarRpcUrl).check(); + * ``` + */ +export function createCapabilityChecker( + rpcUrl: string, + options?: RpcCapabilityCheckerOptions, +): RpcCapabilityChecker { + const server = new StellarSDK.rpc.Server(rpcUrl); + return new RpcCapabilityChecker(server, options); +}