feat(sase): VIN decode anomaly detection + Telegram alerts

Adds the active half of Phase 8: dashboard was passive, this pushes
when something breaks. Targets MTTD < 10 minutes from the PRD.

Detection (panel-web)
- detectVinAnomalies() compares a 15-minute current window against a
  baseline built from the same 15-minute slot across the previous 7
  days (same-hour-of-day, average across days that had ≥ 1 sample).
- Five hits, two severities:
  * success_rate_drop — high at ≥ 5pp drop, critical at ≥ 15pp
  * p95_latency_spike — high at 2× baseline (and > 1s), critical at 3×
  * volume_drop — high at ≥ 80% below baseline, critical at ≥ 95%
  * volume_spike — high at ≥ 5× baseline
  * timeout_dominance — high when ≥ 50% of failures are timeouts
- Minimum-volume guards on both current and baseline so quiet hours
  don't generate noise (MIN_CURRENT_VOLUME=5, MIN_BASELINE_VOLUME=10).
- Each anomaly carries a 15-min-bucket dedupe key — same anomaly type
  fires at most once per bucket regardless of cron cadence.

Endpoint
- GET /api/internal/vin-anomaly-check, gated by x-internal-worker-token
  header (constant-time compare against INTERNAL_WORKER_TOKEN env).
  Returns { current, baseline, anomalies[] }.

Worker
- New job vin-anomaly-detect, BullMQ scheduler */5 * * * *.
  Fetches the panel endpoint, then for each anomaly calls
  alertVinAnomaly() — sendTelegram with the bucket dedupe key, so
  Redis SETEX NX dedupes across the 1h cooldown window.
- alertVinAnomaly() in worker/lib/telegram.ts formats the message with
  severity icon + baseline/observed/volume context + dashboard link.
- runVinAnomalyDetect returns { ok, checked, anomalies, alertsFired,
  alertsDeduped }; pipeline logs only when something happened or the
  check failed.

Env
- INTERNAL_WORKER_TOKEN set on both panel-web and panel-worker
  (32-byte hex, generated in Coolify).
- PANEL_INTERNAL_URL on panel-worker → coolify-network UUID hostname
  for panel-web, no Cloudflare/Tailscale hop on internal calls.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

apps/web/src/app/api/internal/vin-anomaly-check/route.ts

@@ -0,0 +1,25 @@
+import { NextResponse } from "next/server";
+import { timingSafeEqual } from "node:crypto";
+import { detectVinAnomalies } from "@/lib/sase/vin-anomaly";
+
+export const dynamic = "force-dynamic";
+
+const INTERNAL_TOKEN = process.env.INTERNAL_WORKER_TOKEN ?? "";
+
+function validToken(req: Request): boolean {
+  if (!INTERNAL_TOKEN) return false;
+  const provided = req.headers.get("x-internal-worker-token") ?? "";
+  if (!provided) return false;
+  const a = Buffer.from(provided);
+  const b = Buffer.from(INTERNAL_TOKEN);
+  if (a.length !== b.length) return false;
+  return timingSafeEqual(a, b);
+}
+
+export async function GET(req: Request) {
+  if (!validToken(req)) {
+    return NextResponse.json({ ok: false, error: "unauthorized" }, { status: 401 });
+  }
+  const result = await detectVinAnomalies();
+  return NextResponse.json({ ok: true, ...result });
+}

apps/web/src/lib/sase/vin-anomaly.ts

