feat(dashboard): Implemented basic snapshot service, polling service and ws connection management

Author: kushagra0902

src/dashboard-service/polling/polling-scheduler.ts

@@ -0,0 +1,43 @@
+import { createLogger } from '../../factories/logger-factory'
+
+type Tick = () => Promise<void> | void
+
+const debug = createLogger('dashboard-service:polling')
+
+export class PollingScheduler {
+  private timer: NodeJS.Timer | undefined
+
+  public constructor(
+    private readonly intervalMs: number,
+    private readonly tick: Tick,
+  ) { }
+
+  public start(): void {
+    if (this.timer) {
+      return
+    }
+
+    debug('starting scheduler with interval %d ms', this.intervalMs)
+
+    this.timer = setInterval(() => {
+      Promise.resolve(this.tick())
+        .catch((error) => {
+          console.error('dashboard-service: polling tick failed', error)
+        })
+    }, this.intervalMs)
+  }
+
+  public stop(): void {
+    if (!this.timer) {
+      return
+    }
+
+    debug('stopping scheduler')
+    clearInterval(this.timer)
+    this.timer = undefined
+  }
+
+  public isRunning(): boolean {
+    return Boolean(this.timer)
+  }
+}

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

@@ -0,0 +1,34 @@
+import { KPISnapshot } from '../types'
+
+export class SnapshotService {
+  private sequence = 0
+
+  private snapshot: KPISnapshot = {
+    sequence: this.sequence,
+    generatedAt: new Date(0).toISOString(),
+    status: 'placeholder',
+    metrics: {
+      eventsByKind: [],
+      admittedUsers: null,
+      satsPaid: null,
+      topTalkers: [],
+    },
+  }
+
+  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 {
+    this.sequence += 1
+
+    this.snapshot = {
+      ...this.snapshot,
+      sequence: this.sequence,
+      generatedAt: new Date().toISOString(),
+    }
+
+    return this.snapshot
+  }
+}

src/dashboard-service/ws/dashboard-ws-hub.ts

@@ -0,0 +1,141 @@
+import { DashboardServerMessage, KPISnapshot } from '../types'
+import { RawData, WebSocketServer } from 'ws'
+import { createLogger } from '../../factories/logger-factory'
+import WebSocket from 'ws'
+
+const debug = createLogger('dashboard-service:ws')
+
+export class DashboardWebSocketHub {
+  public constructor(
+    private readonly webSocketServer: WebSocketServer,
+    private readonly getSnapshot: () => KPISnapshot,
+  ) {
+    console.info('dashboard-service: websocket hub initialized')
+
+    this.webSocketServer
+      .on('connection', this.onConnection.bind(this))
+      .on('close', () => {
+        console.info('dashboard-service: websocket server closed')
+      })
+      .on('error', (error) => {
+        console.error('dashboard-service: websocket server error', error)
+      })
+  }
+
+  public broadcastSnapshot(snapshot: KPISnapshot): void {
+    this.broadcast({
+      type: 'kpi.snapshot',
+      payload: snapshot,
+    })
+  }
+
+  public broadcastTick(sequence: number): void {
+    this.broadcast({
+      type: 'kpi.tick',
+      payload: {
+        at: new Date().toISOString(),
+        sequence,
+      },
+    })
+  }
+
+  public close(): void {
+    console.info('dashboard-service: closing websocket hub')
+    this.webSocketServer.clients.forEach((client) => {
+      client.close()
+    })
+    this.webSocketServer.removeAllListeners()
+  }
+
+  private onConnection(client: WebSocket): void {
+    const connectedClients = this.getConnectedClientsCount()
+    console.info('dashboard-service: websocket client connected (clients=%d)', connectedClients)
+
+    client
+      .on('close', (code, reason) => {
+        console.info(
+          'dashboard-service: websocket client disconnected (code=%d, reason=%s, clients=%d)',
+          code,
+          reason.toString(),
+          this.getConnectedClientsCount(),
+        )
+      })
+      .on('error', (error) => {
+        console.error('dashboard-service: websocket client error', error)
+      })
+      .on('message', (raw) => {
+        this.onClientMessage(raw)
+      })
+
+    this.send(client, {
+      type: 'dashboard.connected',
+      payload: {
+        at: new Date().toISOString(),
+      },
+    })
+
+    this.send(client, {
+      type: 'kpi.snapshot',
+      payload: this.getSnapshot(),
+    })
+
+    debug('dashboard websocket bootstrap snapshot sent')
+  }
+
+  private onClientMessage(raw: RawData): void {
+    try {
+      const rawMessage = this.toUtf8(raw)
+      const message = JSON.parse(rawMessage)
+      debug('dashboard websocket client message received: %o', message)
+    } catch (error) {
+      console.error('dashboard-service: websocket message parsing failed', error)
+    }
+  }
+
+  private broadcast(message: DashboardServerMessage): void {
+    this.webSocketServer.clients.forEach((client) => {
+      if (client.readyState !== WebSocket.OPEN) {
+        return
+      }
+      this.send(client, message)
+    })
+  }
+
+  private send(client: WebSocket, message: DashboardServerMessage): void {
+    if (client.readyState !== WebSocket.OPEN) {
+      return
+    }
+
+    try {
+      client.send(JSON.stringify(message))
+    } catch (error) {
+      console.error('dashboard-service: websocket send failed', error)
+    }
+  }
+
+  private toUtf8(raw: RawData): string {
+    if (typeof raw === 'string') {
+      return raw
+    }
+
+    if (Buffer.isBuffer(raw)) {
+      return raw.toString('utf8')
+    }
+
+    if (Array.isArray(raw)) {
+      return raw.map((chunk) => {
+        if (Buffer.isBuffer(chunk)) {
+          return chunk.toString('utf8')
+        }
+
+        return Buffer.from(chunk as ArrayBuffer).toString('utf8')
+      }).join('')
+    }
+
+    return Buffer.from(raw).toString('utf8')
+  }
+
+  private getConnectedClientsCount(): number {
+    return Array.from(this.webSocketServer.clients).filter((client) => client.readyState === WebSocket.OPEN).length
+  }
+}