feat(indexer): add handler registry, stream lifecycle handlers, distribution event handlers

indexer/common/src/handlers/registry.test.ts

@@ -0,0 +1,180 @@
+import { describe, expect, test, vi } from "vitest";
+
+import { EventHandlerRegistry, createHandlerRegistry } from "./registry.js";
+import type { HandlerResult, SorobanEvent } from "./types.js";
+
+function makeEvent(overrides?: Partial<SorobanEvent>): SorobanEvent {
+  return {
+    contractId: "CONTRACT_A",
+    ledger: 1000,
+    txHash: "TX_HASH_1",
+    eventIndex: 0,
+    topics: ["StreamCreated"],
+    data: JSON.stringify({ streamId: "s-1" }),
+    ...overrides,
+  };
+}
+
+const OK: HandlerResult = { success: true, summary: "ok" };
+
+describe("EventHandlerRegistry (issue #30)", () => {
+  describe("registration and matching", () => {
+    test("a handler with no filter fields matches every event", async () => {
+      const registry = createHandlerRegistry();
+      const handler = vi.fn().mockResolvedValue(OK);
+
+      registry.register({}, handler);
+
+      const eventA = makeEvent({ contractId: "A" });
+      const eventB = makeEvent({ contractId: "B", topics: ["Transfer"] });
+
+      expect(registry.match(eventA)).toHaveLength(1);
+      expect(registry.match(eventB)).toHaveLength(1);
+    });
+
+    test("contractId filter only matches events from that contract", () => {
+      const registry = createHandlerRegistry();
+      registry.register({ contractId: "CONTRACT_A" }, vi.fn().mockResolvedValue(OK));
+
+      expect(registry.match(makeEvent({ contractId: "CONTRACT_A" }))).toHaveLength(1);
+      expect(registry.match(makeEvent({ contractId: "CONTRACT_B" }))).toHaveLength(0);
+    });
+
+    test("eventName filter only matches events whose first topic equals it", () => {
+      const registry = createHandlerRegistry();
+      registry.register({ eventName: "StreamCreated" }, vi.fn().mockResolvedValue(OK));
+
+      expect(registry.match(makeEvent({ topics: ["StreamCreated"] }))).toHaveLength(1);
+      expect(registry.match(makeEvent({ topics: ["StreamCancelled"] }))).toHaveLength(0);
+    });
+
+    test("both contractId and eventName must match", () => {
+      const registry = createHandlerRegistry();
+      registry.register(
+        { contractId: "CONTRACT_A", eventName: "StreamCreated" },
+        vi.fn().mockResolvedValue(OK),
+      );
+
+      expect(
+        registry.match(makeEvent({ contractId: "CONTRACT_A", topics: ["StreamCreated"] })),
+      ).toHaveLength(1);
+
+      // contract matches but event name does not
+      expect(
+        registry.match(makeEvent({ contractId: "CONTRACT_A", topics: ["Transfer"] })),
+      ).toHaveLength(0);
+
+      // event name matches but contract does not
+      expect(
+        registry.match(makeEvent({ contractId: "CONTRACT_B", topics: ["StreamCreated"] })),
+      ).toHaveLength(0);
+    });
+
+    test("multiple handlers can match the same event", () => {
+      const registry = createHandlerRegistry();
+      registry.register({ eventName: "StreamCreated" }, vi.fn().mockResolvedValue(OK));
+      registry.register({ contractId: "CONTRACT_A" }, vi.fn().mockResolvedValue(OK));
+      registry.register({}, vi.fn().mockResolvedValue(OK));
+
+      expect(registry.match(makeEvent())).toHaveLength(3);
+    });
+
+    test("returns handlers in registration order", async () => {
+      const registry = createHandlerRegistry();
+      const order: number[] = [];
+
+      registry.register({}, async () => { order.push(1); return OK; });
+      registry.register({}, async () => { order.push(2); return OK; });
+      registry.register({}, async () => { order.push(3); return OK; });
+
+      await registry.dispatch(makeEvent());
+
+      expect(order).toEqual([1, 2, 3]);
+    });
+  });
+
+  describe("dispatch", () => {
+    test("calls all matching handlers and collects results", async () => {
+      const registry = createHandlerRegistry();
+      const h1 = vi.fn().mockResolvedValue({ success: true, summary: "h1" });
+      const h2 = vi.fn().mockResolvedValue({ success: true, summary: "h2" });
+
+      registry.register({ eventName: "StreamCreated" }, h1);
+      registry.register({ eventName: "StreamCreated" }, h2);
+
+      const results = await registry.dispatch(makeEvent());
+
+      expect(results).toEqual([
+        { success: true, summary: "h1" },
+        { success: true, summary: "h2" },
+      ]);
+      expect(h1).toHaveBeenCalledOnce();
+      expect(h2).toHaveBeenCalledOnce();
+    });
+
+    test("catches a throwing handler and records success:false without stopping others", async () => {
+      const registry = createHandlerRegistry();
+      const failing = vi.fn().mockRejectedValue(new Error("db connection lost"));
+      const succeeding = vi.fn().mockResolvedValue({ success: true });
+
+      registry.register({}, failing);
+      registry.register({}, succeeding);
+
+      const results = await registry.dispatch(makeEvent());
+
+      expect(results[0]).toEqual({ success: false, summary: "db connection lost" });
+      expect(results[1]).toEqual({ success: true });
+      expect(succeeding).toHaveBeenCalledOnce();
+    });
+
+    test("returns empty array when no handlers match", async () => {
+      const registry = createHandlerRegistry();
+      registry.register({ eventName: "StreamCreated" }, vi.fn().mockResolvedValue(OK));
+
+      const results = await registry.dispatch(makeEvent({ topics: ["Transfer"] }));
+
+      expect(results).toEqual([]);
+    });
+
+    test("passes the full event to each handler", async () => {
+      const registry = createHandlerRegistry();
+      const handler = vi.fn().mockResolvedValue(OK);
+
+      registry.register({}, handler);
+
+      const event = makeEvent({ contractId: "SPECIFIC", ledger: 9999, txHash: "TX_XYZ" });
+      await registry.dispatch(event);
+
+      expect(handler).toHaveBeenCalledWith(event);
+    });
+
+    test("is idempotent — dispatching the same event twice calls handlers twice", async () => {
+      const registry = createHandlerRegistry();
+      const handler = vi.fn().mockResolvedValue(OK);
+      registry.register({}, handler);
+
+      const event = makeEvent();
+      await registry.dispatch(event);
+      await registry.dispatch(event);
+
+      expect(handler).toHaveBeenCalledTimes(2);
+    });
+  });
+
+  describe("createHandlerRegistry factory", () => {
+    test("returns a fresh registry with no handlers", async () => {
+      const registry = createHandlerRegistry();
+      expect(registry.match(makeEvent())).toHaveLength(0);
+    });
+
+    test("two registries are independent", () => {
+      const r1 = createHandlerRegistry();
+      const r2 = createHandlerRegistry();
+
+      r1.register({}, vi.fn().mockResolvedValue(OK));
+
+      expect(r1.match(makeEvent())).toHaveLength(1);
+      expect(r2.match(makeEvent())).toHaveLength(0);
+    });
+  });
+});

