feat: add anomaly detection logic

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,103 @@
+CREATE TABLE IF NOT EXISTS cwv_stats_hourly
+(
+    project_id UUID,
+    route String,
+    device_type LowCardinality(String),
+    metric_name LowCardinality(String),
+    hour DateTime,
+    sum_value AggregateFunction(sum, Float64),
+    sum_squares AggregateFunction(sum, Float64),
+    count_value AggregateFunction(count, UInt64)
+)
+ENGINE = AggregatingMergeTree
+PARTITION BY toYYYYMM(hour)
+ORDER BY (project_id, hour, metric_name, route, device_type)
+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,
+    sumState(metric_value) AS sum_value,
+    sumState(metric_value * metric_value) AS sum_squares,
+    countState() AS count_value
+FROM cwv_events
+GROUP BY project_id, route, device_type, metric_name, hour;
+
+CREATE VIEW IF NOT EXISTS v_cwv_anomalies AS
+WITH 
+    toStartOfHour(now()) AS current_hour_mark,
+    current_hour AS (
+        SELECT 
+            project_id, route, device_type, metric_name, hour,
+            sumMerge(sum_value) / countMerge(count_value) AS avg_val,
+            countMerge(count_value) AS n
+        FROM cwv_stats_hourly
+        WHERE hour >= current_hour_mark - INTERVAL 1 HOUR
+        GROUP BY project_id, route, device_type, metric_name, hour
+    ),
+    baseline_stats AS (
+        SELECT 
+            project_id, route, device_type, metric_name,
+            sumMerge(sum_value) / countMerge(count_value) AS b_avg,
+            sqrt(max2((sumMerge(sum_squares) / countMerge(count_value)) - pow(b_avg, 2), 0)) AS b_stddev,
+            countMerge(count_value) AS b_n
+        FROM cwv_stats_hourly
+        WHERE hour >= current_hour_mark - INTERVAL 7 DAY 
+          AND hour < current_hour_mark - INTERVAL 1 HOUR
+        GROUP BY project_id, route, device_type, metric_name
+    )   
+SELECT
+    lower(hex(MD5(concat(
+        toString(curr.project_id), 
+        curr.route, 
+        curr.metric_name, 
+        curr.device_type, 
+        toString(curr.hour)
+    )))) AS anomaly_id,
+    curr.project_id, curr.route, curr.metric_name, curr.device_type,
+    curr.hour AS detection_time,
+    curr.avg_val AS current_avg,
+    base.b_avg AS baseline_avg,
+    base.b_stddev AS baseline_stddev,
+    (curr.avg_val - base.b_avg) / IF(base.b_stddev = 0, 0.00001, base.b_stddev) AS z_score,
+    curr.n AS sample_size
+FROM current_hour curr
+JOIN baseline_stats base ON 
+    curr.project_id = base.project_id AND 
+    curr.route = base.route AND 
+    curr.metric_name = base.metric_name AND
+    curr.device_type = base.device_type
+WHERE curr.n >= 20 AND base.b_n >= 100;
+
+CREATE TABLE IF NOT EXISTS processed_anomalies
+(
+    anomaly_id String, 
+    project_id UUID,
+    metric_name LowCardinality(String),
+    route String,
+    last_z_score Float64,
+    notified_at DateTime DEFAULT now(),
+    status Enum8('new' = 1, 'notified' = 2, 'acknowledged' = 3, 'resolved' = 4) DEFAULT 'new'
+)
+ENGINE = ReplacingMergeTree()
+ORDER BY (project_id, anomaly_id);
+
+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, UPDATE ON processed_anomalies TO r_ai_analyst;
+
+CREATE USER IF NOT EXISTS ai_analyst_user 
+IDENTIFIED WITH sha256_password BY 'ai_analyst_password';
+GRANT r_ai_analyst TO ai_analyst_user;
+ALTER USER ai_analyst_user DEFAULT ROLE r_ai_analyst;

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,60 @@
       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 currentHourMark = new Date(now.setMinutes(0, 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 < 40; 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())
+      });
+    }
+  }
+
+  for (let i = 0; i < 30; i++) {
+    const sessionId = randomUUID();
+    const recordedAt = formatDateTime64Utc(new Date(currentHourMark.getTime() + i * 60_000));
+
+    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_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,12 @@
+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 ${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(),
     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();
 }