Watch SAC events by topic filter with cursor-paged drains; restore periodic topology re-sync

README.md

@@ -10,7 +10,7 @@ Streams a live view of the Moonlight network over WebSocket. Cold-start
 self-syncs from `council-platform`'s public endpoints and Soroban RPC; no
 persistence.
 
-Six event kinds drive the dashboard (see
+Seven event kinds drive the dashboard (see
 [design sketch](../pm-theahaco/network-dashboard-design-sketch.md)):
 
 | Kind                 | Chain source                                                          |
@@ -21,27 +21,48 @@ Six event kinds drive the dashboard (see
 | `asset_registered`   | New `assetContractId` appearing in `council-platform/public/channels` |
 | `channel_deposit`    | SAC `transfer` event TO a known channel address                       |
 | `channel_settlement` | SAC `transfer` event FROM a known channel address                     |
+| `channel_bundle`     | SAC `fee` event whose payer is a registered PP                        |
 
 ## Endpoints
 
 | Path                     | Description                                                    |
 | ------------------------ | -------------------------------------------------------------- |
 | `GET /api/v1/health`     | Liveness — `{status,service,version}`                          |
-| `GET /api/v1/network/ws` | Public WebSocket. Subprotocol `moonlight.network.v1`. No auth. |
+| `GET /api/v1/network/ws` | Public WebSocket. Subprotocol `moonlight.network.v2`. No auth. |
 
 ### WebSocket frame protocol
 
 Server → client, JSON-encoded:
 
 ```jsonc
-// Sent once on open.
+// Sent once on open. (v2 additions over v1: throughputPerMin + latencyMs
+// counters, sparklines, assetBreakdown, councilRolling.)
 {
   "type": "snapshot",
   "counters": {
     "councils": 4,
     "activePPs": 7,
     "eventsLast24h": 31,
-    "assetsRegistered": 2
+    "assetsRegistered": 2,
+    "throughputPerMin": 3,
+    "latencyMs": 4200
+  },
+  "sparklines": {
+    "throughput": [/* 60 per-minute counts, oldest first */],
+    "latency": [/* 60 per-minute avg ms (null where no samples) */],
+    "volume": [/* 60 per-minute deposit+settlement totals (whole units) */]
+  },
+  "assetBreakdown": [
+    { "assetContractId": "CDMLF…", "assetCode": "XLM", "amountStroops": "12000000", "percent": 100 }
+  ],
+  "councilRolling": {
+    "CBPHGAJ4F7…": {
+      "bundlesLastHour": 2,
+      "eventsLastHour": 5,
+      "ratePerMin": 0.1,
+      "depositVolumeStroops": "200500000",
+      "settlementVolumeStroops": "100000000"
+    }
   },
   "topology": [
     {
@@ -71,8 +92,8 @@ Server → client, JSON-encoded:
 }
 ```
 
-No client → server frames in v1. Clients reconnect rather than ping; on
-reconnect they receive a fresh snapshot.
+No client → server frames. Clients reconnect rather than ping; on reconnect they
+receive a fresh snapshot.
 
 ## Architecture
 
@@ -81,15 +102,28 @@ council-platform /public/* ─┐
                             ├─►  in-memory state ─►  WS clients
 Soroban /getEvents (poll) ──┘         ▲
                                        │
-        hourly re-sync + minute window sweep
+        minute topology re-sync + minute window sweep
 ```
 
 - **Cold start**: fetch `council-platform/api/v1/public/councils` (one call
   carries councils + channels + providers + jurisdictions), walk trailing 24h on
-  every watched contract via `rpc.getEvents`, seed the rolling counter window +
-  activity-feed ring buffer.
-- **Forward poll**: 5s cursor-based poll on the same contractId set.
-- **Hourly re-sync**: refresh topology + re-anchor the rolling counter window.
+  the subscription set via `rpc.getEvents`, seed the rolling counter window and
+  the activity-feed ring buffer.
+- **Subscriptions**: Channel Auth (council) contracts are watched by contractId.
+  SAC contracts are watched by TOPIC only — `transfer` patterns pinned to known
+  channel addresses (deposit/settlement) and `fee` patterns pinned to registered
+  PP keys (bundles). The mainnet XLM SAC emits hundreds of events per ledger; an
+  unfiltered contractId subscription drowns in them.
+- **Forward poll**: 5s poll over the subscription set, paged via the RPC event
+  cursor so consumption is gap-free even mid-ledger; the shared poll position
+  only advances across fully-consumed pages (re-read overlap is deduped by event
+  id).
+- **New-council discovery**: network-wide `contract_initialized` poll feeds the
+  adoption pipeline (topology refresh + historical back-fill per adopted
+  council).
+- **Minute topology re-sync**: refresh topology from council-platform — the
+  backstop for DB-only registrations (channels, jurisdictions, labels), which
+  emit no chain event.
 - **Minute sweep**: drop window entries older than 24h.
 
 ## Running locally

deno.json

@@ -1,6 +1,6 @@
 {
   "name": "@moonlight-protocol/network-dashboard-platform",
-  "version": "0.1.15",
+  "version": "0.1.16",
   "license": "MIT",
   "exports": "./src/main.ts",
   "tasks": {

src/core/state/store.ts

@@ -82,6 +82,15 @@ export class NetworkStateStore {
   private providerToCouncil = new Map<string, string>();
   /** Ring buffer of most-recent events for the activity feed. */
   private recent: NetworkEvent[] = [];
+  /**
+   * Ids of every event currently tracked (ring buffer + 24h window).
+   * The watcher's cursor-based drains re-read a small overlap by design
+   * (the shared poll position advances to the LOWEST drain point across
+   * calls), so dedup must outlive the 20-entry ring — otherwise a
+   * re-read event past the ring would double-count metrics and re-emit
+   * to WS clients. Pruned by the minute sweep alongside `metrics`.
+   */
+  private seenIds = new Set<string>();
   /**
    * Sliding 24h metric records (one per event observed within the window).
    * Newest at the end; the cold-start scan seeds in chronological order.
@@ -168,6 +177,10 @@ export class NetworkStateStore {
     return this.providerToCouncil.get(publicKey);
   }
 
+  getProviderPublicKeys(): string[] {
+    return Array.from(this.providerToCouncil.keys());
+  }
+
   /**
    * Surgical, idempotent insert into the providerToCouncil map. Called by
    * the watcher when a `provider_added` chain event is observed, so a PP
@@ -210,9 +223,10 @@ export class NetworkStateStore {
    * close time; pass null when the watcher couldn't determine it.
    */
   recordEvent(event: NetworkEvent, latencyMs: number | null = null): boolean {
-    if (this.recent.some((e) => e.id === event.id)) {
+    if (this.seenIds.has(event.id)) {
       return false;
     }
+    this.seenIds.add(event.id);
     this.recent.unshift(event);
     if (this.recent.length > RING_BUFFER_SIZE) {
       this.recent.length = RING_BUFFER_SIZE;
@@ -234,12 +248,24 @@ export class NetworkStateStore {
     return true;
   }
 
-  /** Drop metric records older than 24h. Called by the minute-sweep. */
+  /**
+   * Drop metric records older than 24h. Called by the minute-sweep.
+   * `seenIds` is rebuilt from the survivors (+ the ring buffer) so the
+   * dedup set stays bounded; a pruned id cannot legitimately reappear —
+   * the watcher's poll position only moves forward.
+   */
   sweepWindow(now: number = Date.now()): number {
     const cutoff = now - ROLLING_WINDOW_MS;
     const before = this.metrics.length;
     this.metrics = this.metrics.filter((m) => m.occurredAt >= cutoff);
-    return before - this.metrics.length;
+    const purged = before - this.metrics.length;
+    if (purged > 0) {
+      this.seenIds = new Set([
+        ...this.metrics.map((m) => m.id),
+        ...this.recent.map((e) => e.id),
+      ]);
+    }
+    return purged;
   }
 
   /**
@@ -259,6 +285,7 @@ export class NetworkStateStore {
       if (existing.has(e.id)) continue;
       const ts = Date.parse(e.occurredAt);
       if (Number.isNaN(ts)) continue;
+      this.seenIds.add(e.id);
       const { assetContractId, amountStroops } = readPayload(e);
       this.metrics.push({
         id: e.id,
@@ -275,6 +302,7 @@ export class NetworkStateStore {
   /** Seed the ring buffer at cold start (newest entries first). */
   seedRecent(events: NetworkEvent[]): void {
     this.recent = events.slice(0, RING_BUFFER_SIZE);
+    for (const e of this.recent) this.seenIds.add(e.id);
   }
 
   recentEvents(): NetworkEvent[] {
@@ -467,6 +495,7 @@ export class NetworkStateStore {
     this.providerToCouncil.clear();
     this.recent = [];
     this.metrics = [];
+    this.seenIds.clear();
   }
 }
 

src/core/state/store_test.ts

@@ -305,3 +305,36 @@ Deno.test("councilRollingMetrics returns zero rows for known but quiet councils"
   assertEquals(rolling["CB"].bundlesLastHour, 0);
   assertEquals(rolling["CB"].eventsLastHour, 0);
 });
+
+Deno.test("recordEvent dedups by id past the ring buffer (cursor re-reads)", () => {
+  const s = makeStore();
+  // Overflow the 20-entry ring so the first event is long gone from it.
+  for (let i = 0; i < RING_BUFFER_CAPACITY + 5; i++) {
+    assertEquals(s.recordEvent(event(`e${i}`)), true);
+  }
+  // The watcher's shared-cursor advance re-reads overlap by design; a
+  // re-observed event must stay a no-op even when it left the ring.
+  assertEquals(s.recordEvent(event("e0")), false);
+  assertEquals(s.countEventsLast24h(), RING_BUFFER_CAPACITY + 5);
+});
+
+Deno.test("sweepWindow prunes the dedup set with the metrics", () => {
+  const s = makeStore();
+  s.recordEvent(event("old", "provider_added", ROLLING_WINDOW_DURATION_MS * 2));
+  s.recordEvent(event("fresh"));
+  // "old" is beyond 24h: swept from metrics. It is still in the ring
+  // buffer, so it must survive in the dedup set via the ring rebuild.
+  assertEquals(s.sweepWindow(), 1);
+  assertEquals(s.recordEvent(event("old")), false);
+  assertEquals(s.recordEvent(event("fresh")), false);
+  assertEquals(s.countEventsLast24h(), 1);
+});
+
+Deno.test("getProviderPublicKeys reflects topology + chain-event writes", () => {
+  const s = makeStore();
+  s.replaceTopology([council("CA", [], ["GA", "GB"])]);
+  s.registerProvider("GC", "CA");
+  assertEquals(s.getProviderPublicKeys().sort(), ["GA", "GB", "GC"]);
+  s.unregisterProvider("GB");
+  assertEquals(s.getProviderPublicKeys().sort(), ["GA", "GC"]);
+});

src/core/sync/contract-init-listener.ts

@@ -29,14 +29,25 @@ import type { NetworkEventBus } from "@/core/events/bus.ts";
 
 /**
  * `contract_initialized` fires at deploy time, which precedes the
- * council-platform `PUT /council/metadata` call by some script-controlled
- * window. We keep an unknown contract in `pendingAdoption` and retry
- * topology refresh each poll tick for up to PENDING_TTL_MS so a council
- * registered AFTER its on-chain deploy still gets adopted. Past the TTL we
- * cache as notMoonlight so a chatty unrelated contract can't drag us into
- * repeated topology refreshes forever.
+ * council-platform `PUT /council/metadata` call by some window — script
+ * flows register within seconds, but a HUMAN-driven flow (console/CLI,
+ * the mainnet demo path) can take many minutes. We keep an unknown
+ * contract in `pendingAdoption` for up to PENDING_TTL_MS so a council
+ * registered well after its on-chain deploy still gets adopted (adoption
+ * is what triggers the historical back-fill; the periodic topology
+ * re-sync would surface the council's topology either way, but not its
+ * pre-adoption events). Past the TTL we cache as notMoonlight so an
+ * unrelated contract doesn't stay pending forever. The previous 120s TTL
+ * permanently blacklisted the mainnet demo councils — their platform
+ * registration landed minutes after deploy.
+ *
+ * Refreshes for pending unknowns are throttled to REFRESH_MIN_INTERVAL_MS
+ * (except when a NEW unknown just arrived, which refreshes immediately) so
+ * a chatty unrelated contract costs at most a couple of council-platform
+ * fetches per minute during its pending hour.
  */
-const PENDING_TTL_MS = 120_000;
+const PENDING_TTL_MS = 60 * 60 * 1000;
+const REFRESH_MIN_INTERVAL_MS = 30_000;
 
 interface PendingEntry {
   firstSeenMs: number;
@@ -46,6 +57,9 @@ interface PendingEntry {
 
 const notMoonlight = new Set<string>();
 const pendingAdoption = new Map<string, PendingEntry>();
+/** Set when an unknown was JUST registered — bypasses the refresh throttle. */
+let hasFreshPending = false;
+let lastPendingRefreshMs = 0;
 
 /**
  * Soroban event filter pattern matching `contract_initialized` events with
@@ -87,6 +101,7 @@ export function evaluateUnknownContract(
     firstSeenMs: Date.now(),
     observedAtLedger,
   });
+  hasFreshPending = true;
   log.event("registered unknown contractId for topology adoption");
 }
 
@@ -117,7 +132,17 @@ export async function drainPendingAdoptions(
   log.info("drainPendingAdoptions");
   log.debug("pendingCount", pendingAdoption.size);
 
-  await refreshTopology(`pending=${pendingAdoption.size}`, deps);
+  // Refresh immediately for a brand-new unknown (fast adoption of real
+  // councils); otherwise throttle — the periodic topology re-sync also
+  // updates `networkState`, so waiting pendings still get their
+  // `hasCouncil` check against fresh state below.
+  const refreshNow = hasFreshPending ||
+    Date.now() - lastPendingRefreshMs >= REFRESH_MIN_INTERVAL_MS;
+  if (refreshNow) {
+    hasFreshPending = false;
+    lastPendingRefreshMs = Date.now();
+    await refreshTopology(`pending=${pendingAdoption.size}`, deps);
+  }
 
   const now = Date.now();
   /** Earliest observed ledger across freshly-adopted contracts in this drain pass. */
@@ -162,4 +187,6 @@ export async function drainPendingAdoptions(
 export function __resetForTests(): void {
   notMoonlight.clear();
   pendingAdoption.clear();
+  hasFreshPending = false;
+  lastPendingRefreshMs = 0;
 }