indexer/common/src/handlers/registry.ts

@@ -0,0 +1,58 @@
+import type {
+  EventHandler,
+  HandlerEntry,
+  HandlerFilter,
+  HandlerRegistry,
+  HandlerResult,
+  SorobanEvent,
+} from "./types.js";
+
+function matchesFilter(filter: HandlerFilter, event: SorobanEvent): boolean {
+  if (filter.contractId !== undefined && filter.contractId !== event.contractId) {
+    return false;
+  }
+  if (filter.eventName !== undefined && filter.eventName !== event.topics[0]) {
+    return false;
+  }
+  return true;
+}
+
+export class EventHandlerRegistry implements HandlerRegistry {
+  private readonly entries: HandlerEntry[] = [];
+
+  register<TEvent extends SorobanEvent>(
+    filter: HandlerFilter,
+    handler: EventHandler<TEvent>,
+  ): void {
+    this.entries.push({ filter, handler: handler as EventHandler });
+  }
+
+  match(event: SorobanEvent): ReadonlyArray<EventHandler> {
+    return this.entries
+      .filter((e) => matchesFilter(e.filter, event))
+      .map((e) => e.handler);
+  }
+
+  async dispatch(event: SorobanEvent): Promise<HandlerResult[]> {
+    const handlers = this.match(event);
+    const results: HandlerResult[] = [];
+
+    for (const handler of handlers) {
+      try {
+        const result = await handler(event);
+        results.push(result);
+      } catch (error) {
+        results.push({
+          success: false,
+          summary: error instanceof Error ? error.message : String(error),
+        });
+      }
+    }
+
+    return results;
+  }
+}
+
+export function createHandlerRegistry(): HandlerRegistry {
+  return new EventHandlerRegistry();
+}

