fix: one multi-contract EventWatcher (collapse per-membership watchers)

deno.json

@@ -1,6 +1,6 @@
 {
   "name": "@moonlight-protocol/provider-platform",
-  "version": "0.9.1",
+  "version": "0.9.2",
   "license": "MIT",
   "exports": "./src/main.ts",
   "tasks": {

src/core/service/event-watcher/event-watcher.process.test.ts

@@ -4,6 +4,7 @@ import { EventWatcher } from "./event-watcher.process.ts";
 import { newNoop } from "@/utils/logger/index.ts";
 
 const CONTRACT = "CAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAFCT4";
+const CONTRACT_B = "CBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBV6";
 
 /**
  * Records the `startLedger` of each getEvents call so a test can assert exactly
@@ -26,13 +27,35 @@ function mockRpc(opts: { oldestLedger: number; latestLedger: number }) {
   return { rpc, startLedgers };
 }
 
+/**
+ * Like `mockRpc` but also records the contractIds in each poll's filter, so a
+ * test can assert exactly which contracts a single watcher covered per poll.
+ */
+function capturingRpc(opts: { oldestLedger: number; latestLedger: number }) {
+  const polledContractIds: string[][] = [];
+  const rpc = {
+    // deno-lint-ignore require-await -- mock satisfies async getHealth contract
+    getHealth: async () => ({ oldestLedger: opts.oldestLedger }),
+    // deno-lint-ignore require-await -- mock satisfies async getEvents contract
+    getEvents: async (
+      req: { startLedger: number; filters: { contractIds?: string[] }[] },
+    ) => {
+      polledContractIds.push(req.filters.flatMap((f) => f.contractIds ?? []));
+      return { events: [], latestLedger: opts.latestLedger };
+    },
+    // deno-lint-ignore require-await -- mock satisfies async getLatestLedger contract
+    getLatestLedger: async () => ({ sequence: opts.latestLedger }),
+  } as unknown as Server;
+  return { rpc, polledContractIds };
+}
+
 Deno.test("EventWatcher - override set → first getEvents at exactly that ledger", async () => {
   const { rpc, startLedgers } = mockRpc({
     oldestLedger: 5000,
     latestLedger: 5100,
   });
   const watcher = new EventWatcher(
-    { contractId: CONTRACT, intervalMs: 60_000 },
+    { contractIds: [CONTRACT], intervalMs: 60_000 },
     { log: newNoop(), rpc, startLedgerBlock: 12345 },
   );
 
@@ -48,7 +71,7 @@ Deno.test("EventWatcher - override unset → first getEvents at oldest available
     latestLedger: 5100,
   });
   const watcher = new EventWatcher(
-    { contractId: CONTRACT, intervalMs: 60_000 },
+    { contractIds: [CONTRACT], intervalMs: 60_000 },
     { log: newNoop(), rpc, startLedgerBlock: null },
   );
 
@@ -65,7 +88,7 @@ Deno.test("EventWatcher - holds no durable cursor: a restart re-syncs from oldes
   // First watcher polls once and advances its in-memory cursor past 5100.
   const first = mockRpc({ oldestLedger: 5000, latestLedger: 5100 });
   const w1 = new EventWatcher(
-    { contractId: CONTRACT, intervalMs: 60_000 },
+    { contractIds: [CONTRACT], intervalMs: 60_000 },
     { log: newNoop(), rpc: first.rpc, startLedgerBlock: null },
   );
   await w1.start();
@@ -77,11 +100,88 @@ Deno.test("EventWatcher - holds no durable cursor: a restart re-syncs from oldes
   // again — nothing was persisted, so it does not resume at 5101.
   const second = mockRpc({ oldestLedger: 5000, latestLedger: 5100 });
   const w2 = new EventWatcher(
-    { contractId: CONTRACT, intervalMs: 60_000 },
+    { contractIds: [CONTRACT], intervalMs: 60_000 },
     { log: newNoop(), rpc: second.rpc, startLedgerBlock: null },
   );
   await w2.start();
   w2.stop();
 
   assertEquals(second.startLedgers[0], 5000);
 });
+
+Deno.test("EventWatcher - getContractIds reflects in-place add/remove", () => {
+  const { rpc } = mockRpc({ oldestLedger: 5000, latestLedger: 5100 });
+  const watcher = new EventWatcher(
+    { contractIds: [CONTRACT], intervalMs: 60_000 },
+    { log: newNoop(), rpc, startLedgerBlock: 1 },
+  );
+
+  assertEquals(watcher.getContractIds(), [CONTRACT]);
+
+  watcher.addContract(CONTRACT_B);
+  assertEquals(watcher.getContractIds().sort(), [CONTRACT, CONTRACT_B].sort());
+
+  watcher.addContract(CONTRACT_B); // idempotent
+  assertEquals(watcher.getContractIds().length, 2);
+
+  watcher.removeContract(CONTRACT);
+  assertEquals(watcher.getContractIds(), [CONTRACT_B]);
+});
+
+Deno.test("EventWatcher - one watcher polls every contract added before start", async () => {
+  const { rpc, polledContractIds } = capturingRpc({
+    oldestLedger: 5000,
+    latestLedger: 5100,
+  });
+  const watcher = new EventWatcher(
+    { contractIds: [CONTRACT], intervalMs: 60_000 },
+    { log: newNoop(), rpc, startLedgerBlock: 1 },
+  );
+  watcher.addContract(CONTRACT_B); // a second council joins before boot completes
+
+  await watcher.start();
+  await tick(); // let the first poll run
+  watcher.stop();
+
+  // A single poll covered BOTH councils' contracts (not one poll per council).
+  assertEquals(polledContractIds.length, 1);
+  assertEquals(polledContractIds[0].sort(), [CONTRACT, CONTRACT_B].sort());
+});
+
+Deno.test("EventWatcher - removed contract is no longer polled", async () => {
+  const { rpc, polledContractIds } = capturingRpc({
+    oldestLedger: 5000,
+    latestLedger: 5100,
+  });
+  const watcher = new EventWatcher(
+    { contractIds: [CONTRACT, CONTRACT_B], intervalMs: 60_000 },
+    { log: newNoop(), rpc, startLedgerBlock: 1 },
+  );
+  watcher.removeContract(CONTRACT); // membership in CONTRACT went inactive
+
+  await watcher.start();
+  await tick();
+  watcher.stop();
+
+  assertEquals(polledContractIds[0], [CONTRACT_B]); // CONTRACT dropped from poll
+});
+
+Deno.test("EventWatcher - empty contract set holds the cursor and skips the RPC", async () => {
+  const { rpc, polledContractIds } = capturingRpc({
+    oldestLedger: 5000,
+    latestLedger: 5100,
+  });
+  const watcher = new EventWatcher(
+    { contractIds: [], intervalMs: 60_000 },
+    { log: newNoop(), rpc, startLedgerBlock: 7000 },
+  );
+
+  await watcher.start();
+  await tick();
+  watcher.stop();
+
+  // No contracts → never query getEvents, and the cursor stays at the boot
+  // ledger so a later join resumes from there.
+  assertEquals(polledContractIds.length, 0);
+  assertEquals(watcher.getLastLedger(), 7000);
+});

src/core/service/event-watcher/event-watcher.process.ts

@@ -1,10 +1,7 @@
 import type { Logger } from "@/utils/logger/index.ts";
 import type { Server } from "stellar-sdk/rpc";
 import { fetchChannelAuthEvents } from "./event-watcher.service.ts";
-import type {
-  ChannelAuthEvent,
-  EventWatcherConfig,
-} from "./event-watcher.types.ts";
+import type { ChannelAuthEvent } from "./event-watcher.types.ts";
 import { withSpan } from "@/core/tracing.ts";
 import { recoverFromOutOfRetention } from "./retention.ts";
 import { resolveBootStartLedger } from "./start-ledger.ts";
@@ -14,7 +11,14 @@ export type ResyncHandler = () => void | Promise<void>;
 
 /**
  * EventWatcher polls Stellar RPC for Channel Auth contract events
- * (provider_added, provider_removed, contract_initialized).
+ * (provider_added, provider_removed, contract_initialized, channel_state_changed).
+ *
+ * A SINGLE watcher covers EVERY active council: it holds a set of channel-auth
+ * contract IDs and polls them all in one batched getEvents call, then dispatches
+ * each event tagged with the contract it came from. Membership changes mutate
+ * this set in place (`addContract`/`removeContract`) — the watcher keeps running
+ * rather than being created/destroyed per council, so boot spins up exactly one
+ * poller regardless of how many councils are active.
  *
  * Uses a self-scheduling pattern (setTimeout after each poll completes)
  * to prevent concurrent polls when RPC is slow.
@@ -32,26 +36,53 @@ export class EventWatcher {
   private timeoutId: number | null = null;
   private isRunning = false;
   private lastLedger: number | null = null;
-  private config: EventWatcherConfig;
+  private contractIds: Set<string>;
+  private intervalMs: number;
   private handlers: EventHandler[] = [];
   private resyncHandlers: ResyncHandler[] = [];
   private rpc: Server;
   private startLedgerBlock: number | null;
   private log: Logger;
 
   constructor(
-    config: { contractId: string; intervalMs?: number },
+    config: { contractIds: string[]; intervalMs?: number },
     deps: { log: Logger; rpc: Server; startLedgerBlock: number | null },
   ) {
-    this.config = {
-      contractId: config.contractId,
-      intervalMs: config.intervalMs ?? 30_000,
-    };
+    this.contractIds = new Set(config.contractIds);
+    this.intervalMs = config.intervalMs ?? 30_000;
     this.rpc = deps.rpc;
     this.startLedgerBlock = deps.startLedgerBlock;
     this.log = deps.log.scope("EventWatcher");
   }
 
+  /**
+   * Add a channel-auth contract to the watched set (e.g. when a PP joins a new
+   * council). Idempotent; the next poll picks it up — no new watcher is spun up.
+   */
+  addContract(contractId: string): void {
+    if (this.contractIds.has(contractId)) return;
+    this.contractIds.add(contractId);
+    this.log.debug("contractId", contractId);
+    this.log.debug("contractCount", this.contractIds.size);
+    this.log.event("added contract to event watcher set");
+  }
+
+  /**
+   * Remove a channel-auth contract from the watched set (e.g. when no active
+   * membership references it anymore). Idempotent; the watcher keeps running.
+   */
+  removeContract(contractId: string): void {
+    if (!this.contractIds.delete(contractId)) return;
+    this.log.debug("contractId", contractId);
+    this.log.debug("contractCount", this.contractIds.size);
+    this.log.event("removed contract from event watcher set");
+  }
+
+  /** The contracts currently covered by this watcher. */
+  getContractIds(): string[] {
+    return Array.from(this.contractIds);
+  }
+
   /**
    * Register a handler that will be called for each new event.
    */
@@ -84,7 +115,7 @@ export class EventWatcher {
       this.rpc,
       this.startLedgerBlock,
     );
-    this.log.debug("contractId", this.config.contractId);
+    this.log.debug("contractCount", this.contractIds.size);
     this.log.debug("startLedger", this.lastLedger);
     this.log.event("EventWatcher initialized boot start ledger");
 
@@ -123,7 +154,7 @@ export class EventWatcher {
     if (this.isRunning) {
       this.timeoutId = setTimeout(
         () => this.scheduleNext(),
-        this.config.intervalMs,
+        this.intervalMs,
       ) as unknown as number;
       this.log.event("next poll scheduled");
     }
@@ -139,7 +170,7 @@ export class EventWatcher {
 
         const { events, latestLedger } = await fetchChannelAuthEvents(
           this.rpc,
-          this.config.contractId,
+          this.getContractIds(),
           this.lastLedger,
           { log: this.log },
         );