Phase 1 for dashboard: skeleton api server and defining data types to be sent

package.json

@@ -23,10 +23,12 @@
   "main": "src/index.ts",
   "scripts": {
     "dev": "node -r ts-node/register src/index.ts",
+    "dev:dashboard": "node -r ts-node/register src/dashboard-service/index.ts",
     "clean": "rimraf ./{dist,.nyc_output,.test-reports,.coverage}",
     "build": "tsc --project tsconfig.build.json",
     "prestart": "npm run build",
     "start": "cd dist && node src/index.js",
+    "start:dashboard": "node dist/src/dashboard-service/index.js",
     "build:check": "npm run build -- --noEmit",
     "lint": "eslint --ext .ts ./src ./test",
     "lint:report": "eslint -o .lint-reports/eslint.json -f json --ext .ts ./src ./test",
@@ -37,6 +39,8 @@
     "db:seed": "knex seed:run",
     "pretest:unit": "mkdir -p .test-reports/unit",
     "test:unit": "mocha 'test/**/*.spec.ts'",
+    "test:unit:dashboard": "mocha 'test/unit/dashboard-service/**/*.spec.ts'",
+    "test:e2e:dashboard:ws": "node scripts/check_dashboard_ws_updates.js",
     "test:unit:watch": "npm run test:unit -- --min --watch --watch-files src/**/*,test/**/*",
     "cover:unit": "nyc --report-dir .coverage/unit npm run test:unit",
     "docker:build": "docker build -t nostream .",

src/dashboard-service/api/dashboard-router.ts

@@ -0,0 +1,12 @@
+import { Router } from 'express'
+
+import { createGetKPISnapshotRequestHandler } from '../handlers/request-handlers/get-kpi-snapshot-request-handler'
+import { SnapshotService } from '../services/snapshot-service'
+
+export const createDashboardRouter = (snapshotService: SnapshotService): Router => {
+  const router = Router()
+
+  router.get('/snapshot', createGetKPISnapshotRequestHandler(snapshotService))
+
+  return router
+}

src/dashboard-service/app.ts

@@ -0,0 +1,116 @@
+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 { PollingScheduler } from './polling/polling-scheduler'
+import { SnapshotService } from './services/snapshot-service'
+import { WebSocketServer } from 'ws'
+
+const debug = createLogger('dashboard-service:app')
+
+export interface DashboardService {
+  readonly config: DashboardServiceConfig
+  readonly snapshotService: SnapshotService
+  readonly pollingScheduler: PollingScheduler
+  start(): Promise<void>
+  stop(): Promise<void>
+  getHttpPort(): number
+}
+
+export const createDashboardService = (config: DashboardServiceConfig): DashboardService => {
+  console.info(
+    'dashboard-service: creating service (host=%s, port=%d, wsPath=%s, pollIntervalMs=%d)',
+    config.host,
+    config.port,
+    config.wsPath,
+    config.pollIntervalMs,
+  )
+
+  const snapshotService = new SnapshotService()
+
+  const app = express()
+    .disable('x-powered-by')
+    .get('/healthz', getHealthRequestHandler)
+    .use('/api/v1/kpis', createDashboardRouter(snapshotService))
+
+  const webServer = http.createServer(app)
+  const webSocketServer = new WebSocketServer({
+    server: webServer,
+    path: config.wsPath,
+  })
+
+  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 start = async () => {
+    if (webServer.listening) {
+      debug('start requested but service is already listening')
+      return
+    }
+
+    console.info('dashboard-service: starting http and websocket servers')
+
+    await new Promise<void>((resolve, reject) => {
+      webServer.listen(config.port, config.host, () => {
+        const address = webServer.address()
+        debug('listening on %o', address)
+        console.info('dashboard-service: listening on %o', address)
+        resolve()
+      })
+      webServer.once('error', (error) => {
+        console.error('dashboard-service: failed to start server', error)
+        reject(error)
+      })
+    })
+
+    pollingScheduler.start()
+    console.info('dashboard-service: polling scheduler started')
+  }
+
+  const stop = async () => {
+    console.info('dashboard-service: stopping service')
+    pollingScheduler.stop()
+    webSocketHub.close()
+    await new Promise<void>((resolve, reject) => {
+      if (!webServer.listening) {
+        debug('stop requested while server was already stopped')
+        resolve()
+        return
+      }
+
+      webServer.close((error) => {
+        if (error) {
+          console.error('dashboard-service: failed to stop cleanly', error)
+          reject(error)
+          return
+        }
+
+        console.info('dashboard-service: http server closed')
+        resolve()
+      })
+    })
+  }
+
+  const getHttpPort = (): number => {
+    const address = webServer.address()
+    return typeof address === 'object' && address !== null ? address.port : config.port
+  }
+
+  return {
+    config,
+    snapshotService,
+    pollingScheduler,
+    start,
+    stop,
+    getHttpPort,
+  }
+}

src/dashboard-service/config.ts

@@ -0,0 +1,28 @@
+export interface DashboardServiceConfig {
+  host: string
+  port: number
+  wsPath: string
+  pollIntervalMs: number
+}
+
+const parseInteger = (value: string | undefined, fallback: number): number => {
+  if (typeof value === 'undefined' || value === '') {
+    return fallback
+  }
+
+  const parsed = Number(value)
+  if (!Number.isInteger(parsed) || parsed < 0) {
+    return fallback
+  }
+
+  return parsed
+}
+
+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),
+  }
+}