indexer/common/src/handlers/types.ts

@@ -0,0 +1,79 @@
+/**
+ * Core handler input/result types and registry interface for the Fundable
+ * Soroban event indexer (issue #30).
+ *
+ * Domain packages import these shared types so they can register event handlers
+ * without coupling to the poller internals.
+ */
+
+/** Identity fields that uniquely identify any Soroban event. */
+export interface SorobanEventIdentity {
+  contractId: string;
+  ledger: number;
+  txHash: string;
+  eventIndex: number;
+}
+
+/** Minimal shape of a Soroban event delivered to handlers. */
+export interface SorobanEvent extends SorobanEventIdentity {
+  /** Array of XDR-encoded topics. The first element is conventionally the event name. */
+  topics: string[];
+  /** JSON-encoded event data. */
+  data: string;
+}
+
+/** What every handler must return. */
+export interface HandlerResult {
+  /** Whether the handler processed the event without error. */
+  success: boolean;
+  /** Human-readable summary (logged at debug level). */
+  summary?: string;
+}
+
+/** The function signature every event handler must conform to. */
+export type EventHandler<TEvent extends SorobanEvent = SorobanEvent> = (
+  event: TEvent,
+) => Promise<HandlerResult>;
+
+/** Criteria used to match events to registered handlers. */
+export interface HandlerFilter {
+  /** Match only events emitted by this contract address. Omit to match all. */
+  contractId?: string;
+  /** Match only events whose first topic equals this value (the event name). */
+  eventName?: string;
+}
+
+/** A handler entry stored in the registry. */
+export interface HandlerEntry<TEvent extends SorobanEvent = SorobanEvent> {
+  filter: HandlerFilter;
+  handler: EventHandler<TEvent>;
+}
+
+/** Public API of the handler registry. */
+export interface HandlerRegistry {
+  /**
+   * Register a handler that will be called for events matching `filter`.
+   * Multiple handlers may match the same event; all are called in
+   * registration order.
+   */
+  register<TEvent extends SorobanEvent>(
+    filter: HandlerFilter,
+    handler: EventHandler<TEvent>,
+  ): void;
+
+  /**
+   * Return all handlers whose filter matches the given event.
+   * Matching rules:
+   * - If `filter.contractId` is set, the event's `contractId` must equal it.
+   * - If `filter.eventName` is set, the event's first topic must equal it.
+   * - A filter with no fields matches every event.
+   */
+  match(event: SorobanEvent): ReadonlyArray<EventHandler>;
+
+  /**
+   * Dispatch an event to every matching handler sequentially.
+   * Returns the array of results in the order handlers were called.
+   * A handler that throws is caught; `success: false` is recorded for it.
+   */
+  dispatch(event: SorobanEvent): Promise<HandlerResult[]>;
+}

indexer/common/src/index.ts

@@ -2,3 +2,14 @@ export const commonPackage = {
   name: "@fundable-indexer/common",
   role: "shared-infrastructure",
 } as const;
+
+export { createHandlerRegistry, EventHandlerRegistry } from "./handlers/registry.js";
+export type {
+  EventHandler,
+  HandlerEntry,
+  HandlerFilter,
+  HandlerRegistry,
+  HandlerResult,
+  SorobanEvent,
+  SorobanEventIdentity,
+} from "./handlers/types.js";