@@ -0,0 +1,250 @@
+import { saseDb } from "@/lib/db-sase";
+
+const CURRENT_WINDOW_MIN = 15;
+const BASELINE_DAYS = 7;
+const MIN_BASELINE_VOLUME = 10; // skip if baseline volume below this — too noisy
+const MIN_CURRENT_VOLUME = 5; // skip checks that need volume to be meaningful
+
+export type AnomalyType =
+  | "success_rate_drop"
+  | "error_rate_spike"
+  | "p95_latency_spike"
+  | "volume_drop"
+  | "volume_spike"
+  | "timeout_dominance";
+
+export type AnomalyHit = {
+  type: AnomalyType;
+  severity: "high" | "critical";
+  message: string;
+  baseline: number;
+  observed: number;
+  current_volume: number;
+  baseline_volume: number;
+  detected_at: string;
+  // unique key for Telegram dedupe — same anomaly type within cooldown window
+  dedupe_key: string;
+};
+
+type WindowStats = {
+  total: number;
+  succeeded: number;
+  failed: number;
+  successRate: number;
+  errorRate: number;
+  p95: number | null;
+  avgMs: number | null;
+  timeouts: number;
+};
+
+async function windowStats(start: Date, end: Date): Promise<WindowStats> {
+  const rows = await saseDb.$queryRaw<
+    Array<{
+      total: bigint;
+      succeeded: bigint;
+      failed: bigint;
+      p95: number | null;
+      avg_ms: number | null;
+      timeouts: bigint;
+    }>
+  >`
+    SELECT
+      count(*) AS total,
+      count(*) FILTER (WHERE success = true) AS succeeded,
+      count(*) FILTER (WHERE success = false) AS failed,
+      percentile_cont(0.95) WITHIN GROUP (ORDER BY response_time_ms)
+        FILTER (WHERE success = true AND response_time_ms IS NOT NULL)::int AS p95,
+      avg(response_time_ms) FILTER (WHERE success = true AND response_time_ms IS NOT NULL)::int AS avg_ms,
+      count(*) FILTER (
+        WHERE success = false
+          AND (
+            error_message ILIKE '%budget%'
+            OR error_message ILIKE '%timeout%'
+            OR (timings->>'aborted')::boolean = true
+          )
+      ) AS timeouts
+    FROM query_logs
+    WHERE created_at >= ${start} AND created_at < ${end}
+  `;
+  const r = rows[0] ?? {
+    total: 0n,
+    succeeded: 0n,
+    failed: 0n,
+    p95: null,
+    avg_ms: null,
+    timeouts: 0n,
+  };
+  const total = Number(r.total);
+  const succeeded = Number(r.succeeded);
+  const failed = Number(r.failed);
+  return {
+    total,
+    succeeded,
+    failed,
+    successRate: total > 0 ? succeeded / total : 0,
+    errorRate: total > 0 ? failed / total : 0,
+    p95: r.p95,
+    avgMs: r.avg_ms,
+    timeouts: Number(r.timeouts),
+  };
+}
+
+/**
+ * Baseline = same hour-of-day across the previous BASELINE_DAYS days, in
+ * CURRENT_WINDOW_MIN-wide buckets aligned to the current window's offset.
+ * For "now is 14:32–14:47", we pull 14:32–14:47 from each of yesterday and
+ * the prior 6 days. Average the per-day stats.
+ */
+async function baselineStats(currentEnd: Date): Promise<WindowStats> {
+  const samples: WindowStats[] = [];
+  for (let dayOffset = 1; dayOffset <= BASELINE_DAYS; dayOffset++) {
+    const end = new Date(currentEnd.getTime() - dayOffset * 24 * 60 * 60_000);
+    const start = new Date(end.getTime() - CURRENT_WINDOW_MIN * 60_000);
+    const s = await windowStats(start, end);
+    if (s.total >= MIN_BASELINE_VOLUME / BASELINE_DAYS) samples.push(s);
+  }
+  if (samples.length === 0) {
+    return {
+      total: 0,
+      succeeded: 0,
+      failed: 0,
+      successRate: 0,
+      errorRate: 0,
+      p95: null,
+      avgMs: null,
+      timeouts: 0,
+    };
+  }
+  const total = samples.reduce((a, b) => a + b.total, 0);
+  const succeeded = samples.reduce((a, b) => a + b.succeeded, 0);
+  const failed = samples.reduce((a, b) => a + b.failed, 0);
+  const p95Values = samples.map((s) => s.p95).filter((v): v is number => v != null);
+  const avgValues = samples.map((s) => s.avgMs).filter((v): v is number => v != null);
+  const timeouts = samples.reduce((a, b) => a + b.timeouts, 0);
+  return {
+    total,
+    succeeded,
+    failed,
+    successRate: total > 0 ? succeeded / total : 0,
+    errorRate: total > 0 ? failed / total : 0,
+    p95: p95Values.length ? Math.round(p95Values.reduce((a, b) => a + b, 0) / p95Values.length) : null,
+    avgMs: avgValues.length ? Math.round(avgValues.reduce((a, b) => a + b, 0) / avgValues.length) : null,
+    timeouts,
+  };
+}
+
+export async function detectVinAnomalies(): Promise<{
+  current: WindowStats;
+  baseline: WindowStats;
+  anomalies: AnomalyHit[];
+}> {
+  const now = new Date();
+  const currentStart = new Date(now.getTime() - CURRENT_WINDOW_MIN * 60_000);
+
+  const [current, baseline] = await Promise.all([
+    windowStats(currentStart, now),
+    baselineStats(now),
+  ]);
+
+  const hits: AnomalyHit[] = [];
+  const ts = now.toISOString();
+  const bucket = Math.floor(now.getTime() / (CURRENT_WINDOW_MIN * 60_000));
+
+  // 1. Success rate drop: >= 5 percentage points below baseline (high), >= 15pp (critical)
+  if (
+    current.total >= MIN_CURRENT_VOLUME &&
+    baseline.total >= MIN_BASELINE_VOLUME &&
+    baseline.successRate > 0
+  ) {
+    const dropPp = (baseline.successRate - current.successRate) * 100;
+    if (dropPp >= 5) {
+      hits.push({
+        type: "success_rate_drop",
+        severity: dropPp >= 15 ? "critical" : "high",
+        message: `Decode başarı oranı ${pct(baseline.successRate)} → ${pct(current.successRate)} (−${dropPp.toFixed(1)}pp, son ${CURRENT_WINDOW_MIN}dk)`,
+        baseline: baseline.successRate,
+        observed: current.successRate,
+        current_volume: current.total,
+        baseline_volume: baseline.total,
+        detected_at: ts,
+        dedupe_key: `vin:success_rate_drop:${bucket}`,
+      });
+    }
+  }
+
+  // 2. P95 latency spike: 2x baseline + over an absolute floor (1s)
+  if (current.p95 != null && baseline.p95 != null && current.p95 > 1000) {
+    const ratio = current.p95 / baseline.p95;
+    if (ratio >= 2) {
+      hits.push({
+        type: "p95_latency_spike",
+        severity: ratio >= 3 ? "critical" : "high",
+        message: `P95 yanıt ${baseline.p95}ms → ${current.p95}ms (${ratio.toFixed(1)}× baseline)`,
+        baseline: baseline.p95,
+        observed: current.p95,
+        current_volume: current.total,
+        baseline_volume: baseline.total,
+        detected_at: ts,
+        dedupe_key: `vin:p95_spike:${bucket}`,
+      });
+    }
+  }
+
+  // 3. Volume drop: >= 80% below baseline (high), >= 95% (critical) — outage signal
+  if (baseline.total >= MIN_BASELINE_VOLUME) {
+    const expectedPerWindow = baseline.total / BASELINE_DAYS;
+    if (expectedPerWindow >= 1 && current.total / expectedPerWindow <= 0.2) {
+      const dropPct = (1 - current.total / expectedPerWindow) * 100;
+      hits.push({
+        type: "volume_drop",
+        severity: dropPct >= 95 ? "critical" : "high",
+        message: `Sorgu hacmi ${expectedPerWindow.toFixed(0)} → ${current.total} (−${dropPct.toFixed(0)}%, son ${CURRENT_WINDOW_MIN}dk)`,
+        baseline: expectedPerWindow,
+        observed: current.total,
+        current_volume: current.total,
+        baseline_volume: baseline.total,
+        detected_at: ts,
+        dedupe_key: `vin:volume_drop:${bucket}`,
+      });
+    }
+  }
+
+  // 4. Volume spike: 5x baseline — fraud or viral
+  if (baseline.total >= MIN_BASELINE_VOLUME) {
+    const expectedPerWindow = baseline.total / BASELINE_DAYS;
+    if (expectedPerWindow >= 1 && current.total / expectedPerWindow >= 5) {
+      hits.push({
+        type: "volume_spike",
+        severity: "high",
+        message: `Sorgu hacmi ${expectedPerWindow.toFixed(0)} → ${current.total} (${(current.total / expectedPerWindow).toFixed(1)}×, son ${CURRENT_WINDOW_MIN}dk)`,
+        baseline: expectedPerWindow,
+        observed: current.total,
+        current_volume: current.total,
+        baseline_volume: baseline.total,
+        detected_at: ts,
+        dedupe_key: `vin:volume_spike:${bucket}`,
+      });
+    }
+  }
+
+  // 5. Timeout dominance: > 50% of failures are timeouts AND failure volume > 5
+  if (current.failed >= 5 && current.timeouts / Math.max(current.failed, 1) >= 0.5) {
+    hits.push({
+      type: "timeout_dominance",
+      severity: "high",
+      message: `Hataların ${pct(current.timeouts / current.failed)}'i timeout (${current.timeouts}/${current.failed}, son ${CURRENT_WINDOW_MIN}dk)`,
+      baseline: 0.2,
+      observed: current.timeouts / current.failed,
+      current_volume: current.failed,
+      baseline_volume: 0,
+      detected_at: ts,
+      dedupe_key: `vin:timeout_dominance:${bucket}`,
+    });
+  }
+
+  return { current, baseline, anomalies: hits };
+}
+
+function pct(v: number): string {
+  return `${(v * 100).toFixed(1)}%`;
+}

