import { db } from '../db/index.js'; import type { CapacityHistory, CapacityQuery, CapacityRange } from '../../capacity.js'; import type { HostMode } from '../../hosts.js'; import { readTemperatureHistory } from './temperatures.js'; const windows: Record = { '1h': { duration: 3_600_000, bucketMs: 60_000 }, '24h': { duration: 86_400_000, bucketMs: 600_000 }, '7d': { duration: 7 * 86_400_000, bucketMs: 3_600_000 } }; /** Read on the report worker: heartbeat history can contain millions of rows. */ export function readCapacityHistory(query: CapacityQuery, now = Date.now()): CapacityHistory { const handle = db(); const { duration, bucketMs } = windows[query.range]; const startAt = now - duration; const count = duration / bucketMs; const hosts = handle .prepare( `SELECT id,name,created_at,mode FROM ec_hosts WHERE deleted_at IS NULL ${query.hostId ? 'AND id=?' : ''} ORDER BY id` ) .all(...(query.hostId ? [query.hostId] : [])) as { id: string; name: string; created_at: string; mode: HostMode; }[]; // The primary key bounds each host scan to the selected interval. Aggregate // before crossing the worker boundary; do not return raw heartbeat rows. const samples = handle.prepare(`SELECT MIN(?, CAST(((CAST(strftime('%s',sampled_at) AS INTEGER)*1000 + CAST(substr(strftime('%f',sampled_at),4,3) AS INTEGER)) - ?) / ? AS INTEGER)) AS bucket, AVG(MAX(0,cpu_count * (0.95 - cpu_usage_percent / 100.0))) AS cpu, AVG(MAX(0,memory_available - 500000000)) AS memory FROM ec_host_resource_samples WHERE host_id=? AND sampled_at>=? AND sampled_at<=? AND cpu_count>0 AND cpu_usage_percent BETWEEN 0 AND 100 AND memory_total>0 AND memory_available BETWEEN 0 AND memory_total GROUP BY bucket`); const points = Array.from({ length: count }, (_, index) => ({ at: startAt + index * bucketMs, cpuCoresAvailable: null as number | null, memoryBytesAvailable: null as number | null, reportingHosts: 0, expectedHosts: 0 })); for (const host of hosts) { // Current mode determines available capacity; keep inactive hosts in // the separate temperature history for diagnostics. if (host.mode !== 'active') continue; const createdAt = Date.parse(host.created_at); for (const point of points) { if (createdAt <= Math.min(now, point.at + bucketMs)) point.expectedHosts++; } const buckets = samples.all( count - 1, startAt, bucketMs, host.id, new Date(startAt).toISOString(), new Date(now).toISOString() ) as { bucket: number; cpu: number; memory: number }[]; for (const bucket of buckets) { const point = points[bucket.bucket]; if (!point || createdAt > Math.min(now, point.at + bucketMs)) continue; // Average each host first so one host reporting more frequently // cannot change another host's contribution to fleet headroom. point.reportingHosts++; point.cpuCoresAvailable = (point.cpuCoresAvailable ?? 0) + bucket.cpu; point.memoryBytesAvailable = (point.memoryBytesAvailable ?? 0) + bucket.memory; } } for (const point of points) { // A missing host is unknown, not zero available capacity. Keep the gap // visible instead of presenting a partial sum as the fleet total. if (point.reportingHosts !== point.expectedHosts) { point.cpuCoresAvailable = null; point.memoryBytesAvailable = null; } } return { range: query.range, startAt, endAt: now, bucketMs, points, temperatures: readTemperatureHistory(query, hosts, startAt, now, bucketMs) }; }