feat: add anomaly detection logic

.github/workflows/check.yml

@@ -92,6 +92,12 @@ jobs:
         env:
           CI: true
         run: pnpm --filter cwv-monitor-app test:perf
+
+      - name: Run monitor anomaly detection tests
+        if: steps.changes.outputs.monitor_changed == 'true'
+        env:
+          CI: true
+        run: pnpm --filter cwv-monitor-app test:anomaly
 
         # SDK can destroy client build, so let's verify it here too
       - name: Check demo (Build)

apps/monitor-app/.env.ci

@@ -7,6 +7,8 @@ CLICKHOUSE_PORT=18123
 CLICKHOUSE_DB=cwv_monitor_test
 CLICKHOUSE_USER=default
 CLICKHOUSE_PASSWORD=secret
+AI_ANALYST_CLICKHOUSE_USER=ai_analyst_user
+AI_ANALYST_CLICKHOUSE_PASSWORD=ai_analyst_password
 
 # Auth (required by `src/env.ts`)
 BETTER_AUTH_SECRET=ci-test-secret-that-is-at-least-32-chars-long

apps/monitor-app/.env.example

@@ -7,6 +7,8 @@ CLICKHOUSE_PORT=8123
 CLICKHOUSE_USER=default
 CLICKHOUSE_PASSWORD=secret
 CLICKHOUSE_DB=cwv_monitor
+AI_ANALYST_CLICKHOUSE_USER=ai_analyst_user
+AI_ANALYST_CLICKHOUSE_PASSWORD=ai_analyst_password
 
 # Optional overrides for the clickhouse-migrations CLI
 CH_MIGRATIONS_HOST=${CLICKHOUSE_HOST}

apps/monitor-app/.env.test

@@ -7,6 +7,8 @@ CLICKHOUSE_PORT=18123
 CLICKHOUSE_DB=cwv_monitor_test
 CLICKHOUSE_USER=default
 CLICKHOUSE_PASSWORD=secret
+AI_ANALYST_CLICKHOUSE_USER=ai_analyst_user
+AI_ANALYST_CLICKHOUSE_PASSWORD=ai_analyst_password
 
 CH_MIGRATIONS_HOST=localhost
 CH_MIGRATIONS_PORT=8123

apps/monitor-app/clickhouse/migrations/007_anomaly_detection.sql

