feat: add anomaly detection pipeline

.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)

ARCHITECTURE.md

@@ -5,6 +5,7 @@
 ```
 apps/
   monitor-app        # Next.js dashboard & API
+  anomaly-worker     # Sidecar service for anomaly detection (new)
 
 packages/
   client-sdk              # Browser SDK for CWV collection
@@ -22,15 +23,24 @@ flowchart LR
     Ingest["POST /api/ingest<br>schema + rate limit"]
     Dashboard["Dashboard UI + auth"]
   end
+  subgraph Worker["Anomaly Worker"]
+    Poller["Poller (node-cron)"]
+    Notifier["Slack/Teams Notifier"]
+  end
   subgraph CH["ClickHouse"]
     Raw["cwv_events & custom_events<br>MergeTree, TTL 90d"]
     Agg["cwv_daily_aggregates<br>AggregatingMergeTree, 365d"]
+    Anom["v_cwv_anomalies (View)<br>Z-Score logic"]
   end
   SDK -->|"batched payload"| Ingest
   Ingest -->|"validated & enriched"| Raw
   Raw -->|"mv_cwv_daily_aggregates"| Agg
   Dashboard -->|"analytics queries"| Agg
   Dashboard -->|"drill-down"| Raw
+  Poller -->|"polls every hour"| Anom
+  Poller -->|"fetches project info"| Raw
+  Poller -->|"marks as notified"| Raw
+  Poller -->|"sends notifications"| Notifier
 ```
 
 ```mermaid
@@ -74,6 +84,16 @@ Lightweight browser SDK that collects CWV metrics (LCP, INP, CLS, TTFB, FCP), `$
 
 Shared TypeScript schemas for ingest payloads. Imported by both SDK and monitor app to prevent drift.
 
+### 3.4. Anomaly Worker (`apps/anomaly-worker`)
+
+A lightweight sidecar service that periodically polls ClickHouse for statistical anomalies (z_score > 3) and sends notifications.
+
+- **Stack:** Node.js, node-cron, ClickHouse client, pino
+- **Polling:** Scheduled every hour to detect new regressions
+- **State:** Tracks notified anomalies in `processed_anomalies` table to prevent duplicates
+- **Notifications:** Supports Slack and Microsoft Teams webhooks
+- **Deployment:** Dedicated sidecar container (see `docker/anomaly-worker.Dockerfile`)
+
 ## 4. Data Store
 
 **ClickHouse** — High-performance columnar database for analytics.

apps/anomaly-worker/.env.example

@@ -0,0 +1,20 @@
+# ClickHouse connection details
+CLICKHOUSE_HOST=localhost
+CLICKHOUSE_PORT=8123
+CLICKHOUSE_USER=default
+CLICKHOUSE_PASSWORD=secret
+CLICKHOUSE_DB=cwv_monitor
+AI_ANALYST_CLICKHOUSE_PASSWORD=
+BETTER_AUTH_SECRET=
+INITIAL_USER_EMAIL=
+INITIAL_USER_PASSWORD=
+INITIAL_USER_NAME=
+
+# Application
+LOG_LEVEL=info
+NODE_ENV=development
+AUTH_BASE_URL=http://localhost:3000
+
+# Anomaly Detection Webhooks (Optional)
+SLACK_WEBHOOK_URL=
+TEAMS_WEBHOOK_URL=

apps/anomaly-worker/.gitignore

@@ -0,0 +1 @@
+.env

apps/anomaly-worker/package.json

@@ -0,0 +1,41 @@
+{
+  "name": "anomaly-worker",
+  "version": "0.1.0",
+  "private": true,
+  "description": "A lightweight worker for detecting anomalies.",
+  "main": "dist/anomaly-worker/src/index.js",
+  "type": "module",
+  "scripts": {
+    "prebuild": "rm -rf dist",
+    "build": "tsc",
+    "start": "cross-env NODE_OPTIONS=\"-r dotenv/config\" tsx src/index.ts",
+    "dev": "tsx src/index.ts | pino-pretty",
+    "test": "vitest",
+    "test:watch": "vitest --watch"
+  },
+  "keywords": [],
+  "author": "",
+  "license": "ISC",
+  "dependencies": {
+    "@clickhouse/client": "^1.14.0",
+    "@t3-oss/env-nextjs": "^0.13.8",
+    "arktype": "^2.1.28",
+    "date-fns": "^4.1.0",
+    "node-cron": "^3.0.3",
+    "pino": "^10.1.0",
+    "remeda": "^2.32.0",
+    "waddler": "^0.1.1",
+    "zod": "^4.1.13",
+    "dotenv": "^16.4.5"
+  },
+  "devDependencies": {
+    "@types/node": "^24",
+    "@types/node-cron": "^3.0.11",
+    "pino-pretty": "^13.0.0",
+    "tsx": "^4.19.2",
+    "typescript": "^5.9.3",
+    "vitest": "^4.0.15",
+    "tsconfig-paths": "^4.2.0",
+    "cross-env": "^7.0.3"
+  }
+}

