feat: connect cwv to ai provider

.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
@@ -16,5 +18,10 @@ INITIAL_USER_EMAIL=test@example.com
 INITIAL_USER_PASSWORD=testpassword123
 INITIAL_USER_NAME=Test User
 
+# AI Configuration (required by `src/env.ts`)
+AI_API_KEY=api_key
+AI_PROVIDER=ai_provider
+AI_MODEL=ai_model
+
 # Logging (keep CI output quieter)
 LOG_LEVEL=silent

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}
@@ -23,3 +25,7 @@ INITIAL_USER_PASSWORD=password
 INITIAL_USER_NAME=User
 
 MIN_PASSWORD_SCORE=2
+
+AI_API_KEY=
+AI_PROVIDER=
+AI_MODEL=

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,9 +14,14 @@
     "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": {
+    "@ai-sdk/anthropic": "^3.0.47",
+    "@ai-sdk/google": "^3.0.31",
+    "@ai-sdk/openai": "^3.0.34",
+    "@ai-sdk/react": "^3.0.103",
     "@clickhouse/client": "^1.14.0",
     "@hookform/resolvers": "^5.2.2",
     "@next-safe-action/adapter-react-hook-form": "^2.0.0",
@@ -30,6 +35,7 @@
     "@radix-ui/react-tooltip": "^1.2.8",
     "@t3-oss/env-nextjs": "^0.13.8",
     "@tanstack/react-table": "^8.21.3",
+    "ai": "^6.0.101",
     "arktype": "^2.1.28",
     "better-auth": "^1.4.9",
     "class-variance-authority": "^0.7.1",

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/api/chat/[projectId]/route.ts

@@ -0,0 +1,25 @@
+import { createAgent } from "@/app/server/lib/agent/agent";
+import { createAgentUIStreamResponse, UIMessage } from "ai";
+import type { NextRequest } from "next/server";
+
+export async function POST(req: NextRequest, { params }: { params: Promise<{ projectId: string }> }) {
+  try {
+    const { projectId } = await params;
+    const { messages }: { messages: UIMessage[] } = await req.json();
+
+    const agent = createAgent(projectId);
+
+    return createAgentUIStreamResponse({
+      agent,
+      uiMessages: messages,
+    });
+  } catch (error) {
+    const errorMessage = error instanceof Error ? error.message : String(error);
+    return Response.json(
+      { ok: false, error: errorMessage },
+      {
+        status: 500,
+      },
+    );
+  }
+}

apps/monitor-app/src/app/server/lib/agent/agent.ts

@@ -0,0 +1,46 @@
+import { env } from "@/env";
+import { stepCountIs, ToolLoopAgent } from "ai";
+import { executeClickHouse } from "@/app/server/lib/agent/tools/execute-clickhouse";
+import { buildSystemPrompt } from "@/app/server/lib/agent/utils/system-prompt";
+import { createAnthropic } from "@ai-sdk/anthropic";
+import { createGoogleGenerativeAI } from "@ai-sdk/google";
+import { createOpenAI } from "@ai-sdk/openai";
+
+const getModel = () => {
+  const provider = env.AI_PROVIDER;
+  const apiKey = env.AI_API_KEY;
+  const model = env.AI_MODEL;
+
+  if (!provider || !apiKey || !model) {
+    throw new Error("AI_PROVIDER, AI_API_KEY, and AI_MODEL must be set to use the agent");
+  }
+
+  switch (provider) {
+    case "anthropic": {
+      const anthropic = createAnthropic({ apiKey });
+      return anthropic(model);
+    }
+    case "google": {
+      const google = createGoogleGenerativeAI({ apiKey });
+      return google(model);
+    }
+    case "openai": {
+      const openai = createOpenAI({ apiKey });
+      return openai(model);
+    }
+    default: {
+      throw new Error(`Provider not supported: ${provider}`);
+    }
+  }
+};
+
+export function createAgent(projectId: string) {
+  return new ToolLoopAgent({
+    model: getModel(),
+    instructions: buildSystemPrompt(projectId),
+    tools: {
+      executeClickHouse,
+    },
+    stopWhen: [stepCountIs(15)],
+  });
+}