apps/worker/src/jobs/vin-anomaly.ts

@@ -0,0 +1,100 @@
+import { alertVinAnomaly, isTelegramConfigured } from "../lib/telegram";
+
+const PANEL_BASE =
+  process.env.PANEL_INTERNAL_URL ??
+  process.env.PANEL_PUBLIC_URL ??
+  "http://panel-web:3000";
+
+const WORKER_TOKEN = process.env.INTERNAL_WORKER_TOKEN ?? "";
+
+type AnomalyHit = {
+  type: string;
+  severity: "high" | "critical";
+  message: string;
+  baseline: number;
+  observed: number;
+  current_volume: number;
+  baseline_volume: number;
+  detected_at: string;
+  dedupe_key: string;
+};
+
+type CheckResponse = {
+  ok: boolean;
+  error?: string;
+  current?: { total: number; successRate: number };
+  baseline?: { total: number; successRate: number };
+  anomalies?: AnomalyHit[];
+};
+
+export async function runVinAnomalyDetect(): Promise<{
+  ok: boolean;
+  checked: boolean;
+  anomalies: number;
+  alertsFired: number;
+  alertsDeduped: number;
+  reason?: string;
+}> {
+  if (!WORKER_TOKEN) {
+    return { ok: false, checked: false, anomalies: 0, alertsFired: 0, alertsDeduped: 0, reason: "INTERNAL_WORKER_TOKEN not set" };
+  }
+  if (!isTelegramConfigured()) {
+    return { ok: false, checked: false, anomalies: 0, alertsFired: 0, alertsDeduped: 0, reason: "telegram not configured" };
+  }
+
+  let res: Response;
+  try {
+    res = await fetch(`${PANEL_BASE}/api/internal/vin-anomaly-check`, {
+      headers: {
+        "x-internal-worker-token": WORKER_TOKEN,
+        "cache-control": "no-store",
+      },
+    });
+  } catch (e) {
+    return {
+      ok: false,
+      checked: false,
+      anomalies: 0,
+      alertsFired: 0,
+      alertsDeduped: 0,
+      reason: `fetch failed: ${(e as Error).message}`,
+    };
+  }
+  if (!res.ok) {
+    return {
+      ok: false,
+      checked: false,
+      anomalies: 0,
+      alertsFired: 0,
+      alertsDeduped: 0,
+      reason: `panel returned ${res.status}`,
+    };
+  }
+  const data = (await res.json()) as CheckResponse;
+  if (!data.ok || !data.anomalies) {
+    return {
+      ok: !!data.ok,
+      checked: true,
+      anomalies: 0,
+      alertsFired: 0,
+      alertsDeduped: 0,
+      reason: data.error,
+    };
+  }
+
+  const panelPublic = process.env.PANEL_PUBLIC_URL ?? "https://sp.semih.ai";
+  let fired = 0;
+  let deduped = 0;
+  for (const a of data.anomalies) {
+    const r = await alertVinAnomaly({ panelUrl: panelPublic, ...a });
+    if (r.ok && r.deduped) deduped++;
+    else if (r.ok) fired++;
+  }
+  return {
+    ok: true,
+    checked: true,
+    anomalies: data.anomalies.length,
+    alertsFired: fired,
+    alertsDeduped: deduped,
+  };
+}

