UC5: notify removed PP backend on provider removal

deno.json

@@ -1,6 +1,6 @@
 {
   "name": "@moonlight-protocol/council-platform",
-  "version": "0.6.2",
+  "version": "0.6.3",
   "license": "MIT",
   "exports": "./src/main.ts",
   "tasks": {

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

@@ -0,0 +1,138 @@
+import type { Logger } from "@/utils/logger/index.ts";
+import type { CouncilProviderRepository } from "@/persistence/drizzle/repository/council-provider.repository.ts";
+import type { CouncilChannelRepository } from "@/persistence/drizzle/repository/council-channel.repository.ts";
+import { ProviderStatus } from "@/persistence/drizzle/entity/council-provider.entity.ts";
+import { ChannelStatus } from "@/persistence/drizzle/entity/council-channel.entity.ts";
+import type { ChannelAuthEvent } from "./event-watcher.types.ts";
+import {
+  type NotifyProviderRemoved,
+  notifyProviderRemoved,
+} from "./notify-provider-removed.ts";
+
+/**
+ * Apply a single Channel Auth event to the council tables, using the
+ * transaction-bound repositories passed in so the write joins the poll's atomic
+ * commit.
+ *
+ * Kept free of any env-bound imports so it can be unit-tested with fake repos
+ * (importing the watcher's index.ts would pull in `@/config/env.ts`, which
+ * requires DATABASE_URL at module load).
+ */
+export async function applyEvent(
+  councilId: string,
+  event: ChannelAuthEvent,
+  repos: {
+    providerRepo: CouncilProviderRepository;
+    channelRepo: CouncilChannelRepository;
+  },
+  log: Logger,
+  notify: NotifyProviderRemoved = notifyProviderRemoved,
+): Promise<void> {
+  const { providerRepo, channelRepo } = repos;
+  switch (event.type) {
+    case "provider_added": {
+      log.debug("councilId", councilId);
+      log.debug("address", event.address);
+      log.debug("ledger", event.ledger);
+      log.event("provider added on-chain");
+
+      const existing = await providerRepo.findByPublicKey(
+        councilId,
+        event.address,
+      );
+      if (existing) {
+        if (existing.status === ProviderStatus.REMOVED) {
+          await providerRepo.update(existing.id, {
+            status: ProviderStatus.ACTIVE,
+            registeredByEvent: `ledger:${event.ledger}`,
+            removedByEvent: null,
+          });
+          log.event("provider re-activated");
+        }
+      } else {
+        await providerRepo.create({
+          id: crypto.randomUUID(),
+          councilId,
+          publicKey: event.address,
+          status: ProviderStatus.ACTIVE,
+          registeredByEvent: `ledger:${event.ledger}`,
+          createdAt: new Date(),
+          updatedAt: new Date(),
+        });
+        log.event("provider registered");
+      }
+      break;
+    }
+
+    case "provider_removed": {
+      log.debug("councilId", councilId);
+      log.debug("address", event.address);
+      log.debug("ledger", event.ledger);
+      log.event("provider removed on-chain");
+
+      const provider = await providerRepo.findByPublicKey(
+        councilId,
+        event.address,
+      );
+      if (provider) {
+        // Only the ACTIVE→REMOVED transition is a real removal. Re-delivery of
+        // the same event, or a boot replay where the row is already REMOVED,
+        // is a no-op for the notify so the PP backend isn't pinged twice.
+        const wasActive = provider.status === ProviderStatus.ACTIVE;
+        await providerRepo.update(provider.id, {
+          status: ProviderStatus.REMOVED,
+          removedByEvent: `ledger:${event.ledger}`,
+        });
+        log.event("provider marked as removed");
+
+        if (wasActive && provider.providerUrl) {
+          // Live signal to the removed PP's backend. Fire-and-forget (not
+          // awaited) so it can't stall or roll back this poll's atomic commit;
+          // the PP's own event-watcher remains the can't-miss path.
+          notify(
+            provider.providerUrl,
+            { councilId, publicKey: event.address, ledger: event.ledger },
+            { log },
+          );
+        }
+      }
+      break;
+    }
+
+    case "channel_state_changed": {
+      // SOLE authoritative writer of channel status. The DB only ever reflects
+      // CONFIRMED on-chain state — never written ahead of the chain. Any
+      // optimistic pendingAction marker is cleared here on confirmation.
+      const channel = event.channel ?? event.address;
+      const status = event.enabled
+        ? ChannelStatus.ENABLED
+        : ChannelStatus.DISABLED;
+      log.debug("councilId", councilId);
+      log.debug("channel", channel);
+      log.debug("asset", event.asset ?? "");
+      log.debug("ledger", event.ledger);
+      log.debug("status", status);
+      log.event("channel state changed on-chain");
+
+      const updated = await channelRepo.setStatusByContractId(
+        councilId,
+        channel,
+        status,
+      );
+      if (updated) {
+        log.event("channel status reconciled from chain");
+      } else {
+        log.event("channel state event for unknown channel (ignored)");
+      }
+      break;
+    }
+
+    case "contract_initialized": {
+      log.debug("councilId", councilId);
+      log.debug("address", event.address);
+      log.debug("ledger", event.ledger);
+      log.event("channel auth contract initialized");
+      break;
+    }
+  }
+}

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

@@ -11,9 +11,7 @@ import { CouncilMetadataRepository } from "@/persistence/drizzle/repository/coun
 import { CouncilProviderRepository } from "@/persistence/drizzle/repository/council-provider.repository.ts";
 import { CouncilChannelRepository } from "@/persistence/drizzle/repository/council-channel.repository.ts";
 import { WatcherCursorRepository } from "@/persistence/drizzle/repository/watcher-cursor.repository.ts";
-import { ProviderStatus } from "@/persistence/drizzle/entity/council-provider.entity.ts";
-import { ChannelStatus } from "@/persistence/drizzle/entity/council-channel.entity.ts";
-import type { ChannelAuthEvent } from "./event-watcher.types.ts";
+import { applyEvent } from "./apply-event.ts";
 
 // Wire env CHALLENGE_TTL (seconds) to council auth (ms)
 setChallengeTtlMs(CHALLENGE_TTL * 1000);
@@ -43,114 +41,6 @@ const DB_SYNC_INTERVAL_MS = 5_000;
 const activeWatchers = new Map<string, EventWatcher>(); // councilId → watcher
 let dbSyncTimer: number | null = null;
 
-/**
- * Apply a single Channel Auth event to the council tables, using the
- * transaction-bound repositories passed in so the write joins the poll's atomic
- * commit.
- */
-async function applyEvent(
-  councilId: string,
-  event: ChannelAuthEvent,
-  repos: {
-    providerRepo: CouncilProviderRepository;
-    channelRepo: CouncilChannelRepository;
-  },
-  log: Logger,
-): Promise<void> {
-  const { providerRepo, channelRepo } = repos;
-  switch (event.type) {
-    case "provider_added": {
-      log.debug("councilId", councilId);
-      log.debug("address", event.address);
-      log.debug("ledger", event.ledger);
-      log.event("provider added on-chain");
-
-      const existing = await providerRepo.findByPublicKey(
-        councilId,
-        event.address,
-      );
-      if (existing) {
-        if (existing.status === ProviderStatus.REMOVED) {
-          await providerRepo.update(existing.id, {
-            status: ProviderStatus.ACTIVE,
-            registeredByEvent: `ledger:${event.ledger}`,
-            removedByEvent: null,
-          });
-          log.event("provider re-activated");
-        }
-      } else {
-        await providerRepo.create({
-          id: crypto.randomUUID(),
-          councilId,
-          publicKey: event.address,
-          status: ProviderStatus.ACTIVE,
-          registeredByEvent: `ledger:${event.ledger}`,
-          createdAt: new Date(),
-          updatedAt: new Date(),
-        });
-        log.event("provider registered");
-      }
-      break;
-    }
-
-    case "provider_removed": {
-      log.debug("councilId", councilId);
-      log.debug("address", event.address);
-      log.debug("ledger", event.ledger);
-      log.event("provider removed on-chain");
-
-      const provider = await providerRepo.findByPublicKey(
-        councilId,
-        event.address,
-      );
-      if (provider) {
-        await providerRepo.update(provider.id, {
-          status: ProviderStatus.REMOVED,
-          removedByEvent: `ledger:${event.ledger}`,
-        });
-        log.event("provider marked as removed");
-      }
-      break;
-    }
-
-    case "channel_state_changed": {
-      // SOLE authoritative writer of channel status. The DB only ever reflects
-      // CONFIRMED on-chain state — never written ahead of the chain. Any
-      // optimistic pendingAction marker is cleared here on confirmation.
-      const channel = event.channel ?? event.address;
-      const status = event.enabled
-        ? ChannelStatus.ENABLED
-        : ChannelStatus.DISABLED;
-      log.debug("councilId", councilId);
-      log.debug("channel", channel);
-      log.debug("asset", event.asset ?? "");
-      log.debug("ledger", event.ledger);
-      log.debug("status", status);
-      log.event("channel state changed on-chain");
-
-      const updated = await channelRepo.setStatusByContractId(
-        councilId,
-        channel,
-        status,
-      );
-      if (updated) {
-        log.event("channel status reconciled from chain");
-      } else {
-        log.event("channel state event for unknown channel (ignored)");
-      }
-      break;
-    }
-
-    case "contract_initialized": {
-      log.debug("councilId", councilId);
-      log.debug("address", event.address);
-      log.debug("ledger", event.ledger);
-      log.event("channel auth contract initialized");
-      break;
-    }
-  }
-}
-
 /**
  * Build the atomic commit for a council's watcher: apply every event of a poll
  * and advance the cursor in ONE transaction. A failure rolls back both the

src/core/service/event-watcher/notify-on-removal.test.ts

@@ -0,0 +1,106 @@
+import { assertEquals } from "@std/assert";
+import { newNoop } from "@/utils/logger/index.ts";
+import { applyEvent } from "./apply-event.ts";
+import type { ChannelAuthEvent } from "./event-watcher.types.ts";
+import type { ProviderRemovedNotice } from "./notify-provider-removed.ts";
+import {
+  type CouncilProvider,
+  ProviderStatus,
+} from "@/persistence/drizzle/entity/council-provider.entity.ts";
+import type { CouncilProviderRepository } from "@/persistence/drizzle/repository/council-provider.repository.ts";
+import type { CouncilChannelRepository } from "@/persistence/drizzle/repository/council-channel.repository.ts";
+
+const COUNCIL = "CCOUNCIL";
+const PROVIDER_PK = "GPROVIDER";
+
+function removedEvent(ledger = 42): ChannelAuthEvent {
+  return {
+    type: "provider_removed",
+    address: PROVIDER_PK,
+    ledger,
+    contractId: COUNCIL,
+  };
+}
+
+/** Minimal provider row; only fields applyEvent touches need to be real. */
+function providerRow(over: Partial<CouncilProvider>): CouncilProvider {
+  return {
+    id: "row-1",
+    councilId: COUNCIL,
+    publicKey: PROVIDER_PK,
+    status: ProviderStatus.ACTIVE,
+    providerUrl: "https://pp.example.com",
+    ...over,
+  } as CouncilProvider;
+}
+
+/** Fake repo capturing update() calls and returning a fixed provider row. */
+function fakeRepos(provider: CouncilProvider | undefined) {
+  const updates: Array<{ id: string; fields: Record<string, unknown> }> = [];
+  const providerRepo = {
+    findByPublicKey: (_c: string, _pk: string) => Promise.resolve(provider),
+    update: (id: string, fields: Record<string, unknown>) => {
+      updates.push({ id, fields });
+      return Promise.resolve(undefined);
+    },
+  } as unknown as CouncilProviderRepository;
+  const channelRepo = {} as unknown as CouncilChannelRepository;
+  return { repos: { providerRepo, channelRepo }, updates };
+}
+
+function spyNotify() {
+  const calls: Array<{ url: string; notice: ProviderRemovedNotice }> = [];
+  const notify = (url: string, notice: ProviderRemovedNotice) => {
+    calls.push({ url, notice });
+  };
+  return { notify, calls };
+}
+
+Deno.test("active provider removal marks REMOVED and notifies its backend once", async () => {
+  const { repos, updates } = fakeRepos(providerRow({}));
+  const { notify, calls } = spyNotify();
+
+  await applyEvent(COUNCIL, removedEvent(7), repos, newNoop(), notify);
+
+  assertEquals(updates.length, 1);
+  assertEquals(updates[0].fields.status, ProviderStatus.REMOVED);
+  assertEquals(calls.length, 1);
+  assertEquals(calls[0].url, "https://pp.example.com");
+  assertEquals(calls[0].notice, {
+    councilId: COUNCIL,
+    publicKey: PROVIDER_PK,
+    ledger: 7,
+  });
+});
+
+Deno.test("re-delivered/boot-replayed removal of an already-REMOVED provider does not re-notify", async () => {
+  const { repos, updates } = fakeRepos(
+    providerRow({ status: ProviderStatus.REMOVED }),
+  );
+  const { notify, calls } = spyNotify();
+
+  await applyEvent(COUNCIL, removedEvent(), repos, newNoop(), notify);
+
+  // Idempotent: the write still runs, but no second notify is sent.
+  assertEquals(updates.length, 1);
+  assertEquals(calls.length, 0);
+});
+
+Deno.test("provider with no provider_url is not notified (dormant)", async () => {
+  const { repos } = fakeRepos(providerRow({ providerUrl: null }));
+  const { notify, calls } = spyNotify();
+
+  await applyEvent(COUNCIL, removedEvent(), repos, newNoop(), notify);
+
+  assertEquals(calls.length, 0);
+});
+
+Deno.test("removal of an unknown provider is a no-op", async () => {
+  const { repos, updates } = fakeRepos(undefined);
+  const { notify, calls } = spyNotify();
+
+  await applyEvent(COUNCIL, removedEvent(), repos, newNoop(), notify);
+
+  assertEquals(updates.length, 0);
+  assertEquals(calls.length, 0);
+});