feat: add Server-Sent Events endpoint

backend/src/events/__tests__/events.integration.spec.ts

@@ -0,0 +1,241 @@
+/**
+ * SSE events integration tests.
+ *
+ * Tests the full pipeline: ClaimEventsService.publish() -> Redis pub/sub ->
+ * SseConnectionRegistry.broadcast() -> Subject emission.
+ *
+ * Uses an in-memory Redis mock (no real Redis required for CI).
+ */
+
+import { Test, TestingModule } from "@nestjs/testing";
+import { INestApplication, HttpStatus } from "@nestjs/common";
+import * as request from "supertest";
+import { ConfigModule } from "@nestjs/config";
+import { EventsModule } from "../events.module";
+import { ClaimEventsService, ClaimStatusChangedEvent } from "../claim-events.service";
+import { SseConnectionRegistry } from "../sse-connection.registry";
+
+// ── Helpers ───────────────────────────────────────────────────────────────────
+
+function createMockRedis() {
+  const subscribers: Map<string, ((channel: string, msg: string) => void)[]> = new Map();
+  const publishedMessages: { channel: string; message: string }[] = [];
+
+  const instance = {
+    _isConnected: false,
+    connect: jest.fn(async () => {
+      instance._isConnected = true;
+    }),
+    quit: jest.fn(async () => {}),
+    subscribe: jest.fn(async (channel: string) => {
+      if (!subscribers.has(channel)) subscribers.set(channel, []);
+    }),
+    unsubscribe: jest.fn(async () => {}),
+    publish: jest.fn(async (channel: string, message: string) => {
+      publishedMessages.push({ channel, message });
+      // Simulate local delivery
+      const handlers = subscribers.get(channel) ?? [];
+      for (const h of handlers) h(channel, message);
+      return 1;
+    }),
+    on: jest.fn((event: string, handler: (...args: unknown[]) => void) => {
+      if (event === "message") {
+        const channel = "claim:status:changed";
+        if (!subscribers.has(channel)) subscribers.set(channel, []);
+        subscribers.get(channel)!.push(handler as (channel: string, msg: string) => void);
+      }
+    }),
+    _getPublished: () => publishedMessages,
+  };
+
+  return instance;
+}
+
+// ── Tests ─────────────────────────────────────────────────────────────────────
+
+describe("EventsController SSE (integration)", () => {
+  let app: INestApplication;
+  let claimEventsService: ClaimEventsService;
+  let registry: SseConnectionRegistry;
+
+  beforeEach(async () => {
+    const moduleRef: TestingModule = await Test.createTestingModule({
+      imports: [
+        ConfigModule.forRoot({
+          isGlobal: true,
+          ignoreEnvFile: true,
+          load: [
+            () => ({
+              REDIS_URL: "redis://mock:6379",
+              SSE_MAX_CONNECTIONS: 10,
+            }),
+          ],
+        }),
+        EventsModule,
+      ],
+    }).compile();
+
+    app = moduleRef.createNestApplication();
+    app.setGlobalPrefix("api");
+
+    claimEventsService = moduleRef.get(ClaimEventsService);
+    registry = moduleRef.get(SseConnectionRegistry);
+
+    // Mock Redis connections — prevent real network calls in CI
+    const mockRedis = createMockRedis();
+    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    (claimEventsService as any).subscriber = mockRedis;
+    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    (claimEventsService as any).publisher = mockRedis;
+
+    await app.init();
+  });
+
+  afterEach(async () => {
+    await app.close();
+  });
+
+  // ── Connection validation ─────────────────────────────────────────────────
+
+  it("returns 400 when no claimId is provided", async () => {
+    const res = await request(app.getHttpServer()).get("/api/events/claims");
+    expect(res.status).toBe(HttpStatus.BAD_REQUEST);
+  });
+
+  // ── Registry unit tests ───────────────────────────────────────────────────
+
+  describe("SseConnectionRegistry", () => {
+    it("registers and unregisters connections", () => {
+      const conn = registry.register("conn-1", ["42", "43"]);
+      expect(registry.activeCount()).toBe(1);
+      expect(conn.claimIds.has("42")).toBe(true);
+
+      registry.unregister("conn-1");
+      expect(registry.activeCount()).toBe(0);
+    });
+
+    it("broadcasts only to connections watching the given claimId", () => {
+      const conn1 = registry.register("conn-a", ["1"]);
+      const conn2 = registry.register("conn-b", ["2"]);
+
+      const received1: object[] = [];
+      const received2: object[] = [];
+
+      conn1.subject.subscribe((e) => received1.push(e.data as object));
+      conn2.subject.subscribe((e) => received2.push(e.data as object));
+
+      registry.broadcast("1", { claimId: "1", status: "approved" });
+
+      expect(received1).toHaveLength(1);
+      expect(received2).toHaveLength(0); // conn2 watches claimId=2, not 1
+
+      registry.unregister("conn-a");
+      registry.unregister("conn-b");
+    });
+
+    it("rejects registration when connection limit is reached", () => {
+      // Limit is 10 in test config
+      for (let i = 0; i < 10; i++) {
+        registry.register(`conn-${i}`, ["1"]);
+      }
+
+      expect(() => registry.register("conn-overflow", ["1"])).toThrow();
+
+      // Cleanup
+      for (let i = 0; i < 10; i++) {
+        registry.unregister(`conn-${i}`);
+      }
+    });
+
+    it("drainAll completes all subjects", (done) => {
+      const conn = registry.register("drain-test", ["99"]);
+      let completed = false;
+      conn.subject.subscribe({
+        complete: () => {
+          completed = true;
+        },
+      });
+
+      registry.drainAll();
+
+      setImmediate(() => {
+        expect(completed).toBe(true);
+        done();
+      });
+    });
+  });
+
+  // ── ClaimEventsService unit tests ─────────────────────────────────────────
+
+  describe("ClaimEventsService", () => {
+    it("publish() sends event to Redis channel", async () => {
+      const mockRedis = createMockRedis();
+      // eslint-disable-next-line @typescript-eslint/no-explicit-any
+      (claimEventsService as any).publisher = mockRedis;
+
+      const event: ClaimStatusChangedEvent = {
+        claimId: "77",
+        status: "approved",
+        updatedAt: new Date().toISOString(),
+        ledger: 12345,
+      };
+
+      await claimEventsService.publish(event);
+
+      expect(mockRedis.publish).toHaveBeenCalledWith(
+        "claim:status:changed",
+        JSON.stringify(event),
+      );
+    });
+
+    it("publish() fails silently when Redis is unavailable", async () => {
+      // eslint-disable-next-line @typescript-eslint/no-explicit-any
+      (claimEventsService as any).publisher = {
+        publish: jest.fn().mockRejectedValue(new Error("Redis down")),
+      };
+
+      const event: ClaimStatusChangedEvent = {
+        claimId: "1",
+        status: "pending",
+        updatedAt: new Date().toISOString(),
+      };
+
+      await expect(claimEventsService.publish(event)).resolves.toBeUndefined();
+    });
+
+    it("handleMessage broadcasts to matching SSE connections", () => {
+      const conn = registry.register("msg-test", ["55"]);
+      const received: object[] = [];
+      conn.subject.subscribe((e) => received.push(e.data as object));
+
+      const event: ClaimStatusChangedEvent = {
+        claimId: "55",
+        status: "paid",
+        updatedAt: new Date().toISOString(),
+      };
+
+      // Trigger internal handler directly
+      // eslint-disable-next-line @typescript-eslint/no-explicit-any
+      (claimEventsService as any).handleMessage(JSON.stringify(event));
+
+      expect(received).toHaveLength(1);
+      expect((received[0] as ClaimStatusChangedEvent).status).toBe("paid");
+
+      registry.unregister("msg-test");
+    });
+
+    it("handleMessage ignores malformed JSON without throwing", () => {
+      expect(() => {
+        // eslint-disable-next-line @typescript-eslint/no-explicit-any
+        (claimEventsService as any).handleMessage("not-json");
+      }).not.toThrow();
+    });
+
+    it("handleMessage ignores events missing required fields", () => {
+      expect(() => {
+        // eslint-disable-next-line @typescript-eslint/no-explicit-any
+        (claimEventsService as any).handleMessage(JSON.stringify({ claimId: "1" }));
+      }).not.toThrow();
+    });
+  });
+});