apps/worker/src/lib/telegram.ts

@@ -119,6 +119,32 @@ export function alertBudgetCap(opts: {
   return sendTelegram({ text, dedupeKey: `budget:${opts.state}:${day}` });
 }
 
+export type VinAnomalyAlert = {
+  type: string;
+  severity: "high" | "critical";
+  message: string;
+  baseline: number;
+  observed: number;
+  current_volume: number;
+  baseline_volume: number;
+  detected_at: string;
+  dedupe_key: string;
+};
+
+export function alertVinAnomaly(
+  opts: { panelUrl: string } & VinAnomalyAlert,
+): Promise<TelegramSendResult> {
+  const icon = opts.severity === "critical" ? "🔴" : "🟡";
+  const text = [
+    `${icon} <b>VIN decode anomaly</b>: ${opts.type}`,
+    escapeHtml(opts.message),
+    `Volume: now ${opts.current_volume} · baseline ${opts.baseline_volume}`,
+    ``,
+    `<a href="${opts.panelUrl}/projects/sase/vin-decode">VIN Decode dashboard</a>`,
+  ].join("\n");
+  return sendTelegram({ text, dedupeKey: opts.dedupe_key });
+}
+
 function escapeHtml(s: string): string {
   return s.replace(/&/g, "&amp;").replace(/</g, "&lt;").replace(/>/g, "&gt;");
 }

apps/worker/src/schedulers/pipeline.ts

@@ -9,6 +9,7 @@ import { runGithubSync } from "../jobs/github-sync";
 import { runRetention } from "../jobs/retention";
 import { runEvalSet } from "../jobs/eval-run";
 import { runDailyBrief } from "../jobs/daily-brief";
+import { runVinAnomalyDetect } from "../jobs/vin-anomaly";
 
 const QUEUE = "insight-pipeline";
 
@@ -77,6 +78,15 @@ async function runJob(job: Job) {
       console.log(`[pipeline] daily-brief sent=${res.sent}${res.reason ? ` reason=${res.reason}` : ""}`);
       return res;
     }
+    case "vin-anomaly-detect": {
+      const res = await runVinAnomalyDetect();
+      if (res.anomalies > 0 || !res.ok) {
+        console.log(
+          `[pipeline] vin-anomaly checked=${res.checked} anomalies=${res.anomalies} fired=${res.alertsFired} deduped=${res.alertsDeduped}${res.reason ? ` reason=${res.reason}` : ""}`,
+        );
+      }
+      return res;
+    }
     default:
       return { ok: false, error: `unknown job ${job.name}` };
   }
