feat(dashboard): Wired up new collectors

Author: kushagra0902

src/dashboard-service/app.ts

@@ -1,14 +1,17 @@
+import { getMasterDbClient, getReadReplicaDbClient } from '../database/client'
+import { IKPICollector, SnapshotService } from './services/snapshot-service'
 import { createDashboardRouter } from './api/dashboard-router'
 import { createLogger } from '../factories/logger-factory'
 import { DashboardServiceConfig } from './config'
 import { DashboardWebSocketHub } from './ws/dashboard-ws-hub'
 import express from 'express'
 import { getHealthRequestHandler } from './handlers/request-handlers/get-health-request-handler'
 import http from 'http'
+import { IncrementalKPICollectorService } from './services/incremental-kpi-collector-service'
+import { KPICollectorService } from './services/kpi-collector-service'
 import { PollingScheduler } from './polling/polling-scheduler'
-import { SnapshotService } from './services/snapshot-service'
+import { StatefulIncrementalKPICollectorService } from './services/stateful-incremental-service'
 import { WebSocketServer } from 'ws'
-
 const debug = createLogger('dashboard-service:app')
 
 export interface DashboardService {
@@ -22,14 +25,36 @@ export interface DashboardService {
 
 export const createDashboardService = (config: DashboardServiceConfig): DashboardService => {
   console.info(
-    'dashboard-service: creating service (host=%s, port=%d, wsPath=%s, pollIntervalMs=%d)',
+    'dashboard-service: creating service (host=%s, port=%d, wsPath=%s, pollIntervalMs=%d, useDummyData=%s, collectorMode=%s)',
     config.host,
     config.port,
     config.wsPath,
     config.pollIntervalMs,
+    config.useDummyData,
+    config.collectorMode,
   )
 
-  const snapshotService = new SnapshotService()
+  const collector: IKPICollector = config.useDummyData
+    ? {
+      collectMetrics: async () => ({
+        eventsByKind: [],
+        admittedUsers: 0,
+        satsPaid: 0,
+        topTalkers: { allTime: [], recent: [] },
+      }),
+    }
+    : (() => {
+      if (config.collectorMode === 'stateful-incremental') {
+        return new StatefulIncrementalKPICollectorService(getMasterDbClient())
+      }
+
+      const dbClient = getReadReplicaDbClient()
+      return config.collectorMode === 'incremental'
+        ? new IncrementalKPICollectorService(dbClient)
+        : new KPICollectorService(dbClient)
+    })()
+
+  const snapshotService = new SnapshotService(collector)
 
   const app = express()
     .disable('x-powered-by')
@@ -44,11 +69,17 @@ export const createDashboardService = (config: DashboardServiceConfig): Dashboar
 
   const webSocketHub = new DashboardWebSocketHub(webSocketServer, () => snapshotService.getSnapshot())
 
-  const pollingScheduler = new PollingScheduler(config.pollIntervalMs, () => {
-    const nextSnapshot = snapshotService.refreshPlaceholder()
-    debug('poll tick produced snapshot sequence=%d', nextSnapshot.sequence)
-    webSocketHub.broadcastTick(nextSnapshot.sequence)
-    webSocketHub.broadcastSnapshot(nextSnapshot)
+  const pollingScheduler = new PollingScheduler(config.pollIntervalMs, async () => {
+    const { snapshot, changed } = await snapshotService.refresh()
+
+    if (!changed) {
+      debug('poll tick detected no KPI changes')
+      return
+    }
+
+    debug('poll tick produced snapshot sequence=%d status=%s', snapshot.sequence, snapshot.status)
+    webSocketHub.broadcastTick(snapshot.sequence)
+    webSocketHub.broadcastSnapshot(snapshot)
   })
 
   const start = async () => {
@@ -72,13 +103,31 @@ export const createDashboardService = (config: DashboardServiceConfig): Dashboar
       })
     })
 
+    try {
+      const initialSnapshotRefresh = await snapshotService.refresh()
+      if (initialSnapshotRefresh.changed) {
+        debug('initial snapshot prepared with sequence=%d status=%s', initialSnapshotRefresh.snapshot.sequence, initialSnapshotRefresh.snapshot.status)
+      }
+    } catch (error) {
+      console.error('dashboard-service: initial snapshot refresh failed (will retry on next poll)', error)
+    }
+
     pollingScheduler.start()
     console.info('dashboard-service: polling scheduler started')
   }
 
   const stop = async () => {
     console.info('dashboard-service: stopping service')
     pollingScheduler.stop()
+
+    if (collector?.close) {
+      try {
+        await collector.close()
+      } catch (error) {
+        console.error('dashboard-service: failed to close collector resources', error)
+      }
+    }
+
     webSocketHub.close()
     await new Promise<void>((resolve, reject) => {
       if (!webServer.listening) {

src/dashboard-service/config.ts

@@ -3,6 +3,18 @@ export interface DashboardServiceConfig {
   port: number
   wsPath: string
   pollIntervalMs: number
+  useDummyData: boolean
+  collectorMode: DashboardCollectorMode
+}
+
+export type DashboardCollectorMode = 'full' | 'incremental' | 'stateful-incremental'
+
+const parseBoolean = (value: string | undefined, fallback = false): boolean => {
+  if (typeof value === 'undefined') {
+    return fallback
+  }
+
+  return value === '1' || value.toLowerCase() === 'true'
 }
 
 const parseInteger = (value: string | undefined, fallback: number): number => {
@@ -18,11 +30,29 @@ const parseInteger = (value: string | undefined, fallback: number): number => {
   return parsed
 }
 
+const parseCollectorMode = (
+  value: string | undefined,
+  fallback: DashboardCollectorMode = 'full',
+): DashboardCollectorMode => {
+  if (typeof value === 'undefined') {
+    return fallback
+  }
+
+  const normalized = value.toLowerCase()
+  if (normalized === 'full' || normalized === 'incremental' || normalized === 'stateful-incremental') {
+    return normalized
+  }
+  
+  return fallback
+}
+
 export const getDashboardServiceConfig = (): DashboardServiceConfig => {
   return {
     host: process.env.DASHBOARD_SERVICE_HOST ?? '127.0.0.1',
     port: parseInteger(process.env.DASHBOARD_SERVICE_PORT, 8011),
     wsPath: process.env.DASHBOARD_WS_PATH ?? '/api/v1/kpis/stream',
     pollIntervalMs: parseInteger(process.env.DASHBOARD_POLL_INTERVAL_MS, 5000),
+    useDummyData: parseBoolean(process.env.DASHBOARD_USE_DUMMY_DATA, false),
+    collectorMode: parseCollectorMode(process.env.DASHBOARD_COLLECTOR_MODE, 'full'),
   }
 }

src/dashboard-service/services/snapshot-service.ts

@@ -1,34 +1,81 @@
-import { KPISnapshot } from '../types'
+import { DashboardMetrics, KPISnapshot } from '../types'
+import { createLogger } from '../../factories/logger-factory'
+
+const debug = createLogger('dashboard-service:snapshot-service')
+
+const defaultMetrics = (): DashboardMetrics => ({
+  eventsByKind: [],
+  admittedUsers: 0,
+  satsPaid: 0,
+  topTalkers: {
+    allTime: [],
+    recent: [],
+  },
+})
+
+export interface ISnapshotRefreshResult {
+  snapshot: KPISnapshot
+  changed: boolean
+}
+
+export interface IKPICollector {
+  collectMetrics(): Promise<DashboardMetrics>
+  close?(): Promise<void> | void
+}
 
 export class SnapshotService {
+  private metricsFingerprint = JSON.stringify(defaultMetrics())
+
   private sequence = 0
 
   private snapshot: KPISnapshot = {
     sequence: this.sequence,
     generatedAt: new Date(0).toISOString(),
-    status: 'placeholder',
-    metrics: {
-      eventsByKind: [],
-      admittedUsers: null,
-      satsPaid: null,
-      topTalkers: [],
-    },
+    status: 'live',
+    metrics: defaultMetrics(),
   }
 
+  public constructor(private readonly collector: IKPICollector) { }
+
   public getSnapshot(): KPISnapshot {
     return this.snapshot
   }
 
-  // Phase 1 placeholder: advances sequence/time so polling and websocket flow can be validated end-to-end.
-  public refreshPlaceholder(): KPISnapshot {
+  /**
+   * Fetches fresh metrics from the collector and updates the snapshot if the
+   * metrics have changed.  Throws if the collector is unavailable — callers
+   * are responsible for catching and deciding how to surface errors.
+   */
+  public async refresh(): Promise<ISnapshotRefreshResult> {
+    const metrics = await this.collector.collectMetrics()
+    const nextFingerprint = JSON.stringify(metrics)
+
+    if (nextFingerprint === this.metricsFingerprint && this.snapshot.status === 'live') {
+      debug('metrics unchanged, skipping snapshot sequence update')
+      return {
+        snapshot: this.snapshot,
+        changed: false,
+      }
+    }
+
+    this.metricsFingerprint = nextFingerprint
+
+    return this.updateSnapshot(metrics, 'live')
+  }
+
+  private updateSnapshot(metrics: DashboardMetrics, status: 'live' | 'stale'): ISnapshotRefreshResult {
     this.sequence += 1
 
     this.snapshot = {
-      ...this.snapshot,
       sequence: this.sequence,
       generatedAt: new Date().toISOString(),
+      status,
+      metrics,
     }
 
-    return this.snapshot
+    return {
+      snapshot: this.snapshot,
+      changed: true,
+    }
   }
 }

src/dashboard-service/types.ts

@@ -3,16 +3,26 @@ export interface TopTalker {
   count: number
 }
 
+export interface EventsByKindCount {
+  kind: string
+  count: number
+}
+
+export interface DashboardMetrics {
+  eventsByKind: EventsByKindCount[]
+  admittedUsers: number
+  satsPaid: number
+  topTalkers: {
+    allTime: TopTalker[]
+    recent: TopTalker[]
+  }
+}
+
 export interface KPISnapshot {
   sequence: number
   generatedAt: string
-  status: 'placeholder'
-  metrics: {
-    eventsByKind: Array<{ kind: string, count: number }>
-    admittedUsers: number | null
-    satsPaid: number | null
-    topTalkers: TopTalker[]
-  }
+  status: 'live' | 'stale'
+  metrics: DashboardMetrics
 }
 
 export interface DashboardSnapshotResponse {