apps/anomaly-worker/src/env.ts

@@ -0,0 +1,54 @@
+import { createEnv } from "@t3-oss/env-nextjs";
+import { z } from "zod";
+
+const LOG_LEVELS = ["fatal", "error", "warn", "info", "debug", "trace", "silent"] as const;
+
+export const env = createEnv({
+  server: {
+    AUTH_BASE_URL: z.url(),
+    TRUST_PROXY: z.enum(["true", "false"]).default("false"),
+    CLICKHOUSE_HOST: z.string().min(1, "CLICKHOUSE_HOST is required"),
+    CLICKHOUSE_PORT: z.string().min(1, "CLICKHOUSE_PORT is required"),
+    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),
+    RATE_LIMIT_WINDOW_MS: z.coerce.number().positive().default(60_000),
+    MAX_LOGIN_ATTEMPTS: z.coerce.number().positive().default(5),
+    INITIAL_USER_EMAIL: z.email(),
+    INITIAL_USER_PASSWORD: z.string().min(8),
+    INITIAL_USER_NAME: z.string().min(3),
+    NODE_ENV: z.enum(["development", "test", "production"]).default("development"),
+    LOG_LEVEL: z.enum(LOG_LEVELS).default("info"),
+    SLACK_WEBHOOK_URL: z.url().optional(),
+    TEAMS_WEBHOOK_URL: z.url().optional(),
+  },
+  client: {},
+  runtimeEnv: {
+    AUTH_BASE_URL: process.env.AUTH_BASE_URL,
+    TRUST_PROXY: process.env.TRUST_PROXY,
+    CLICKHOUSE_HOST: process.env.CLICKHOUSE_HOST,
+    CLICKHOUSE_PORT: process.env.CLICKHOUSE_PORT,
+    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,
+    MIN_PASSWORD_SCORE: process.env.MIN_PASSWORD_SCORE,
+    RATE_LIMIT_WINDOW_MS: process.env.RATE_LIMIT_WINDOW_MS,
+    MAX_LOGIN_ATTEMPTS: process.env.MAX_LOGIN_ATTEMPTS,
+    INITIAL_USER_EMAIL: process.env.INITIAL_USER_EMAIL,
+    INITIAL_USER_PASSWORD: process.env.INITIAL_USER_PASSWORD,
+    INITIAL_USER_NAME: process.env.INITIAL_USER_NAME,
+    NODE_ENV: process.env.NODE_ENV,
+    LOG_LEVEL: process.env.LOG_LEVEL,
+    SLACK_WEBHOOK_URL: process.env.SLACK_WEBHOOK_URL,
+    TEAMS_WEBHOOK_URL: process.env.TEAMS_WEBHOOK_URL,
+  },
+});

apps/anomaly-worker/src/index.ts

@@ -0,0 +1,57 @@
+import pino from 'pino';
+import cron from 'node-cron';
+import { env } from "./env";
+
+const logger = pino({ name: 'anomaly-worker', level: env.LOG_LEVEL });
+
+let isRunning = false;
+let isShuttingDown = false;
+
+async function runDetectionCycle() {
+  if (isRunning || isShuttingDown) {
+    return;
+  }
+
+  isRunning = true;
+  logger.info("Starting anomaly detection cycle...");
+
+  try {
+    const { NotificationsService } = await import('@monitor-app/app/server/domain/notifications/service');
+    const service = new NotificationsService();
+
+    await service.notifyNewAnomalies();
+    logger.info("Anomaly detection cycle finished.");
+  } catch (error) {
+    logger.error({ err: error }, "Error during detection cycle.");
+  } finally {
+    isRunning = false;
+  }
+}
+
+const task = cron.schedule('30 * * * *', runDetectionCycle);
+logger.info("Anomaly worker scheduled (30 * * * *).");
+
+runDetectionCycle();
+
+process.on('SIGTERM', async () => {
+  isShuttingDown = true;
+  task.stop();
+
+  if (isRunning) {
+    logger.info("Waiting for the current detection cycle to complete...");
+
+    const shutdownTimeout = setTimeout(() => {
+      logger.error("Shutdown timed out; forcing exit.");
+      process.exit(1);
+    }, 20000);
+
+    while (isRunning) {
+      await new Promise(resolve => setTimeout(resolve, 500));
+    }
+
+    clearTimeout(shutdownTimeout);
+  }
+
+  logger.info("Shutdown complete.");
+  process.exit(0);
+});

apps/anomaly-worker/tsconfig.json

@@ -0,0 +1,18 @@
+{
+  "compilerOptions": {
+    "target": "ES2022",
+    "module": "ESNext",
+    "moduleResolution": "bundler",
+    "outDir": "./dist",
+    "strict": true,
+    "esModuleInterop": true,
+    "skipLibCheck": true,
+    "forceConsistentCasingInFileNames": true,
+    "paths": {
+      "@/*": ["./src/*", "../../apps/monitor-app/src/*"],
+      "@monitor-app/*": ["../monitor-app/src/*"]
+    }
+  },
+  "include": ["src/**/*.ts"],
+  "exclude": ["node_modules"]
+}

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}
@@ -23,3 +25,7 @@ INITIAL_USER_PASSWORD=password
 INITIAL_USER_NAME=User
 
 MIN_PASSWORD_SCORE=2
