refactor(event-watcher): drop Deno.KV state store; boot syncs from oldest available

deno.json

@@ -1,10 +1,10 @@
 {
   "name": "@moonlight-protocol/provider-platform",
-  "version": "0.8.0",
+  "version": "0.9.0",
   "license": "MIT",
   "exports": "./src/main.ts",
   "tasks": {
-    "serve": "deno run --allow-all --unstable-kv src/main.ts",
+    "serve": "deno run --allow-all src/main.ts",
     "test": "deno test --allow-all --no-check --ignore=src/http/v1/pay/tests,src/http/v1/entities/tests src/ && deno test --allow-all --no-check --config src/http/v1/pay/tests/deno.json src/http/v1/pay/tests/ && deno test --allow-all --no-check --config src/http/v1/entities/tests/deno.json src/http/v1/entities/tests/",
     "test:integration": "deno test --allow-all --no-check --config tests/deno.json tests/",
     "test:unit": "deno test --allow-all --no-check --ignore=src/http/v1/pay/tests,src/http/v1/entities/tests src/",

src/config/env.ts

@@ -121,3 +121,20 @@ export const TRANSACTION_EXPIRATION_OFFSET = _rawTxExpirationOffset;
 export const EVENT_WATCHER_INTERVAL_MS = Number(
   Deno.env.get("EVENT_WATCHER_INTERVAL_MS") ?? "30000",
 );
+
+// Where a fresh watcher (no stored position) begins polling. Unset → SYNC ALL
+// AVAILABLE: start from the oldest ledger the RPC still retains, so no in-window
+// event is skipped on a cold boot. Set → start at exactly that ledger (e.g. to
+// skip ancient history on a known-fresh deploy). Never defaults to "latest".
+const _rawBootSyncStart = loadOptionalEnv("BOOT_SYNC_START_LEDGER_BLOCK");
+let _bootSyncStart: number | null = null;
+if (_rawBootSyncStart !== undefined && _rawBootSyncStart !== "") {
+  const n = Number(_rawBootSyncStart);
+  if (!Number.isInteger(n) || n < 0) {
+    throw new Error(
+      `BOOT_SYNC_START_LEDGER_BLOCK must be a non-negative integer, got: "${_rawBootSyncStart}"`,
+    );
+  }
+  _bootSyncStart = n;
+}
+export const BOOT_SYNC_START_LEDGER_BLOCK = _bootSyncStart;

src/core/service/auth/sessions/in-memory-session-manager.ts

@@ -1,15 +1,24 @@
 import { SESSION_TTL } from "@/config/env.ts";
-import { memDb } from "@/persistence/kv/config.ts";
 import type { Session } from "@/models/auth/session/session.model.ts";
 import type { Logger } from "@/utils/logger/index.ts";
 
+/**
+ * Live store of in-flight handshake sessions, keyed by challenge tx hash.
+ *
+ * Truly in-memory (a plain Map) — these are short-TTL auth handshakes, not
+ * derived state, so there is no benefit to persisting them. A machine
+ * stop/redeploy clears them and the client simply re-authenticates. Nothing
+ * here touches disk or Deno.KV.
+ */
 export class InMemorySessionManager {
+  private sessions: Map<string, Session> = new Map();
   private log: Logger;
 
   constructor(deps: { log: Logger }) {
     this.log = deps.log.scope("InMemorySessionManager");
   }
 
+  // deno-lint-ignore require-await -- async to preserve the manager's contract
   public async addSession(
     txHash: string,
     clientAccount: string,
@@ -19,31 +28,30 @@ export class InMemorySessionManager {
     this.log.info("addSession");
     this.log.debug("txHash", txHash);
 
-    const cr = await memDb.sessions.add({
+    this.sessions.set(txHash, {
       txHash,
       clientAccount,
       requestId,
       expiresAt,
       status: "PENDING",
     });
 
-    if (cr.ok) {
-      this.log.event("session added to store");
-    }
+    this.log.event("session added to store");
   }
 
+  // deno-lint-ignore require-await -- async to preserve the manager's contract
   public async getSession(txHash: string): Promise<Session | undefined> {
     this.log.info("getSession");
     this.log.debug("txHash", txHash);
 
     this.log.event("looking up session by txHash");
-    const session = await memDb.sessions.findByPrimaryIndex("txHash", txHash);
-    if (session && Date.now() < session.value.expiresAt.getTime()) {
+    const session = this.sessions.get(txHash);
+    if (session && Date.now() < session.expiresAt.getTime()) {
       this.log.event("session found and valid");
-      return session.value;
+      return session;
     }
     this.log.event("session missing or expired, deleting");
-    await memDb.sessions.delete(txHash);
+    this.sessions.delete(txHash);
     return undefined;
   }
 
@@ -60,17 +68,20 @@ export class InMemorySessionManager {
     }
 
     this.log.event("persisting session update");
-    await memDb.sessions.update(session.txHash, session);
+    this.sessions.set(session.txHash, session);
   }
 
+  // deno-lint-ignore require-await -- async to preserve the manager's contract
   public async cleanupExpired(): Promise<void> {
     const now = Date.now();
 
     this.log.info("cleanupExpired");
 
-    await memDb.sessions.deleteMany({
-      filter: (doc) => doc.value.expiresAt.getTime() < now,
-    });
+    for (const [txHash, session] of this.sessions) {
+      if (session.expiresAt.getTime() < now) {
+        this.sessions.delete(txHash);
+      }
+    }
 
     this.log.event("expired sessions cleaned");
   }

src/core/service/event-watcher/channel-registry.ts

@@ -24,66 +24,28 @@ export interface ChannelRecord {
   removedAtLedger?: number;
 }
 
-const REGISTRY_KV_KEY = ["channel-registry", "channels"];
-const KV_DIR = new URL("../../../../.data", import.meta.url).pathname;
-const KV_PATH = `${KV_DIR}/memory-kvdb.db`;
-
 /**
  * In-memory registry of channels this PP is registered in.
  *
  * Populated by EventWatcher events, read by dashboard API endpoints.
  * The `configuredChannels` set represents which channels the operator
  * has chosen to actively serve (from instance config).
  *
- * State is persisted to Deno KV so restarts don't lose channel info.
- * On startup, the registry restores from KV first, then seeds any
- * configured channels that aren't already tracked.
+ * State is NOT persisted: every record it holds is authoritatively
+ * re-queryable (channel membership from this PP's own DB, asset-channel status
+ * from the council via converge-by-query on boot), so the registry is rebuilt
+ * from scratch on each start rather than restored from a durable store.
  */
 export class ChannelRegistry {
   private channels: Map<string, ChannelRecord> = new Map();
   private configuredChannels: Set<string>;
-  private kv: Deno.Kv | null = null;
   private log: Logger;
 
   constructor(configuredChannelIds: string[], deps: { log: Logger }) {
     this.configuredChannels = new Set(configuredChannelIds);
     this.log = deps.log.scope("ChannelRegistry");
   }
 
-  /**
-   * Initialize the registry: restore from KV, then seed any configured
-   * channels not already present. Must be called before use.
-   */
-  async initialize(): Promise<void> {
-    await Deno.mkdir(KV_DIR, { recursive: true });
-    this.kv = await Deno.openKv(KV_PATH);
-
-    // Restore persisted state
-    const stored = await this.kv.get<ChannelRecord[]>(REGISTRY_KV_KEY);
-    if (stored.value && stored.value.length > 0) {
-      for (const record of stored.value) {
-        this.channels.set(record.contractId, record);
-      }
-      this.log.debug("count", stored.value.length);
-      this.log.event("ChannelRegistry restored from KV");
-    }
-
-    // Seed configured channels that aren't already tracked
-    for (const contractId of this.configuredChannels) {
-      if (!this.channels.has(contractId)) {
-        this.channels.set(contractId, {
-          contractId,
-          state: "active",
-          registeredAtLedger: 0,
-        });
-        this.log.debug("contractId", contractId);
-        this.log.event("seeded configured channel as active");
-      }
-    }
-
-    await this.persist();
-  }
-
   /** Dynamically add a channel to track (e.g., when a PP joins a new council). */
   addChannel(contractId: string): void {
     this.log.info("addChannel");
@@ -137,6 +99,7 @@ export class ChannelRegistry {
    * `enabled=true` → `active` (full service resumes). Idempotent: re-applying
    * the same state is a no-op.
    */
+  // deno-lint-ignore require-await -- async to preserve the awaited mutation contract
   async applyChannelState(
     channelContractId: string,
     enabled: boolean,
@@ -160,7 +123,6 @@ export class ChannelRegistry {
     this.log.event(
       enabled ? "asset channel enabled" : "asset channel disabled",
     );
-    await this.persist();
   }
 
   /**
@@ -172,6 +134,7 @@ export class ChannelRegistry {
     return this.channels.get(channelContractId)?.state === "disabled";
   }
 
+  // deno-lint-ignore require-await -- async to preserve the awaited mutation contract
   private async onProviderAdded(event: ChannelAuthEvent): Promise<void> {
     const isConfigured = this.configuredChannels.has(event.contractId);
     const state: ChannelState = isConfigured ? "active" : "pending";
@@ -186,10 +149,9 @@ export class ChannelRegistry {
     this.log.debug("state", state);
     this.log.debug("ledger", event.ledger);
     this.log.event("provider registered in channel");
-
-    await this.persist();
   }
 
+  // deno-lint-ignore require-await -- async to preserve the awaited mutation contract
   private async onProviderRemoved(event: ChannelAuthEvent): Promise<void> {
     const existing = this.channels.get(event.contractId);
     if (existing) {
@@ -207,20 +169,6 @@ export class ChannelRegistry {
     this.log.debug("contractId", event.contractId);
     this.log.debug("ledger", event.ledger);
     this.log.event("provider removed from channel");
-
-    await this.persist();
-  }
-
-  /**
-   * Persist current registry state to KV.
-   */
-  private async persist(): Promise<void> {
-    if (!this.kv) return;
-    try {
-      await this.kv.set(REGISTRY_KV_KEY, this.getAll());
-    } catch (error) {
-      this.log.error(error, "failed to persist channel registry");
-    }
   }
 
   /**
@@ -247,6 +195,7 @@ export class ChannelRegistry {
   /**
    * Mark a channel as configured (operator activated it).
    */
+  // deno-lint-ignore require-await -- async to preserve the awaited mutation contract
   async activateChannel(contractId: string): Promise<void> {
     this.log.info("activateChannel");
     this.log.debug("contractId", contractId);
@@ -255,13 +204,13 @@ export class ChannelRegistry {
     if (channel && channel.state === "pending") {
       channel.state = "active";
       this.log.event("channel transitioned to active");
-      await this.persist();
     }
   }
 
   /**
    * Mark a channel as no longer configured by the operator.
    */
+  // deno-lint-ignore require-await -- async to preserve the awaited mutation contract
   async deactivateChannel(contractId: string): Promise<void> {
     this.log.info("deactivateChannel");
     this.log.debug("contractId", contractId);
@@ -270,19 +219,6 @@ export class ChannelRegistry {
     if (channel && channel.state === "active") {
       channel.state = "pending";
       this.log.event("channel transitioned to pending");
-      await this.persist();
-    }
-  }
-
-  /**
-   * Close the KV handle. Call on shutdown.
-   */
-  close(): void {
-    this.log.info("close");
-    if (this.kv) {
-      this.kv.close();
-      this.kv = null;
-      this.log.event("KV handle closed");
     }
   }
 }

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

@@ -0,0 +1,87 @@
+import { assertEquals } from "@std/assert";
+import type { Server } from "stellar-sdk/rpc";
+import { EventWatcher } from "./event-watcher.process.ts";
+import { newNoop } from "@/utils/logger/index.ts";
+
+const CONTRACT = "CAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAFCT4";
+
+/**
+ * Records the `startLedger` of each getEvents call so a test can assert exactly
+ * where a fresh watcher began polling. `latestLedger` controls how far the
+ * in-memory cursor advances after a poll.
+ */
+function mockRpc(opts: { oldestLedger: number; latestLedger: number }) {
+  const startLedgers: number[] = [];
+  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 }) => {
+      startLedgers.push(req.startLedger);
+      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, startLedgers };
+}
+
+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 },
+    { log: newNoop(), rpc, startLedgerBlock: 12345 },
+  );
+
+  await watcher.start();
+  watcher.stop();
+
+  assertEquals(startLedgers[0], 12345);
+});
+
+Deno.test("EventWatcher - override unset → first getEvents at oldest available", async () => {
+  const { rpc, startLedgers } = mockRpc({
+    oldestLedger: 5000,
+    latestLedger: 5100,
+  });
+  const watcher = new EventWatcher(
+    { contractId: CONTRACT, intervalMs: 60_000 },
+    { log: newNoop(), rpc, startLedgerBlock: null },
+  );
+
+  await watcher.start();
+  watcher.stop();
+
+  assertEquals(startLedgers[0], 5000);
+});
+
+// Let the fire-and-forget first poll (scheduled by start()) run to completion.
+const tick = () => new Promise<void>((resolve) => setTimeout(resolve, 0));
+
+Deno.test("EventWatcher - holds no durable cursor: a restart re-syncs from oldest", async () => {
+  // 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 },
+    { log: newNoop(), rpc: first.rpc, startLedgerBlock: null },
+  );
+  await w1.start();
+  await tick(); // wait for the first poll to advance the in-memory cursor
+  w1.stop();
+  assertEquals(w1.getLastLedger(), 5101); // advanced in memory
+
+  // A fresh watcher (simulating a process restart) must start from oldest
+  // 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 },
+    { log: newNoop(), rpc: second.rpc, startLedgerBlock: null },
+  );
+  await w2.start();
+  w2.stop();
+
+  assertEquals(second.startLedgers[0], 5000);
+});