backend/src/events/claim-events.service.ts

@@ -0,0 +1,133 @@
+/**
+ * ClaimEventsService
+ *
+ * Bridges the Redis pub/sub channel `claim:status:changed` to active SSE
+ * connections via the SseConnectionRegistry.
+ *
+ * Internal bus design:
+ *   The indexer's handleClaimFiled / handleClaimProcessed / handleVoteCast
+ *   methods call ClaimEventsService.publish() after each Prisma upsert. This
+ *   keeps the SSE bus decoupled from the Prisma transaction — the transaction
+ *   commits first, then the event is published. No DB transaction is held open
+ *   during SSE delivery.
+ *
+ *   In a multi-instance deployment, each backend instance subscribes to the
+ *   same Redis channel. When any instance calls publish(), all instances
+ *   receive the message and broadcast to their local SSE connections. This
+ *   ensures horizontal scale without a centralised fan-out process.
+ *
+ * Redis channel: `claim:status:changed`
+ * Message format: JSON { claimId: string; status: string; updatedAt: string; ledger?: number }
+ */
+
+import { Injectable, Logger, OnModuleInit, OnModuleDestroy } from "@nestjs/common";
+import Redis from "ioredis";
+import { ConfigService } from "@nestjs/config";
+import { SseConnectionRegistry } from "./sse-connection.registry";
+
+export interface ClaimStatusChangedEvent {
+  claimId: string;
+  status: string;
+  updatedAt: string;
+  ledger?: number;
+}
+
+const REDIS_CHANNEL = "claim:status:changed";
+
+@Injectable()
+export class ClaimEventsService implements OnModuleInit, OnModuleDestroy {
+  private readonly logger = new Logger(ClaimEventsService.name);
+  private subscriber: Redis | null = null;
+  private publisher: Redis | null = null;
+
+  constructor(
+    private readonly config: ConfigService,
+    private readonly registry: SseConnectionRegistry,
+  ) {}
+
+  async onModuleInit(): Promise<void> {
+    const redisUrl = this.config.get<string>("REDIS_URL", "redis://localhost:6379");
+
+    // Dedicated subscriber connection — subscribe() blocks the connection for
+    // pub/sub mode, so it cannot be shared with the general RedisService client.
+    this.subscriber = new Redis(redisUrl, {
+      lazyConnect: true,
+      retryStrategy: (times) => (times > 5 ? null : Math.min(times * 200, 3000)),
+    });
+
+    // Dedicated publisher connection
+    this.publisher = new Redis(redisUrl, {
+      lazyConnect: true,
+      retryStrategy: (times) => (times > 5 ? null : Math.min(times * 200, 3000)),
+    });
+
+    this.subscriber.on("error", (err) =>
+      this.logger.warn(`SSE Redis subscriber error: ${err.message}`),
+    );
+    this.publisher.on("error", (err) =>
+      this.logger.warn(`SSE Redis publisher error: ${err.message}`),
+    );
+
+    try {
+      await this.subscriber.connect();
+      await this.publisher.connect();
+      await this.subscriber.subscribe(REDIS_CHANNEL);
+      this.logger.log(`Subscribed to Redis channel: ${REDIS_CHANNEL}`);
+
+      this.subscriber.on("message", (_channel: string, message: string) => {
+        this.handleMessage(message);
+      });
+    } catch (err) {
+      // Non-fatal at startup: SSE will not push events but the rest of the
+      // application remains functional. Log a clear warning so operators notice.
+      this.logger.warn(
+        `ClaimEventsService: Redis unavailable at startup. SSE push disabled. Error: ${err}`,
+      );
+    }
+  }
+
+  async onModuleDestroy(): Promise<void> {
+    this.registry.drainAll();
+    try {
+      await this.subscriber?.unsubscribe(REDIS_CHANNEL);
+      await this.subscriber?.quit();
+      await this.publisher?.quit();
+    } catch {
+      // Ignore shutdown errors
+    }
+  }
+
+  /**
+   * Publish a claim status change to the Redis channel.
+   * Call this from the indexer after a successful Prisma upsert.
+   * The Prisma transaction must be committed before calling this method.
+   */
+  async publish(event: ClaimStatusChangedEvent): Promise<void> {
+    if (!this.publisher) return;
+
+    try {
+      await this.publisher.publish(REDIS_CHANNEL, JSON.stringify(event));
+    } catch (err) {
+      // Fail silently — pub/sub failure does not affect data integrity
+      this.logger.warn(`Failed to publish claim event: ${err}`);
+    }
+  }
+
+  private handleMessage(message: string): void {
+    let event: ClaimStatusChangedEvent;
+
+    try {
+      event = JSON.parse(message) as ClaimStatusChangedEvent;
+    } catch {
+      this.logger.warn(`Received malformed message on ${REDIS_CHANNEL}: ${message}`);
+      return;
+    }
+
+    if (!event.claimId || !event.status) {
+      this.logger.warn(`Ignoring incomplete claim event: ${message}`);
+      return;
+    }
+
+    this.registry.broadcast(event.claimId, event);
+  }
+}