@@ -0,0 +1,107 @@
+CREATE TABLE IF NOT EXISTS cwv_stats_hourly
+(
+    project_id UUID,
+    route String,
+    device_type LowCardinality(String),
+    metric_name LowCardinality(String),
+    hour DateTime,
+    avg_state AggregateFunction(avg, Float64),
+    var_state AggregateFunction(varSampStable, Float64), 
+    count_state AggregateFunction(count, UInt64)
+)
+ENGINE = AggregatingMergeTree
+PARTITION BY toYYYYMM(hour)
+ORDER BY (project_id, hour, route, device_type, metric_name)
+TTL hour + INTERVAL 30 DAY;
+
+CREATE MATERIALIZED VIEW IF NOT EXISTS mv_cwv_stats_hourly
+TO cwv_stats_hourly
+AS
+SELECT
+    project_id,
+    route,
+    device_type,
+    metric_name,
+    toStartOfHour(recorded_at) AS hour,
+    avgState(log1p(metric_value)) AS avg_state,
+    varSampStableState(log1p(metric_value)) AS var_state,
+    countState() AS count_state
+FROM cwv_events
+GROUP BY project_id, hour, route, device_type, metric_name;
+
+INSERT INTO cwv_stats_hourly
+SELECT
+    project_id,
+    route,
+    device_type,
+    metric_name,
+    toStartOfHour(recorded_at) AS hour,
+    avgState(log1p(metric_value)) AS avg_state,
+    varSampStableState(log1p(metric_value)) AS var_state,
+    countState() AS count_state
+FROM cwv_events
+WHERE recorded_at >= toStartOfHour(now()) - INTERVAL 7 DAY
+GROUP BY project_id, hour, route, device_type, metric_name;
+
+CREATE VIEW IF NOT EXISTS v_cwv_anomalies AS
+WITH 
+    toStartOfHour(now()) AS current_hour_mark,
+    current_hour_mark - INTERVAL 1 HOUR AS gap_hour,
+    current_hour_mark - INTERVAL 7 DAY AS baseline_start
+SELECT
+    lower(hex(MD5(concat(
+        toString(project_id), '\0', route, '\0', metric_name, '\0', device_type, '\0', toString(current_hour_mark)
+    )))) AS anomaly_id,
+    project_id, route, metric_name, device_type,
+    current_hour_mark AS detection_time,
+
+    avgMergeIf(avg_state, hour = current_hour_mark) AS log_avg_curr,
+    avgMergeIf(avg_state, hour >= baseline_start AND hour < gap_hour) AS log_avg_base,
+    sqrt(varSampStableMergeIf(var_state, hour >= baseline_start AND hour < gap_hour)) AS log_stddev_base,
+
+    exp(log_avg_curr) - 1 AS current_avg_raw,
+    exp(log_avg_base) - 1 AS baseline_avg_raw,
+
+    countMergeIf(count_state, hour = current_hour_mark) AS sample_size,
+    countMergeIf(count_state, hour >= baseline_start AND hour < gap_hour) AS baseline_n,
+
+    (log_avg_curr - log_avg_base) / IF(log_stddev_base = 0, 0.00001, log_stddev_base) AS z_score
+FROM cwv_stats_hourly
+WHERE hour >= baseline_start
+GROUP BY project_id, route, device_type, metric_name
+HAVING sample_size >= 20 AND baseline_n >= 100;
+
+CREATE TABLE IF NOT EXISTS processed_anomalies
+(
+    anomaly_id String, 
+    project_id UUID,
+    metric_name LowCardinality(String),
+    route String,
+    device_type LowCardinality(String),
+    last_z_score Float64,
+    notified_at DateTime DEFAULT now(),
+    status Enum8('new' = 1, 'notified' = 2, 'acknowledged' = 3, 'resolved' = 4) DEFAULT 'new',
+    updated_at DateTime64(3) DEFAULT now64(3)
+)
+ENGINE = ReplacingMergeTree(updated_at)
+ORDER BY (project_id, anomaly_id);
+
+CREATE SETTINGS PROFILE IF NOT EXISTS ai_analyst_profile SETTINGS
+    max_execution_time = 15,
+    max_memory_usage = 2000000000,
+    max_rows_to_read = 100000000;
+
+CREATE ROLE IF NOT EXISTS r_ai_analyst;
+GRANT SELECT ON cwv_events TO r_ai_analyst;
+GRANT SELECT ON custom_events TO r_ai_analyst;
+GRANT SELECT ON cwv_daily_aggregates TO r_ai_analyst;
+GRANT SELECT ON cwv_stats_hourly TO r_ai_analyst;
+GRANT SELECT ON v_cwv_anomalies TO r_ai_analyst;
+GRANT SELECT ON projects TO r_ai_analyst;
+GRANT SELECT, INSERT, ALTER UPDATE ON processed_anomalies TO r_ai_analyst;
+
+CREATE USER IF NOT EXISTS ai_analyst_user 
+IDENTIFIED WITH no_password;
+GRANT r_ai_analyst TO ai_analyst_user;
+ALTER USER ai_analyst_user DEFAULT ROLE r_ai_analyst;
+ALTER USER ai_analyst_user SETTINGS PROFILE ai_analyst_profile;

apps/monitor-app/package.json

@@ -14,7 +14,8 @@
     "seed:demo": "node ./scripts/seed-demo-data.mjs",
     "seed:cwv-events": "node ./scripts/seed-demo-data.mjs --custom-events-only",
     "test:integration": "vitest --config vitest.integration.config.ts",
