feat(insights): Telegram alerts + daily brief push notifications

apps/worker/src/lib/telegram.ts: - sendTelegram() with Redis-backed dedupe (NX SETEX, 1h TTL) - Helpers: alertP0Insight, alertRegression, alertSanitizationAnomaly, alertBudgetCap Wired into: - analyze.ts: P0/P1 insight creation → instant alert (dedupe per insight_id); budget guard halt → daily cap alert (dedupe per state per day) - validation.ts: regression detected (≥3 sessions w/ same fingerprint after shippedAt) → alert (dedupe per insight_id) - compress-sessions.ts: sanitization anomaly (>500 tokens, 0 PII matches) → alert (dedupe per session_id) — possible PII leak warning Daily Brief (jobs/daily-brief.ts): - Cron @05:00 UTC (= 08:00 Europe/Istanbul) - 24h: sessions/insights/cost/cache-hit + 3 top priorities + 7d shipped/validated/regressed - POST /api/insights/brief/send for manual trigger / smoke test Env: TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID, PANEL_PUBLIC_URL (Coolify both apps). Bot: @Pl24_mitm_bot (AiFactory), chat 7840804807. Source: airflow3 monitoring DAG. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-14 11:36:10 +00:00
parent 74f0ff4935
commit a713505f44
8 changed files with 298 additions and 3 deletions
--- a/apps/worker/src/jobs/analyze.ts
+++ b/apps/worker/src/jobs/analyze.ts
@@ -4,6 +4,9 @@ import { checkBudget } from "../lib/budget";
 import { pickPromptTag } from "../lib/prompts";
 import { validate } from "../lib/json-validate";
 import { getText } from "../lib/minio";
+import { alertP0Insight, alertBudgetCap } from "../lib/telegram";
+
+const PANEL_URL = process.env.PANEL_PUBLIC_URL ?? "https://sp.semih.ai";

 const ANALYZE_BATCH = Number(process.env.INSIGHT_ANALYZE_BATCH ?? "8");
 const CACHE_TTL_HOURS = Number(process.env.INSIGHT_INSIGHT_CACHE_HOURS ?? "6");
