u

src/services/healthMonitorService.js

@@ -8,8 +8,19 @@ const healthLog = createLogger('health-monitor');
 
 let scheduledTimer = null;
 let scheduledInterval = null;
+let aggregateInterval = null;
 let isServiceRunning = false;
 let redisClient = null;
+
+const RAW_RETENTION_MS = 3 * 60 * 60 * 1000;
+const AGGREGATE_WINDOW_MS = 4 * 60 * 60 * 1000;
+const AGGREGATE_INTERVAL_MS = 30 * 60 * 1000;
+const AGGREGATE_RETENTION_MS = 30 * 24 * 60 * 60 * 1000;
+
+// MARK: redis sorted-set keys (score = unix ms timestamp, member = "<ts>:<value>")
+export const rawKey = () => `${env.redis.key}:raw`;
+export const avgKey = () => `${env.redis.key}:avg`;
+export const maxKey = () => `${env.redis.key}:max`;
 // #endregion
 
 // #region Redis Client Management
@@ -86,16 +97,70 @@ export function performHealthSelftest() {
     });
 }
 
-// MARK: runs one selftest cycle and writes duration to Redis
+// MARK: returns raw (3h), avg and max (30d) latency series as [{ timestamp, value }] oldest first
+export async function getLatencySeries() {
+    const redis = getRedisClient();
+    const parse = (members) =>
+        members.map((m) => {
+            const i = m.indexOf(':');
+            return { timestamp: Number(m.slice(0, i)), value: Number(m.slice(i + 1)) };
+        });
+    const [raw, avg, max] = await Promise.all([
+        redis.zrange(rawKey(), 0, -1),
+        redis.zrange(avgKey(), 0, -1),
+        redis.zrange(maxKey(), 0, -1),
+    ]);
+    return { raw: parse(raw), avg: parse(avg), max: parse(max) };
+}
+
+// MARK: stores one latency sample and trims samples older than 3 hours (round-robin window)
+export async function recordLatency(durationMs, now = Date.now()) {
+    const redis = getRedisClient();
+    await redis
+        .multi()
+        .zadd(rawKey(), now, `${now}:${durationMs}`)
+        .zremrangebyscore(rawKey(), '-inf', now - RAW_RETENTION_MS)
+        .exec();
+}
+
+// MARK: computes avg/max over the last 4 hours, stores both with timestamp, purges aggregates older than 30 days
+export async function aggregateLatency(now = Date.now()) {
+    const redis = getRedisClient();
+    const members = await redis.zrangebyscore(rawKey(), now - AGGREGATE_WINDOW_MS, '+inf');
+    const values = members
+        .map((m) => Number(m.slice(m.indexOf(':') + 1)))
+        .filter(Number.isFinite);
+
+    if (values.length > 0) {
+        const avg = Math.round((values.reduce((a, b) => a + b, 0) / values.length) * 100) / 100;
+        const max = Math.max(...values);
+        await redis
+            .multi()
+            .zadd(avgKey(), now, `${now}:${avg}`)
+            .zadd(maxKey(), now, `${now}:${max}`)
+            .exec();
+        healthLog.debug(`Aggregated ${values.length} samples: avg=${avg}ms max=${max}ms`);
+    } else {
+        healthLog.warn('No latency samples in the last 4h; skipping aggregate.');
+    }
+
+    const cutoff = now - AGGREGATE_RETENTION_MS;
+    await redis
+        .multi()
+        .zremrangebyscore(avgKey(), '-inf', cutoff)
+        .zremrangebyscore(maxKey(), '-inf', cutoff)
+        .exec();
+}
+
+// MARK: runs one selftest cycle and records the latency in Redis
 async function executeHealthCheckCycle() {
     try {
         const result = await performHealthSelftest();
         if (result.success) {
             try {
-                const redis = getRedisClient();
-                await redis.set(env.redis.key, String(result.durationMs));
+                await recordLatency(result.durationMs);
                 healthLog.debug(
-                    `Health selftest passed in ${result.durationMs}ms; recorded in redis key "${env.redis.key}"`
+                    `Health selftest passed in ${result.durationMs}ms; recorded in redis key "${rawKey()}"`
                 );
             } catch (redisErr) {
                 healthLog.warn(`Failed to write latency to Redis: ${redisErr.message}`);
@@ -139,6 +204,13 @@ export function startHealthMonitorService() {
         });
     }, intervalMs);
 
+    // Aggregate avg/max and purge old aggregates every 30 minutes
+    aggregateInterval = setInterval(() => {
+        aggregateLatency().catch((err) => {
+            healthLog.warn(`Latency aggregation failed: ${err.message}`);
+        });
+    }, AGGREGATE_INTERVAL_MS);
+
     healthLog.info(`Health monitor service started (selftest every ${Math.round(intervalMs / 1000)}s via TCP).`);
 }
 
@@ -153,6 +225,10 @@ export async function stopHealthMonitorService() {
         clearInterval(scheduledInterval);
         scheduledInterval = null;
     }
+    if (aggregateInterval) {
+        clearInterval(aggregateInterval);
+        aggregateInterval = null;
+    }
 
     if (redisClient) {
         try {