feat(indexer): event handler registry, stream/distribution handlers, and GraphQL schema

indexer/common/src/handlers/index.ts

@@ -0,0 +1,7 @@
+export { HandlerRegistry } from "./registry.js";
+export type {
+  EventHandler,
+  HandlerFilter,
+  HandlerResult,
+  SorobanEventInput,
+} from "./types.js";

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

@@ -0,0 +1,123 @@
+import { describe, expect, test, vi } from "vitest";
+
+import { HandlerRegistry } from "./registry.js";
+import type { EventHandler, HandlerResult, SorobanEventInput } from "./types.js";
+
+const baseEvent: SorobanEventInput = {
+  contractId: "CABC123",
+  ledger: 100,
+  ledgerClosedAt: "2024-01-01T00:00:00Z",
+  topic: ["stream_created"],
+  data: { amount: 1000 },
+  id: "event-1",
+  pagingToken: "paging-1",
+};
+
+const ok: HandlerResult = { ok: true };
+const makeHandler = (result: HandlerResult = ok): EventHandler =>
+  vi.fn().mockResolvedValue(result);
+
+describe("HandlerRegistry", () => {
+  describe("register and matches", () => {
+    test("matches on contractId", () => {
+      const registry = new HandlerRegistry();
+      const handler = makeHandler();
+      registry.register({ contractId: "CABC123" }, handler);
+
+      expect(registry.matches(baseEvent)).toEqual([handler]);
+    });
+
+    test("does not match wrong contractId", () => {
+      const registry = new HandlerRegistry();
+      registry.register({ contractId: "COTHER" }, makeHandler());
+
+      expect(registry.matches(baseEvent)).toHaveLength(0);
+    });
+
+    test("matches on topic", () => {
+      const registry = new HandlerRegistry();
+      const handler = makeHandler();
+      registry.register({ topic: "stream_created" }, handler);
+
+      expect(registry.matches(baseEvent)).toEqual([handler]);
+    });
+
+    test("does not match absent topic", () => {
+      const registry = new HandlerRegistry();
+      registry.register({ topic: "stream_cancelled" }, makeHandler());
+
+      expect(registry.matches(baseEvent)).toHaveLength(0);
+    });
+
+    test("matches on eventName alias", () => {
+      const registry = new HandlerRegistry();
+      const handler = makeHandler();
+      registry.register({ eventName: "stream_created" }, handler);
+
+      expect(registry.matches(baseEvent)).toEqual([handler]);
+    });
+
+    test("matches combined contractId + topic filter", () => {
+      const registry = new HandlerRegistry();
+      const handler = makeHandler();
+      registry.register({ contractId: "CABC123", topic: "stream_created" }, handler);
+
+      expect(registry.matches(baseEvent)).toEqual([handler]);
+      expect(
+        registry.matches({ ...baseEvent, contractId: "COTHER" }),
+      ).toHaveLength(0);
+    });
+
+    test("returns multiple handlers when several match", () => {
+      const registry = new HandlerRegistry();
+      const h1 = makeHandler();
+      const h2 = makeHandler();
+      registry.register({ contractId: "CABC123" }, h1);
+      registry.register({ topic: "stream_created" }, h2);
+
+      expect(registry.matches(baseEvent)).toEqual([h1, h2]);
+    });
+
+    test("empty filter matches every event", () => {
+      const registry = new HandlerRegistry();
+      const handler = makeHandler();
+      registry.register({}, handler);
+
+      expect(registry.matches(baseEvent)).toEqual([handler]);
+    });
+  });
+
+  describe("dispatch", () => {
+    test("calls all matched handlers and returns their results", async () => {
+      const registry = new HandlerRegistry();
+      const h1 = makeHandler({ ok: true });
+      const h2 = makeHandler({ ok: false, error: "boom", retriable: true });
+      registry.register({}, h1);
+      registry.register({}, h2);
+
+      const results = await registry.dispatch(baseEvent);
+
+      expect(h1).toHaveBeenCalledWith(baseEvent);
+      expect(h2).toHaveBeenCalledWith(baseEvent);
+      expect(results).toEqual([
+        { ok: true },
+        { ok: false, error: "boom", retriable: true },
+      ]);
+    });
+
+    test("returns empty array when no handlers match", async () => {
+      const registry = new HandlerRegistry();
+      const results = await registry.dispatch(baseEvent);
+      expect(results).toEqual([]);
+    });
+  });
+
+  describe("fluent API", () => {
+    test("register returns the registry for chaining", () => {
+      const registry = new HandlerRegistry();
+      const returned = registry.register({}, makeHandler());
+      expect(returned).toBe(registry);
+    });
+  });
+});
+