@@ -23,6 +26,13 @@ export async function runAnalyze(): Promise<AnalyzeResult> {
  const budget = await checkBudget();
  if (!budget.allow) {
    console.log(`[analyze] budget=${budget.state} ${budget.reason}`);
+    // Fire a Telegram alert once per day per state.
+    void alertBudgetCap({
+      state: budget.state,
+      todayUsd: budget.todayUsd,
+      monthUsd: budget.monthUsd,
+      panelUrl: PANEL_URL,
+    });
    return { analyzed: 0, skipped: 0, failed: 0, costUsd: 0, budgetState: budget.state };
  }

@@ -267,7 +277,7 @@ export async function runAnalyze(): Promise<AnalyzeResult> {
      const uniqueUsers = new Set(group.map((g) => g.userIdHash).filter(Boolean));
      const firstSeen = group.reduce((a, b) => (a.startedAt < b.startedAt ? a : b)).startedAt;
      const lastSeen = group.reduce((a, b) => (a.startedAt > b.startedAt ? a : b)).startedAt;
-      await prisma.insight.create({
+      const created = await prisma.insight.create({
        data: {
          projectKey: PROJECT_KEY,
          type,
@@ -289,6 +299,17 @@ export async function runAnalyze(): Promise<AnalyzeResult> {
          sourceCostUsd: result.cost.totalUsd,
        },
      });
+      // Alert on P0 / P1 critical insights immediately.
+      if (sev === "P0" || sev === "P1") {
+        void alertP0Insight({
+          insightId: created.id,
+          title,
+          severity: sev,
+          type,
+          fingerprint: s.fingerprint,
+          panelUrl: PANEL_URL,
+        });
+      }
    }

    // Mark every session in the group as analyzed (bundle covers them all).
--- a/apps/worker/src/jobs/compress-sessions.ts
+++ b/apps/worker/src/jobs/compress-sessions.ts
@@ -4,6 +4,9 @@ import { compressSnapshots, parseSnapshotBlob } from "../lib/compress";
 import { putText } from "../lib/minio";
 import { loadEnrichment } from "../lib/enrich";
 import { getCachedGroup } from "../lib/posthog-cache";
+import { alertSanitizationAnomaly } from "../lib/telegram";
+
+const PANEL_URL = process.env.PANEL_PUBLIC_URL ?? "https://sp.semih.ai";

 const COMPRESSION_BUCKET = process.env.INSIGHT_COMPRESSED_BUCKET ?? "insight-compressed";

@@ -107,6 +110,11 @@ export async function runCompressSessions(): Promise<{ compressed: number; faile

      if (anomaly) {
        console.warn(`[compress] sanitization-anomaly session=${s.id} tokens=${out.tokenCountEstimate}`);
+        void alertSanitizationAnomaly({
+          sessionId: s.id,
+          tokens: out.tokenCountEstimate,
+          panelUrl: PANEL_URL,
+        });
      }
      compressed++;
    } catch (e) {
--- a/apps/worker/src/jobs/daily-brief.ts
+++ b/apps/worker/src/jobs/daily-brief.ts
@@ -0,0 +1,102 @@
+import { prisma } from "../db";
+import { sendTelegram, isTelegramConfigured } from "../lib/telegram";
+
+const PANEL_URL = process.env.PANEL_PUBLIC_URL ?? "https://sp.semih.ai";
+
+export type DailyBriefResult = { sent: boolean; reason?: string };
+
+// Daily Brief — 08:00 local cron. Sends a summary of the last 24h + top
+// priorities to Telegram. Cheap (~1 LLM-free DB-only query).
+export async function runDailyBrief(): Promise<DailyBriefResult> {
+  if (!isTelegramConfigured()) return { sent: false, reason: "telegram_not_configured" };
+
+  const now = new Date();
+  const yesterdayStart = new Date(now.getTime() - 24 * 3600_000);
+  const weekAgo = new Date(now.getTime() - 7 * 24 * 3600_000);
+
+  const [
+    sessionsProcessed,
+    insightsCreated,
+    bySev,
+    costAgg,
+    cacheAgg,
+    topPriorities,
+    validatedThisWeek,
+    regressedThisWeek,
+    shippedThisWeek,
+  ] = await Promise.all([
+    prisma.sessionMeta.count({
+      where: { processedAt: { gte: yesterdayStart }, status: { in: ["analyzed", "compressed", "discarded", "tagged"] } },
+    }),
+    prisma.insight.count({ where: { createdAt: { gte: yesterdayStart } } }),
+    prisma.insight.groupBy({
+      by: ["severity"],
+      where: { createdAt: { gte: yesterdayStart } },
+      _count: { severity: true },
+    }),
+    prisma.costLedger.aggregate({
+      where: { createdAt: { gte: yesterdayStart } },
+      _sum: { costTotalUsd: true },
+      _count: true,
+    }),
+    prisma.costLedger.aggregate({
+      where: { createdAt: { gte: yesterdayStart } },
+      _sum: { tokensInputCacheHit: true, tokensInputCacheMiss: true },
+    }),
+    prisma.insight.findMany({
+      where: { status: { in: ["new", "triaged", "in_backlog", "regressed"] } },
+      orderBy: [{ founderPriority: "desc" }, { priorityScore: "desc" }],
+      take: 3,
+    }),
+    prisma.insight.count({ where: { validatedAt: { gte: weekAgo } } }),
+    prisma.insight.count({ where: { regressionDetected: true, updatedAt: { gte: weekAgo } } }),
+    prisma.insight.count({ where: { shippedAt: { gte: weekAgo } } }),
+  ]);
+
+  const sevCounts: Record<string, number> = {};
+  for (const row of bySev) sevCounts[row.severity] = row._count.severity;
+  const sevHint = ["P0", "P1", "P2", "P3"]
+    .filter((s) => sevCounts[s])
+    .map((s) => `${sevCounts[s]} ${s}`)
+    .join(" · ");
+
+  const cost = Number(costAgg._sum.costTotalUsd ?? 0);
+  const calls = Number(costAgg._count ?? 0);
+  const hit = Number(cacheAgg._sum.tokensInputCacheHit ?? 0);
+  const miss = Number(cacheAgg._sum.tokensInputCacheMiss ?? 0);
+  const hitRate = hit + miss > 0 ? hit / (hit + miss) : 0;
+
+  const lines: string[] = [
+    `🌅 <b>Süper Panel · Daily Brief</b>`,
+    `<i>${now.toISOString().slice(0, 10)}</i>`,
+    ``,
+    `<b>Last 24h</b>`,
+    `• Sessions processed: <b>${sessionsProcessed}</b>`,
+    `• Insights produced: <b>${insightsCreated}</b>${sevHint ? ` (${sevHint})` : ""}`,
+    `• Cost: <b>$${cost.toFixed(4)}</b> · ${calls} calls`,
+    `• Cache hit: <b>${(hitRate * 100).toFixed(0)}%</b>`,
+    ``,
+  ];
+
+  if (topPriorities.length) {
+    lines.push(`<b>Top priorities</b>`);
+    for (const i of topPriorities) {
+      const sev = i.severity === "P0" || i.severity === "P1" ? "🔴" : i.severity === "P2" ? "🟡" : "⚪";
+      lines.push(`${sev} <b>${i.severity}</b> · ${escapeHtml(i.title).slice(0, 80)}`);
+    }
+    lines.push(``);
+  }
+
+  lines.push(`<b>Last 7 days</b>`);
+  lines.push(`• Shipped: ${shippedThisWeek} · Validated: ${validatedThisWeek} · Regressed: ${regressedThisWeek}${regressedThisWeek > 0 ? " ⚠️" : ""}`);
+  lines.push(``);
+  lines.push(`<a href="${PANEL_URL}/insights">Inbox</a> · <a href="${PANEL_URL}/insights/brief">Full brief</a> · <a href="${PANEL_URL}/insights/costs">Costs</a>`);
+
+  // Skip dedupe — we want this to fire daily even if content is similar.
+  const res = await sendTelegram({ text: lines.join("\n") });
+  return { sent: res.ok, reason: res.error };
+}
+
+function escapeHtml(s: string): string {
+  return s.replace(/&/g, "&amp;").replace(/</g, "&lt;").replace(/>/g, "&gt;");
+}
--- a/apps/worker/src/jobs/validation.ts
+++ b/apps/worker/src/jobs/validation.ts
@@ -1,6 +1,8 @@
 import { prisma } from "../db";
+import { alertRegression } from "../lib/telegram";

 const REGRESSION_THRESHOLD = Number(process.env.INSIGHT_REGRESSION_THRESHOLD ?? "3");
+const PANEL_URL = process.env.PANEL_PUBLIC_URL ?? "https://sp.semih.ai";

 export type ValidationResult = {
  shippedScanned: number;
@@ -52,6 +54,13 @@ export async function runValidation(): Promise<ValidationResult> {
      console.log(
        `[validation] regression on insight ${insight.id} (${insight.fingerprint.slice(0, 12)}): ${newOccurrences} new occurrences`,
      );
+      void alertRegression({
+        insightId: insight.id,
+        title: insight.title,
+        fingerprint: insight.fingerprint,
+        occurrenceCount: newOccurrences,
+        panelUrl: PANEL_URL,
+      });
      regressed++;
      continue;
    }
--- a/apps/worker/src/lib/telegram.ts
+++ b/apps/worker/src/lib/telegram.ts
@@ -0,0 +1,124 @@
+// Telegram push notifications. Fire-and-forget; failures swallowed.
+// Throttling: same key cannot fire more often than TELEGRAM_DEDUPE_TTL_SECONDS (per Redis SETEX).
+
+import { redis } from "../redis";
+
+const TOKEN = process.env.TELEGRAM_BOT_TOKEN ?? "";
+const CHAT_ID = process.env.TELEGRAM_CHAT_ID ?? "";
+const DEDUPE_TTL_SECONDS = Number(process.env.TELEGRAM_DEDUPE_TTL_SECONDS ?? "3600");
+
+export function isTelegramConfigured(): boolean {
+  return Boolean(TOKEN && CHAT_ID);
+}
+
+export type TelegramSendResult = { ok: boolean; deduped?: boolean; error?: string };
+
+export async function sendTelegram(opts: {
+  text: string;
+  dedupeKey?: string; // optional — skip if same key fired in last TTL window
+  silent?: boolean;
+  parseMode?: "HTML" | "Markdown";
+}): Promise<TelegramSendResult> {
+  if (!isTelegramConfigured()) return { ok: false, error: "not_configured" };
+
+  // Dedupe check
+  if (opts.dedupeKey) {
+    const key = `tg:dedupe:${opts.dedupeKey}`;
+    const set = await redis.set(key, "1", "EX", DEDUPE_TTL_SECONDS, "NX").catch(() => null);
+    if (set === null) return { ok: true, deduped: true };
+  }
+
+  try {
+    const res = await fetch(`https://api.telegram.org/bot${TOKEN}/sendMessage`, {
+      method: "POST",
+      headers: { "content-type": "application/json" },
+      body: JSON.stringify({
+        chat_id: CHAT_ID,
+        text: opts.text,
+        parse_mode: opts.parseMode ?? "HTML",
+        disable_notification: opts.silent ?? false,
+        disable_web_page_preview: true,
+      }),
+    });
+    if (!res.ok) {
+      const err = await res.text();
+      return { ok: false, error: `${res.status}: ${err.slice(0, 200)}` };
+    }
+    return { ok: true };
+  } catch (e) {
+    return { ok: false, error: (e as Error).message };
+  }
+}
+
+// Helpers for common alert shapes.
+
+export function alertP0Insight(opts: {
+  insightId: string;
+  title: string;
+  severity: string;
+  type: string;
+  fingerprint: string;
+  panelUrl: string;
+}): Promise<TelegramSendResult> {
+  const text = [
+    `🔴 <b>${opts.severity} insight</b> · ${escapeHtml(opts.type)}`,
+    `<b>${escapeHtml(opts.title)}</b>`,
+    ``,
+    `<a href="${opts.panelUrl}/insights/i/${opts.insightId}">Open in panel</a>`,
+    `<code>${opts.fingerprint.slice(0, 16)}</code>`,
+  ].join("\n");
+  return sendTelegram({ text, dedupeKey: `p0:${opts.insightId}` });
+}
+
+export function alertRegression(opts: {
+  insightId: string;
+  title: string;
+  fingerprint: string;
+  occurrenceCount: number;
+  panelUrl: string;
+}): Promise<TelegramSendResult> {
+  const text = [
+    `⚠️ <b>Regression detected</b>`,
+    `<b>${escapeHtml(opts.title)}</b>`,
+    `Same fingerprint sessions: ${opts.occurrenceCount}`,
+    ``,
+    `<a href="${opts.panelUrl}/insights/i/${opts.insightId}">Open in panel</a>`,
+  ].join("\n");
+  return sendTelegram({ text, dedupeKey: `regress:${opts.insightId}` });
+}
+
+export function alertSanitizationAnomaly(opts: {
+  sessionId: string;
+  tokens: number;
+  panelUrl: string;
+}): Promise<TelegramSendResult> {
+  const text = [
+    `🛑 <b>Sanitization anomaly</b>`,
+    `Session ${opts.sessionId.slice(0, 12)}… produced ${opts.tokens} tokens with 0 PII matches.`,
+    `<i>Possible PII leak — review before LLM call.</i>`,
+    ``,
+    `<a href="${opts.panelUrl}/insights/sessions/${opts.sessionId}">Inspect session</a>`,
+  ].join("\n");
+  return sendTelegram({ text, dedupeKey: `sanitize:${opts.sessionId}` });
+}
+
+export function alertBudgetCap(opts: {
+  state: string;
+  todayUsd: number;
+  monthUsd: number;
+  panelUrl: string;
+}): Promise<TelegramSendResult> {
+  const text = [
+    `💸 <b>Budget cap reached</b>: <code>${opts.state}</code>`,
+    `Today $${opts.todayUsd.toFixed(4)} · Month $${opts.monthUsd.toFixed(2)}`,
+    ``,
+    `<a href="${opts.panelUrl}/insights/costs">Cost dashboard</a> · <a href="${opts.panelUrl}/insights/settings/budgets">Adjust limits</a>`,
+  ].join("\n");
+  // Dedupe per-state-per-day so we don't spam after every analyze cycle.
+  const day = new Date().toISOString().slice(0, 10);
+  return sendTelegram({ text, dedupeKey: `budget:${opts.state}:${day}` });
+}
+
+function escapeHtml(s: string): string {
+  return s.replace(/&/g, "&amp;").replace(/</g, "&lt;").replace(/>/g, "&gt;");
+}
--- a/apps/worker/src/schedulers/pipeline.ts
+++ b/apps/worker/src/schedulers/pipeline.ts
@@ -8,6 +8,7 @@ import { runValidation } from "../jobs/validation";
 import { runGithubSync } from "../jobs/github-sync";
 import { runRetention } from "../jobs/retention";
 import { runEvalSet } from "../jobs/eval-run";
+import { runDailyBrief } from "../jobs/daily-brief";

 const QUEUE = "insight-pipeline";

@@ -71,6 +72,11 @@ async function runJob(job: Job) {
      const res = await runEvalSet(evalSetId, promptVersion);
      return res;
    }
+    case "daily-brief": {
+      const res = await runDailyBrief();
+      console.log(`[pipeline] daily-brief sent=${res.sent}${res.reason ? ` reason=${res.reason}` : ""}`);
+      return res;
+    }
    default:
      return { ok: false, error: `unknown job ${job.name}` };
  }
@@ -112,6 +118,11 @@ export async function startInsightPipeline() {
    { pattern: "15 4 * * *" },
    { name: "retention", data: {}, opts: { removeOnComplete: 30, removeOnFail: 15 } },
  );
+  await queue.upsertJobScheduler(
+    "daily-brief",
+    { pattern: "0 5 * * *" }, // 08:00 Europe/Istanbul = 05:00 UTC
+    { name: "daily-brief", data: {}, opts: { removeOnComplete: 30, removeOnFail: 15 } },
+  );

  new Worker(QUEUE, runJob, {
    connection: redis,
@@ -120,6 +131,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",
+    "[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",
  );
 }