src/dashboard-service/controllers/get-health-controller.ts

@@ -0,0 +1,12 @@
+import { Request, Response } from 'express'
+
+import { IController } from '../../@types/controllers'
+
+export class GetHealthController implements IController {
+  public async handleRequest(_request: Request, response: Response): Promise<void> {
+    response
+      .status(200)
+      .setHeader('content-type', 'application/json; charset=utf-8')
+      .send({ status: 'ok' })
+  }
+}

src/dashboard-service/controllers/get-kpi-snapshot-controller.ts

@@ -0,0 +1,20 @@
+import { Request, Response } from 'express'
+
+import { DashboardSnapshotResponse } from '../types'
+import { IController } from '../../@types/controllers'
+import { SnapshotService } from '../services/snapshot-service'
+
+export class GetKPISnapshotController implements IController {
+  public constructor(private readonly snapshotService: SnapshotService) { }
+
+  public async handleRequest(_request: Request, response: Response): Promise<void> {
+    const payload: DashboardSnapshotResponse = {
+      data: this.snapshotService.getSnapshot(),
+    }
+
+    response
+      .status(200)
+      .setHeader('content-type', 'application/json; charset=utf-8')
+      .send(payload)
+  }
+}

src/dashboard-service/handlers/request-handlers/get-health-request-handler.ts

@@ -0,0 +1,5 @@
+import { withController } from '../../../handlers/request-handlers/with-controller-request-handler'
+
+import { GetHealthController } from '../../controllers/get-health-controller'
+
+export const getHealthRequestHandler = withController(() => new GetHealthController())

src/dashboard-service/handlers/request-handlers/get-kpi-snapshot-request-handler.ts

@@ -0,0 +1,8 @@
+import { withController } from '../../../handlers/request-handlers/with-controller-request-handler'
+
+import { GetKPISnapshotController } from '../../controllers/get-kpi-snapshot-controller'
+import { SnapshotService } from '../../services/snapshot-service'
+
+export const createGetKPISnapshotRequestHandler = (snapshotService: SnapshotService) => {
+  return withController(() => new GetKPISnapshotController(snapshotService))
+}

src/dashboard-service/index.ts

@@ -0,0 +1,42 @@
+import { createLogger } from '../factories/logger-factory'
+
+import { createDashboardService } from './app'
+import { getDashboardServiceConfig } from './config'
+
+const debug = createLogger('dashboard-service:index')
+
+const run = async () => {
+  const config = getDashboardServiceConfig()
+  console.info('dashboard-service: bootstrapping with config %o', config)
+  const service = createDashboardService(config)
+
+  const shutdown = async () => {
+    console.info('dashboard-service: received shutdown signal')
+    debug('received shutdown signal')
+    await service.stop()
+    process.exit(0)
+  }
+
+  process
+    .on('SIGINT', shutdown)
+    .on('SIGTERM', shutdown)
+
+  process.on('uncaughtException', (error) => {
+    console.error('dashboard-service: uncaught exception', error)
+  })
+
+  process.on('unhandledRejection', (error) => {
+    console.error('dashboard-service: unhandled rejection', error)
+  })
+
+  await service.start()
+}
+
+if (require.main === module) {
+  run().catch((error) => {
+    console.error('dashboard-service: unable to start', error)
+    process.exit(1)
+  })
+}
+
+export { run }

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)
+  }
+}