diff --git a/deno.json b/deno.json index c6eb2b4..11760b9 100644 --- a/deno.json +++ b/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": { diff --git a/src/core/service/event-watcher/apply-event.ts b/src/core/service/event-watcher/apply-event.ts new file mode 100644 index 0000000..a021a61 --- /dev/null +++ b/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 { + 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; + } + } +} diff --git a/src/core/service/event-watcher/index.ts b/src/core/service/event-watcher/index.ts index da97d9b..6c5faa9 100644 --- a/src/core/service/event-watcher/index.ts +++ b/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(); // 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 { - 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 diff --git a/src/core/service/event-watcher/notify-on-removal.test.ts b/src/core/service/event-watcher/notify-on-removal.test.ts new file mode 100644 index 0000000..e64eac9 --- /dev/null +++ b/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 { + 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 }> = []; + const providerRepo = { + findByPublicKey: (_c: string, _pk: string) => Promise.resolve(provider), + update: (id: string, fields: Record) => { + 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); +}); diff --git a/src/core/service/event-watcher/notify-provider-removed.ts b/src/core/service/event-watcher/notify-provider-removed.ts new file mode 100644 index 0000000..bd03cd3 --- /dev/null +++ b/src/core/service/event-watcher/notify-provider-removed.ts @@ -0,0 +1,57 @@ +import type { Logger } from "@/utils/logger/index.ts"; + +/** Notice payload sent to a removed PP's backend. */ +export interface ProviderRemovedNotice { + /** Council id == the channel-auth contract id that emitted the removal. */ + councilId: string; + /** Public key of the provider that was removed. */ + publicKey: string; + /** Ledger the removal event was observed at. */ + ledger: number; +} + +export type NotifyProviderRemoved = ( + providerUrl: string, + notice: ProviderRemovedNotice, + deps: { log: Logger }, +) => void; + +/** + * Fire-and-forget POST telling a removed PP's backend to re-check its + * membership with the council. + * + * This is a LOW-TRUST live signal, not an authority: the body is a hint that + * "something changed for you, re-query now". The PP converges by calling the + * council's authoritative `/public/provider/membership-status` endpoint, so the + * notice needs no shared secret and a spurious or replayed POST can only make a + * provider re-confirm its own status against the council. + * + * It is NEVER awaited so it cannot stall — or roll back — the watcher's atomic + * poll commit (the same fire-and-forget shape as {@link notifyDiscord}). It is + * dormant when the provider row carries no `provider_url`; the caller skips it. + */ +export const notifyProviderRemoved: NotifyProviderRemoved = ( + providerUrl, + notice, + deps, +) => { + const log = deps.log.scope("notifyProviderRemoved"); + const url = `${providerUrl.replace(/\/+$/, "")}/api/v1/council/removed`; + + fetch(url, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify(notice), + }) + .then((res) => { + if (!res.ok) { + log.error( + new Error(`provider responded ${res.status}`), + "provider removal notify non-2xx", + ); + } else { + log.event("provider removal notify delivered"); + } + }) + .catch((err) => log.error(err, "provider removal notify failed")); +};