+
+# Anomaly Detection Webhooks (Optional)
+# SLACK_WEBHOOK_URL=https://hooks.slack.com/services/...
+# TEAMS_WEBHOOK_URL=https://outlook.office.com/webhook/...

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
@@ -17,5 +19,6 @@ INITIAL_USER_EMAIL=initial@example.com
 INITIAL_USER_PASSWORD=password1234
 INITIAL_USER_NAME=Initial User
 
+
 # Logging
 LOG_LEVEL=debug

apps/monitor-app/AI_AGENT_GUIDE.md

@@ -0,0 +1,88 @@
+# AI Agent Architecture & Integration Guide
+
+This document outlines the bidirectional AI monitoring system for the CWV Monitor. It explains how the **Monitor App** acts as the analytical engine for **AI Agent (Slack/Teams Bot)**.
+
+## 1. System Components
+
+### Monitor App (`apps/monitor-app`)
+The "Source of Truth" and Data Engine.
+- **ClickHouse**: Stores all raw performance events and pre-aggregated metrics.
+- **AiBridge**: A domain service that provides safe, high-level methods for the AI Agent to query data.
+- **Schema Catalog**: A YAML definition (`clickhouse/catalog.yml`) that describes the database to the LLM.
+
+### AI Agent
+The interaction layer (Slack/Teams).
+- **Inbound Handling**: Receives webhooks and message events from Slack/Teams.
+- **LLM Brain**: Uses the `AiBridge` from the Monitor App to fetch context and execute diagnostic SQL.
+- **Conversation State**: Tracks thread IDs and maps them to specific `anomaly_id`s.
+
+### Anomaly Worker (`apps/anomaly-worker`)
+The proactive trigger.
+- Runs on a cron schedule.
+- Scans ClickHouse for statistical anomalies (Z-Score > 3).
+- Triggers the initial outbound notification via the Monitor App's `NotificationsService`.
+
+---
+
+## 2. Security Model
+
+We follow the principle of **Least Privilege** for AI interactions:
+
+- **Restricted User**: The Agent uses `ai_analyst_user` in ClickHouse.
+- **Role-Based Access**: The `r_ai_analyst` role is limited to `SELECT` on core tables and `INSERT/UPDATE` only on the `processed_anomalies` audit table.
+- **Resource Guardrails**: The `ai_analyst_profile` sets hard limits on memory (2GB), execution time (15s), and rows read (100M) to prevent the LLM from generating "expensive" or runaway queries.
+
+---
+
+## 3. Capabilities ("What can be done")
+
+### Outbound Alerts
+When an anomaly is detected, the `NotificationsService` sends a rich payload to Slack/Teams:
+- **Investigate Button**: Links to the Monitor App Web UI.
+- **Chat with AI Button**: Deep-links to the AI Agent to start a diagnostic conversation.
+- **Metadata**: Includes `anomalyId`, `projectId`, and `metricName` so the Agent knows exactly what to talk about.
+
+### Deep-Dive Context
+The AI Agent can call `aiBridge.getAnomalyContext(anomalyId)` to receive a "Context Bundle":
+1. **Anomaly Record**: Specifics about the z-score and metric.
+2. **Project Info**: Domain and configuration.
+3. **24h Trend**: Pre-fetched hourly averages for the last 24 hours.
+4. **Schema Reference**: The `catalog.yml` content to help the LLM generate its own SQL.
+
+### Dynamic SQL Execution
+The Agent can ask the LLM to "Verify if this happens on other routes." The LLM generates a SQL query, and the Agent executes it via `aiBridge.executeSql(query, params)`.
+
+---
+
+## 4. Integration Pattern
+
+To use the Monitor App domain in your  AI application:
+
+```typescript
+import { aiBridge } from "@monitor-app/app/server/domain/ai/service";
+
+// 1. When a user clicks "Chat with AI" in Slack
+async function handleChatStart(anomalyId: string) {
+  // Get everything the LLM needs to know in one call
+  const context = await aiBridge.getAnomalyContext(anomalyId);
+
+  // Provide this context to your LLM system prompt
+  const systemPrompt = `
+    You are a Performance Expert. 
+    Analyze this anomaly: ${JSON.stringify(context.anomaly)}
+    Database Schema: ${context.schemaReference}
+    ...
+  `;
+}
+
+// 2. When the LLM wants to fetch more data
+async function onLlmToolUse(generatedSql: string) {
+  const results = await aiBridge.executeSql(generatedSql);
+  return results;
+}
+```
+
+## 5. Files Reference
+- **Schema Catalog**: `apps/monitor-app/clickhouse/catalog.yml`
+- **AI Domain Logic**: `apps/monitor-app/src/app/server/domain/ai/service.ts`
+- **Notification Logic**: `apps/monitor-app/src/app/server/domain/notifications/`