-    "test:perf": "vitest run ./src/test/performance-guardrails.test.ts --config vitest.performance.config.ts"
+    "test:perf": "vitest run ./src/test/performance-guardrails.test.ts --config vitest.performance.config.ts",
+    "test:anomaly": "vitest run ./src/test/anomaly-detection.test.ts --config vitest.anomaly.config.ts"
   },
   "dependencies": {
     "@clickhouse/client": "^1.14.0",

apps/monitor-app/scripts/seed-demo-data.mjs

@@ -451,7 +451,7 @@
   const existingCount = typeof rawCount === "string" ? Number.parseInt(rawCount, 10) : Number(rawCount);
 
   if (existingCount > 0 && RESET_BEFORE_SEED) {
     console.log(`Resetting existing custom_events for project ${PROJECT_NAME} (${existingCount} rows) before seeding`);
     await client.command({
       query: "ALTER TABLE custom_events DELETE WHERE project_id = {projectId:UUID}",
       query_params: { projectId: PROJECT_ID },
@@ -461,7 +461,7 @@
   const finalExistingCount = RESET_BEFORE_SEED ? 0 : existingCount;
   const remaining = RESET_BEFORE_SEED ? TARGET_EVENTS : Math.max(TARGET_EVENTS - finalExistingCount, 0);
   if (remaining === 0) {
     console.log(
       `custom_events already has ${existingCount} rows for project ${PROJECT_NAME}; target ${TARGET_EVENTS}. Nothing to do.`,
     );
     return;
@@ -478,11 +478,11 @@
     });
 
     if ((index + 1) % 10 === 0 || index === batches.length - 1) {
       console.log(`Inserted custom_events batch ${index + 1}/${batches.length} (${batch.length} rows)`);
     }
   }
 
   console.log(
     `Seeded ${events.length} custom_events over the last ${DAYS_RANGE} days for project ${PROJECT_NAME} (${PROJECT_ID}).`,
   );
 }
@@ -506,7 +506,7 @@
   try {
     await client.query({ query: "SELECT 1" });
   } catch (error) {
     console.error("Unable to reach ClickHouse. Check CLICKHOUSE_* env vars.", error);
     await client.close();
     process.exit(1);
   }
@@ -533,17 +533,17 @@
             });
 
             if ((index + 1) % 20 === 0 || index === batches.length - 1) {
               console.log(`Inserted page_view batch ${index + 1}/${batches.length} (${batch.length} rows)`);
             }
           }
         }
 
         console.log(
           `Demo data already present for project ${DEMO_PROJECT_NAME} (${existingEvents} events). Skipping seeding.`,
         );
       } else {
         if (existingEvents > 0 && RESET_BEFORE_SEED) {
           console.log(`Resetting existing demo data for project ${DEMO_PROJECT_NAME} (${existingEvents} events)`);
           await deleteExistingData(client);
         }
 
@@ -567,7 +567,7 @@
           });
         }
 
         console.log(
           `Seeded ${cwvEvents.length} CWV events and ${pageViewEvents.length} page_view events over ${DAYS_TO_GENERATE + 1} days (today + ${DAYS_TO_GENERATE} days back) for project ${DEMO_PROJECT_NAME} (${DEMO_PROJECT_ID}).`,
         );
       }
@@ -577,13 +577,74 @@
       await seedCustomEventsData(client);
     }
   } catch (error) {
     console.error("Seeding failed", error);
     process.exitCode = 1;
   } finally {
     await client.close();
   }
 }
 