indexer/common/src/handlers/registry.ts

@@ -0,0 +1,50 @@
+import type {
+  EventHandler,
+  HandlerFilter,
+  HandlerResult,
+  SorobanEventInput,
+} from "./types.js";
+
+interface RegisteredHandler {
+  filter: HandlerFilter;
+  handler: EventHandler;
+}
+
+export class HandlerRegistry {
+  private readonly entries: RegisteredHandler[] = [];
+
+  register(filter: HandlerFilter, handler: EventHandler): this {
+    this.entries.push({ filter, handler });
+    return this;
+  }
+
+  matches(event: SorobanEventInput): EventHandler[] {
+    return this.entries
+      .filter(({ filter }) => {
+        if (filter.contractId && filter.contractId !== event.contractId) {
+          return false;
+        }
+        if (filter.topic && !event.topic.includes(filter.topic)) {
+          return false;
+        }
+        if (filter.eventName && !event.topic.includes(filter.eventName)) {
+          return false;
+        }
+        return true;
+      })
+      .map(({ handler }) => handler);
+  }
+
+  async dispatch(event: SorobanEventInput): Promise<HandlerResult[]> {
+    const handlers = this.matches(event);
+    return Promise.all(
+      handlers.map((h) =>
+        h(event).catch((err) => ({
+          ok: false as const,
+          error: err instanceof Error ? err.message : String(err),
+          retriable: true,
+        }))
+      )
+    );
+  }
+}

indexer/common/src/handlers/types.ts

@@ -0,0 +1,21 @@
+export interface SorobanEventInput {
+  contractId: string;
+  ledger: number;
+  ledgerClosedAt: string;
+  topic: string[];
+  data: unknown;
+  id: string;
+  pagingToken: string;
+}
+
+export type HandlerResult =
+  | { ok: true }
+  | { ok: false; error: string; retriable: boolean };
+
+export type EventHandler = (event: SorobanEventInput) => Promise<HandlerResult>;
+
+export interface HandlerFilter {
+  contractId?: string;
+  topic?: string;
+  eventName?: string;
+}

indexer/common/src/index.ts

@@ -8,3 +8,4 @@ export { createSorobanClient, sorobanClient } from "./rpc/client.js";
 export { IndexedEvent } from "./db/entity/IndexedEvent.js";
 export { EventRepository } from "./db/repository.js";
 export { SorobanPoller, type PollerOptions, type PollResult } from "./poller/index.js";
+export * from "./handlers/index.js";

indexer/distributions/src/handlers/distribution-created.handler.ts

@@ -0,0 +1,76 @@
+import type {
+  EventHandler,
+  HandlerResult,
+  SorobanEventInput,
+} from "@fundable-indexer/common";
+import { parseDistributionCreated } from "./types.js";
+
+export const distributionCreatedHandler: EventHandler = async (
+  event: SorobanEventInput,
+): Promise<HandlerResult> => {
+  try {
+    const payload = parseDistributionCreated(event.data);
+
+    if (!payload.distributionId) {
+      return {
+        ok: false,
+        error: "Missing distributionId in distribution_created event",
+        retriable: false,
+      };
+    }
+
+    if (!payload.creator) {
+      return {
+        ok: false,
+        error: "Missing creator in distribution_created event",
+        retriable: false,
+      };
+    }
+
+    if (!payload.token) {
+      return {
+        ok: false,
+        error: "Missing token in distribution_created event",
+        retriable: false,
+      };
+    }
+
+    if (!payload.totalAmount) {
+      return {
+        ok: false,
+        error: "Missing totalAmount in distribution_created event",
+        retriable: false,
+      };
+    }
+
+    if (!payload.transactionHash) {
+      return {
+        ok: false,
+        error: "Missing transactionHash in distribution_created event",
+        retriable: false,
+      };
+    }
+
+    if (payload.recipientCount <= 0) {
+      return {
+        ok: false,
+        error: "Invalid recipientCount in distribution_created event",
+        retriable: false,
+      };
+    }
+
+    // TODO(#36): persist distribution batch via repository once DB schema is merged
+    // TODO(#27): record indexed event identity via event repository
+    console.info(
+      `[distribution-created] id=${payload.distributionId} creator=${payload.creator} token=${payload.token} total=${payload.totalAmount} ledger=${event.ledger}`,
+    );
+
+    return { ok: true };
+  } catch (err) {
+    return {
+      ok: false,
+      error: err instanceof Error ? err.message : String(err),
+      retriable: true,
+    };
+  }
+};