feat(dashboard): Added New test cases for collector services and updated polling services

Author: kushagra0902

package.json

@@ -40,7 +40,6 @@
     "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/polling/polling-scheduler.ts

@@ -4,40 +4,66 @@ type Tick = () => Promise<void> | void
 
 const debug = createLogger('dashboard-service:polling')
 
+/**
+ * Runs a tick callback on a fixed cadence, but — unlike setInterval — never
+ * overlaps: the next tick is only scheduled *after* the current one resolves
+ * or rejects. This prevents DB query storms when a poll takes longer than the
+ * configured interval.
+ */
 export class PollingScheduler {
-  private timer: NodeJS.Timer | undefined
+  private timer: NodeJS.Timeout | undefined
+  private running = false
 
   public constructor(
     private readonly intervalMs: number,
     private readonly tick: Tick,
   ) { }
 
   public start(): void {
-    if (this.timer) {
+    if (this.running) {
       return
     }
 
+    this.running = true
     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)
+    this.scheduleNext()
   }
 
   public stop(): void {
-    if (!this.timer) {
+    if (!this.running) {
       return
     }
 
     debug('stopping scheduler')
-    clearInterval(this.timer)
-    this.timer = undefined
+    this.running = false
+
+    if (this.timer) {
+      clearTimeout(this.timer)
+      this.timer = undefined
+    }
   }
 
   public isRunning(): boolean {
-    return Boolean(this.timer)
+    return this.running
+  }
+
+  private scheduleNext(): void {
+    if (!this.running) {
+      return
+    }
+
+    this.timer = setTimeout(() => {
+      this.timer = undefined
+
+      Promise.resolve(this.tick())
+        .catch((error) => {
+          console.error('dashboard-service: polling tick failed', error)
+        })
+        .finally(() => {
+          // Schedule the next tick only after the current one completes,
+          // regardless of success or failure.
+          this.scheduleNext()
+        })
+    }, this.intervalMs)
   }
 }

test/unit/dashboard-service/app.spec.ts

@@ -10,6 +10,8 @@ describe('dashboard-service app', () => {
       port: 0,
       wsPath: '/api/v1/kpis/stream',
       pollIntervalMs: 1000,
+      useDummyData: true,
+      collectorMode: 'full',
     })
 
     await service.start()

test/unit/dashboard-service/polling-scheduler.spec.ts

@@ -14,6 +14,17 @@ describe('PollingScheduler', () => {
     clock.restore()
   })
 
+  /**
+   * The scheduler uses recursive setTimeout (not setInterval), so each tick
+   * is only enqueued after the previous one resolves.  With instant-resolving
+   * stubs the sequence is:
+   *   T=0     start() → schedules tick at T=1000
+   *   T=1000  tick #1 resolves → schedules tick at T=2000
+   *   T=2000  tick #2 resolves → schedules tick at T=3000
+   *   T=3000  tick #3 resolves → schedules tick at T=4000
+   * tickAsync drives the microtask queue between timer firings, so all three
+   * ticks complete inside tickAsync(3000).
+   */
   it('runs tick callback on interval while running', async () => {
     const tick = Sinon.stub().resolves(undefined)
     const scheduler = new PollingScheduler(1000, tick)
@@ -30,10 +41,57 @@ describe('PollingScheduler', () => {
     const scheduler = new PollingScheduler(500, tick)
 
     scheduler.start()
-    await clock.tickAsync(1000)
+    await clock.tickAsync(1000)   // ticks at 500ms, 1000ms → 2 calls
     scheduler.stop()
-    await clock.tickAsync(1000)
+    await clock.tickAsync(1000)   // no more ticks after stop
 
     expect(tick.callCount).to.equal(2)
   })
+
+  it('does not overlap ticks when callback is slow', async () => {
+    // Tick takes 1500ms — longer than the 1000ms interval.
+    // With setInterval this would cause overlap; with recursive setTimeout it must not.
+    let running = 0
+    let maxConcurrent = 0
+
+    const tick = Sinon.stub().callsFake(async () => {
+      running++
+      maxConcurrent = Math.max(maxConcurrent, running)
+      await clock.tickAsync(1500)
+      running--
+    })
+
+    const scheduler = new PollingScheduler(1000, tick)
+    scheduler.start()
+    // Drive enough time for two potential overlapping cycles
+    await clock.tickAsync(4000)
+    scheduler.stop()
+
+    expect(maxConcurrent).to.equal(1, 'ticks must not run concurrently')
+  })
+
+  it('continues scheduling after a failed tick', async () => {
+    const tick = Sinon.stub()
+      .onFirstCall().rejects(new Error('transient error'))
+      .resolves(undefined)
+
+    const scheduler = new PollingScheduler(1000, tick)
+    scheduler.start()
+    await clock.tickAsync(3000)
+    scheduler.stop()
+
+    // First tick rejects, but the scheduler must recover and run two more.
+    expect(tick.callCount).to.be.greaterThanOrEqual(2)
+  })
+
+  it('isRunning reflects scheduler state', () => {
+    const tick = Sinon.stub().resolves(undefined)
+    const scheduler = new PollingScheduler(1000, tick)
+
+    expect(scheduler.isRunning()).to.equal(false)
+    scheduler.start()
+    expect(scheduler.isRunning()).to.equal(true)
+    scheduler.stop()
+    expect(scheduler.isRunning()).to.equal(false)
+  })
 })

test/unit/dashboard-service/snapshot-service.spec.ts

@@ -0,0 +1,107 @@
+import chai, { expect } from 'chai'
+import chaiAsPromised from 'chai-as-promised'
+import Sinon from 'sinon'
+
+import { IKPICollector, SnapshotService } from '../../../src/dashboard-service/services/snapshot-service'
+import { DashboardMetrics } from '../../../src/dashboard-service/types'
+
+chai.use(chaiAsPromised)
+
+const createMetrics = (overrides: Partial<DashboardMetrics> = {}): DashboardMetrics => ({
+  eventsByKind: [],
+  admittedUsers: 0,
+  satsPaid: 0,
+  topTalkers: {
+    allTime: [],
+    recent: [],
+  },
+  ...overrides,
+})
+
+const makeCollector = (stub: Sinon.SinonStub): IKPICollector => ({
+  collectMetrics: stub,
+})
+
+describe('SnapshotService', () => {
+  let sandbox: Sinon.SinonSandbox
+
+  beforeEach(() => {
+    sandbox = Sinon.createSandbox()
+  })
+
+  afterEach(() => {
+    sandbox.restore()
+  })
+
+  it('updates snapshot when collected metrics change', async () => {
+    const firstMetrics = createMetrics({ admittedUsers: 1 })
+    const nextMetrics = createMetrics({ admittedUsers: 2 })
+
+    const stub = sandbox.stub()
+      .onFirstCall().resolves(firstMetrics)
+      .onSecondCall().resolves(firstMetrics)
+      .onThirdCall().resolves(nextMetrics)
+
+    const service = new SnapshotService(makeCollector(stub))
+
+    const first = await service.refresh()
+    expect(first.changed).to.equal(true, 'first refresh should report changed')
+    expect(first.snapshot.sequence).to.equal(1)
+    expect(first.snapshot.status).to.equal('live')
+
+    const second = await service.refresh()
+    expect(second.changed).to.equal(false, 'second refresh with same metrics should not change')
+    expect(second.snapshot.sequence).to.equal(1, 'sequence must not advance when metrics are unchanged')
+
+    const third = await service.refresh()
+    expect(third.changed).to.equal(true, 'third refresh with new metrics should report changed')
+    expect(third.snapshot.sequence).to.equal(2)
+    expect(third.snapshot.metrics.admittedUsers).to.equal(2)
+  })
+
+  it('does not advance sequence when metrics are identical across refreshes', async () => {
+    const metrics = createMetrics({ satsPaid: 100 })
+    const stub = sandbox.stub().resolves(metrics)
+
+    const service = new SnapshotService(makeCollector(stub))
+
+    const first = await service.refresh()
+    expect(first.changed).to.equal(true)
+    expect(first.snapshot.sequence).to.equal(1)
+
+    const second = await service.refresh()
+    expect(second.changed).to.equal(false)
+    expect(second.snapshot.sequence).to.equal(1)
+  })
+
+  it('propagates collector errors to the caller', async () => {
+    const stub = sandbox.stub().rejects(new Error('db down'))
+
+    const service = new SnapshotService(makeCollector(stub))
+
+    await expect(service.refresh()).to.be.rejectedWith('db down')
+  })
+
+  it('returns the last known snapshot via getSnapshot()', async () => {
+    const metrics = createMetrics({ admittedUsers: 5 })
+    const stub = sandbox.stub().resolves(metrics)
+
+    const service = new SnapshotService(makeCollector(stub))
+
+    await service.refresh()
+
+    const snap = service.getSnapshot()
+    expect(snap.sequence).to.equal(1)
+    expect(snap.status).to.equal('live')
+    expect(snap.metrics.admittedUsers).to.equal(5)
+  })
+
+  it('sets status to live after a successful refresh', async () => {
+    const stub = sandbox.stub().resolves(createMetrics())
+
+    const service = new SnapshotService(makeCollector(stub))
+
+    const { snapshot } = await service.refresh()
+    expect(snapshot.status).to.equal('live')
+  })
+})