+export async function seedAnomalyTestPattern(client, projectId) {
+  const now = new Date();
+
+  const minutesPastHour = now.getMinutes();
+  const currentHourMark = new Date(now.setMinutes(0, 0, 0));
+
+  const events = [];
+  const route = "/checkout";
+  const device = "desktop";
+
+  for (let dayOffset = 1; dayOffset <= 3; dayOffset++) {
+    const dayStart = new Date(currentHourMark.getTime() - dayOffset * 86_400_000);
+    for (let i = 0; i < 50; i++) {
+      const sessionId = randomUUID();
+      const recordedAt = formatDateTime64Utc(new Date(dayStart.getTime() + i * 60_000));
+
+      events.push({
+        project_id: projectId, session_id: sessionId, route, path: "/checkout",
+        device_type: device, metric_name: "LCP", metric_value: 2000 + (rng() * 300),
+        rating: "good", recorded_at: recordedAt, ingested_at: formatDateTime64Utc(new Date())
+      }, 
+      {
+        project_id: projectId, session_id: sessionId, route, path: "/checkout",
+        device_type: device, metric_name: "TTFB", metric_value: 400 + (rng() * 100),
+        rating: "good", recorded_at: recordedAt, ingested_at: formatDateTime64Utc(new Date())
+      });
+    }
+  }
+
+  const intervalMs = minutesPastHour > 30 
+    ? 60_000
+    : Math.floor((minutesPastHour * 60_000) / 35);
+
+  for (let i = 0; i < 30; i++) {
+    const sessionId = randomUUID();
+    const offset = 5000 + (i * intervalMs);
+    const recordedAtDate = new Date(now.getTime() - offset);
+
+    if (recordedAtDate < currentHourMark) {
+        recordedAtDate.setTime(currentHourMark.getTime() + (i * 1000));
+    }
+
+    const recordedAt = formatDateTime64Utc(recordedAtDate);
+
+    events.push({
+      project_id: projectId, session_id: sessionId, route, path: "/checkout",
+      device_type: device, metric_name: "LCP", metric_value: 8000 + (rng() * 1000),
+      rating: "poor", recorded_at: recordedAt, ingested_at: formatDateTime64Utc(new Date())
+    }, 
+    {
+      project_id: projectId, session_id: sessionId, route, path: "/checkout",
+      device_type: device, metric_name: "TTFB", metric_value: 600 + (rng() * 200),
+      rating: "good", recorded_at: recordedAt, ingested_at: formatDateTime64Utc(new Date())
+    });
+  }
+
+  await client.insert({ table: "cwv_events", values: events, format: "JSONEachRow" });
+  await client.command({ query: "OPTIMIZE TABLE cwv_events FINAL" });
+  await client.command({ query: "OPTIMIZE TABLE cwv_stats_hourly FINAL" });
+}
+
 const isCliInvocation = import.meta.url === pathToFileURL(process.argv[1]).href;
 if (isCliInvocation) {
   const { seedCwvEvents, seedCustomEvents } = parseArgs(process.argv.slice(2));

apps/monitor-app/src/app/server/lib/clickhouse/bootstrap.ts

@@ -0,0 +1,13 @@
+import { sql } from "@/app/server/lib/clickhouse/client";
+import { env } from "@/env";
+
+export async function syncDatabaseRoles() {
+  const aiUser = env.AI_ANALYST_CLICKHOUSE_USER;
+  const aiPass = env.AI_ANALYST_CLICKHOUSE_PASSWORD;
+
+  if (!aiUser || !aiPass) return;
+
+  await sql`
+    ALTER USER IF EXISTS ${sql.identifier(aiUser)} IDENTIFIED WITH sha256_password BY ${aiPass}
+  `.command();
+}

apps/monitor-app/src/env.ts

@@ -33,6 +33,8 @@ export const env = createEnv({
     CLICKHOUSE_USER: z.string().min(1, "CLICKHOUSE_USER is required"),
     CLICKHOUSE_PASSWORD: z.string(),
     CLICKHOUSE_DB: z.string().min(1, "CLICKHOUSE_DB is required"),
+    AI_ANALYST_CLICKHOUSE_USER: z.string().min(1).default("ai_analyst_user"),
+    AI_ANALYST_CLICKHOUSE_PASSWORD: z.string().min(1),
     BETTER_AUTH_SECRET: z.string(),
     CLICKHOUSE_ADAPTER_DEBUG_LOGS: z.coerce.boolean().default(false),
     MIN_PASSWORD_SCORE: z.coerce.number().min(0).max(4).default(2),
@@ -57,6 +59,8 @@ export const env = createEnv({
     CLICKHOUSE_USER: process.env.CLICKHOUSE_USER,
     CLICKHOUSE_PASSWORD: process.env.CLICKHOUSE_PASSWORD,
     CLICKHOUSE_DB: process.env.CLICKHOUSE_DB,
+    AI_ANALYST_CLICKHOUSE_USER: process.env.AI_ANALYST_CLICKHOUSE_USER,
+    AI_ANALYST_CLICKHOUSE_PASSWORD: process.env.AI_ANALYST_CLICKHOUSE_PASSWORD,
     BETTER_AUTH_SECRET: process.env.BETTER_AUTH_SECRET,
     CLICKHOUSE_ADAPTER_DEBUG_LOGS: process.env.CLICKHOUSE_ADAPTER_DEBUG_LOGS,
     INITIAL_USER_EMAIL: process.env.INITIAL_USER_EMAIL,

apps/monitor-app/src/instrumentation.ts

@@ -1,5 +1,7 @@
+import { syncDatabaseRoles } from "@/app/server/lib/clickhouse/bootstrap";
 import { provisionInitialUser } from "@/lib/provision-initial-user";
 
 export async function register() {
   await provisionInitialUser();
+  await syncDatabaseRoles();
 }