@@ -123,6 +133,11 @@ export async function startInsightPipeline() {
     { pattern: "0 5 * * *" }, // 08:00 Europe/Istanbul = 05:00 UTC
     { name: "daily-brief", data: {}, opts: { removeOnComplete: 30, removeOnFail: 15 } },
   );
+  await queue.upsertJobScheduler(
+    "vin-anomaly-detect",
+    { pattern: "*/5 * * * *" },
+    { name: "vin-anomaly-detect", data: {}, opts: { removeOnComplete: 30, removeOnFail: 15 } },
+  );
 
   new Worker(QUEUE, runJob, {
     connection: redis,
@@ -131,6 +146,6 @@ export async function startInsightPipeline() {
     stalledInterval: 60_000,
   });
   console.log(
-    "[pipeline] armed: posthog-ingest@*/5min, tag-sessions@*/2min, compress-sessions@*/3min, analyze@*/4min, validation@05:00, github-sync@*/10min, retention@04:15, daily-brief@05:00",
+    "[pipeline] armed: posthog-ingest@*/5min, tag-sessions@*/2min, compress-sessions@*/3min, analyze@*/4min, validation@05:00, github-sync@*/10min, retention@04:15, daily-brief@05:00, vin-anomaly-detect@*/5min",
   );
 }