perf: maintain telemetry source rollups

Author: chrisbbreuer

packages/ts-cloud/src/control-plane/migrations.ts

@@ -4,7 +4,7 @@ export interface ControlPlaneMigration {
   sql: string
 }
 
-export const CONTROL_PLANE_SCHEMA_VERSION: number = 35
+export const CONTROL_PLANE_SCHEMA_VERSION: number = 36
 
 export const controlPlaneMigrations: readonly ControlPlaneMigration[] = [
   {
@@ -1420,4 +1420,49 @@ export const controlPlaneMigrations: readonly ControlPlaneMigration[] = [
     CREATE INDEX dr_drill_status ON disaster_recovery_drills(project_id,status,created_at DESC);
   `,
   },
+  {
+    version: 36,
+    name: 'telemetry_source_rollups',
+    sql: `
+      CREATE TABLE telemetry_source_rollups (
+        project_id TEXT NOT NULL REFERENCES projects(id) ON DELETE CASCADE,
+        environment_id TEXT NOT NULL DEFAULT '',
+        resource_id TEXT NOT NULL DEFAULT '',
+        source TEXT NOT NULL,
+        first_observed_at TEXT NOT NULL,
+        latest_observed_at TEXT NOT NULL,
+        ingested_bytes INTEGER NOT NULL DEFAULT 0 CHECK (ingested_bytes >= 0),
+        record_count INTEGER NOT NULL DEFAULT 0 CHECK (record_count >= 0),
+        PRIMARY KEY(project_id, environment_id, resource_id, source)
+      ) STRICT;
+
+      INSERT INTO telemetry_source_rollups (
+        project_id, environment_id, resource_id, source,
+        first_observed_at, latest_observed_at, ingested_bytes, record_count
+      )
+      SELECT
+        project_id, COALESCE(environment_id, ''), COALESCE(resource_id, ''), source,
+        MIN(observed_at), MAX(observed_at), SUM(ingested_bytes), COUNT(*)
+      FROM telemetry_records
+      GROUP BY project_id, COALESCE(environment_id, ''), COALESCE(resource_id, ''), source;
+
+      CREATE TRIGGER telemetry_source_rollups_insert
+      AFTER INSERT ON telemetry_records
+      BEGIN
+        INSERT INTO telemetry_source_rollups (
+          project_id, environment_id, resource_id, source,
+          first_observed_at, latest_observed_at, ingested_bytes, record_count
+        )
+        VALUES (
+          NEW.project_id, COALESCE(NEW.environment_id, ''), COALESCE(NEW.resource_id, ''), NEW.source,
+          NEW.observed_at, NEW.observed_at, NEW.ingested_bytes, 1
+        )
+        ON CONFLICT(project_id, environment_id, resource_id, source) DO UPDATE SET
+          first_observed_at = MIN(first_observed_at, excluded.first_observed_at),
+          latest_observed_at = MAX(latest_observed_at, excluded.latest_observed_at),
+          ingested_bytes = ingested_bytes + excluded.ingested_bytes,
+          record_count = record_count + 1;
+      END;
+    `,
+  },
 ]

packages/ts-cloud/src/control-plane/store.test.ts

@@ -85,15 +85,15 @@ describe('ControlPlaneStore schema and persistence', () => {
     const backup = migrated.getSetting('storage.last_backup') as { path: string }
     expect(existsSync(backup.path)).toBe(true)
     expect(backup.path.startsWith(`${path}.v1.`)).toBe(true)
-    expect(migrated.health().schemaVersion).toBe(35)
+    expect(migrated.health().schemaVersion).toBe(36)
     expect(existsSync(parent)).toBe(true)
     migrated.close()
   })
 
   it('keeps migration numbering contiguous', () => {
     expect(controlPlaneMigrations.map((migration) => migration.version)).toEqual([
       1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31,
-      32, 33, 34, 35,
+      32, 33, 34, 35, 36,
     ])
   })
 })

packages/ts-cloud/src/telemetry/store.test.ts

@@ -133,6 +133,52 @@ describe('telemetry safety and persistence', () => {
     expect(resumed.records.map((item) => item.message)).toEqual(['second'])
   })
 
+  it('maintains exact source status rollups across inserts and retention', () => {
+    const { controlPlane, project, environment, resource, telemetry } = fixture()
+    telemetry.appendMany([
+      {
+        projectId: project.id,
+        environmentId: environment.id,
+        resourceId: resource.id,
+        kind: 'metric',
+        source: 'host',
+        name: 'cpu',
+        timestamp: '2026-06-01T00:00:00Z',
+        value: 10,
+      },
+      {
+        projectId: project.id,
+        environmentId: environment.id,
+        resourceId: resource.id,
+        kind: 'metric',
+        source: 'host',
+        name: 'cpu',
+        timestamp: '2026-07-21T11:59:00Z',
+        value: 20,
+      },
+    ])
+
+    expect(telemetry.status(project.id, environment.id, 30, 1, [resource.id])).toMatchObject([
+      {
+        source: 'host',
+        lastObservedAt: '2026-07-21T12:00:00.000Z',
+        freshness: 'live',
+      },
+    ])
+    expect(telemetry.status(project.id, environment.id, 30, 1, [])).toEqual([])
+
+    telemetry.enforceRetention(
+      { rawDays: 30, downsampleAfterDays: 30, downsampleBucketMs: 3_600_000, maxRecords: 100 },
+      project.id,
+    )
+    const rollup = controlPlane.database
+      .query<{ count: number }, []>(
+        'SELECT SUM(record_count) count FROM telemetry_source_rollups',
+      )
+      .get()
+    expect(Number(rollup?.count)).toBe(1)
+  })
+
   it('saves bounded actor-scoped queries without allowing project changes', () => {
     const { controlPlane, project, environment, telemetry } = fixture()
     const actor = controlPlane.createActor({ kind: 'user', externalId: 'user:chris', displayName: 'Chris' })

packages/ts-cloud/src/telemetry/store.ts

@@ -412,11 +412,12 @@ export class TelemetryStore {
     const environment = environmentId ? ' AND environment_id=?' : ''
     if (environmentId) bindings.push(environmentId)
     const boundedResources = [...new Set(resourceIds ?? [])].slice(0, 100)
+    if (resourceIds && boundedResources.length === 0) return []
     const resources = resourceIds ? ` AND resource_id IN (${boundedResources.map(() => '?').join(',') || "''"})` : ''
     bindings.push(...boundedResources)
     const rows = this.controlPlane.database
       .query<Row, SQLQueryBindings[]>(
-        `SELECT source, MAX(observed_at) latest, SUM(ingested_bytes) bytes, COUNT(*) count, MIN(observed_at) first FROM telemetry_records WHERE project_id=?${environment}${resources} GROUP BY source`,
+        `SELECT source, MAX(latest_observed_at) latest, SUM(ingested_bytes) bytes, SUM(record_count) count, MIN(first_observed_at) first FROM telemetry_source_rollups WHERE project_id=?${environment}${resources} GROUP BY source`,
       )
       .all(...bindings)
     const now = this.now().getTime()
@@ -590,6 +591,31 @@ export class TelemetryStore {
         `DELETE FROM telemetry_records WHERE id IN (SELECT id FROM telemetry_records${projectId ? ' WHERE project_id=?' : ''} ORDER BY timestamp ASC LIMIT ?)`,
         projectId ? [projectId, excess] : [excess],
       )
+    this.rebuildStatusRollups(projectId)
     return { deleted: removed + excess, ...downsampled }
   }
+
+  private rebuildStatusRollups(projectId?: string): void {
+    const scope = projectId ? ' WHERE project_id=?' : ''
+    const bindings: SQLQueryBindings[] = projectId ? [projectId] : []
+    const rebuild = this.controlPlane.database.transaction(() => {
+      this.controlPlane.database.run(
+        `DELETE FROM telemetry_source_rollups${scope}`,
+        bindings,
+      )
+      this.controlPlane.database.run(
+        `INSERT INTO telemetry_source_rollups (
+          project_id, environment_id, resource_id, source,
+          first_observed_at, latest_observed_at, ingested_bytes, record_count
+        )
+        SELECT
+          project_id, COALESCE(environment_id, ''), COALESCE(resource_id, ''), source,
+          MIN(observed_at), MAX(observed_at), SUM(ingested_bytes), COUNT(*)
+        FROM telemetry_records${scope}
+        GROUP BY project_id, COALESCE(environment_id, ''), COALESCE(resource_id, ''), source`,
+        bindings,
+      )
+    })
+    rebuild()
+  }
 }