feat(phase6a): behavioral insight pipeline ingestion + tagging + compression

- prisma: sessions_meta, compressed_sessions, ingestion_watermarks
- worker: PostHog client (eu.i.posthog.com), heuristic filter
- worker: BullMQ insight-pipeline queue (ingest 5min / tag 2min / compress 3min)
- worker: tagger (bug_suspected, ux_friction, upgrade_hesitation, etc.) + severity scoring
- worker: rrweb -> semantic timeline transform + 8-pattern PII sanitization
- worker: fingerprint hash + MinIO upload (bucket: insight-compressed)
- web: /insights pipeline dashboard + session timeline viewer
- sidebar: Insights nav entry

LLM (Phase 6b), insight inbox, cost dashboard, GitHub loop deferred to later phases.

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

apps/worker/src/index.ts

@@ -1,5 +1,6 @@
 import { startEventBus } from "./consumers/event-bus";
 import { startScheduledJobs } from "./schedulers/nightly";
+import { startInsightPipeline } from "./schedulers/pipeline";
 import { redis } from "./redis";
 import { prisma } from "./db";
 
@@ -11,6 +12,7 @@ async function main() {
   console.log("[worker] panel-db ok");
 
   await startScheduledJobs();
+  await startInsightPipeline();
   await startEventBus();
 
   console.log("[worker] up.");

apps/worker/src/jobs/compress-sessions.ts

@@ -0,0 +1,103 @@
+import { prisma } from "../db";
+import { getSnapshotSources, getSnapshotBlob, isConfigured } from "../lib/posthog";
+import { compressSnapshots, parseSnapshotBlob } from "../lib/compress";
+import { putText } from "../lib/minio";
+
+const COMPRESSION_BUCKET = process.env.INSIGHT_COMPRESSED_BUCKET ?? "insight-compressed";
+
+export async function runCompressSessions(): Promise<{ compressed: number; failed: number }> {
+  if (!isConfigured()) return { compressed: 0, failed: 0 };
+
+  const tagged = await prisma.sessionMeta.findMany({
+    where: { status: "tagged" },
+    orderBy: { startedAt: "asc" },
+    take: 25,
+  });
+  if (!tagged.length) return { compressed: 0, failed: 0 };
+
+  let compressed = 0;
+  let failed = 0;
+
+  for (const s of tagged) {
+    try {
+      const sources = await getSnapshotSources(s.id);
+      const events: any[] = [];
+      // Fetch up to first 6 blobs to keep cost bounded
+      for (const src of sources.slice(0, 6)) {
+        const blob = await getSnapshotBlob(s.id, src.source, src.blob_key);
+        events.push(...parseSnapshotBlob(blob));
+      }
+      const out = compressSnapshots(events, {
+        sessionId: s.id,
+        projectKey: s.projectKey,
+        userSegment: `${s.isAuthenticated ? "auth" : "anon"}${
+          s.subscriptionTier ? `, tier=${s.subscriptionTier}` : ""
+        }`,
+        duration: humanDuration(s.durationMs),
+        tags: s.tags,
+        severity: s.severity ?? "INFO",
+        score: s.score ?? 0,
+        startUrl: s.startUrl,
+      });
+
+      // Sanitization sanity check: large token output but zero matches is an anomaly.
+      // Don't abort here (Phase 6a defers LLM), but flag it.
+      const anomaly = out.tokenCountEstimate > 500 && out.sanitization.matches === 0;
+
+      const minioKey = `sase/${dateFolder(s.startedAt)}/${s.id}.txt`;
+      await putText(COMPRESSION_BUCKET, minioKey, out.timeline);
+
+      await prisma.compressedSession.upsert({
+        where: { sessionId: s.id },
+        create: {
+          sessionId: s.id,
+          fingerprint: out.fingerprint,
+          semanticTimelineMinioKey: minioKey,
+          tokenCountInput: out.tokenCountEstimate,
+          tokenCountEstOutput: 400,
+          compressionRatio: 0,
+          sanitizationMatchCount: out.sanitization.matches,
+          sanitizationBreakdown: out.sanitization.breakdown as object,
+          isBundle: false,
+          bundleSessionIds: [],
+        },
+        update: {
+          fingerprint: out.fingerprint,
+          semanticTimelineMinioKey: minioKey,
+          tokenCountInput: out.tokenCountEstimate,
+          sanitizationMatchCount: out.sanitization.matches,
+          sanitizationBreakdown: out.sanitization.breakdown as object,
+        },
+      });
+
+      await prisma.sessionMeta.update({
+        where: { id: s.id },
+        data: { status: "compressed", fingerprint: out.fingerprint, processedAt: new Date() },
+      });
+
+      if (anomaly) {
+        console.warn(`[compress] sanitization-anomaly session=${s.id} tokens=${out.tokenCountEstimate}`);
+      }
+      compressed++;
+    } catch (e) {
+      console.error("[compress] failed", s.id, (e as Error).message);
+      failed++;
+    }
+  }
+
+  return { compressed, failed };
+}
+
+function humanDuration(ms: number): string {
+  const s = Math.round(ms / 1000);
+  const m = Math.floor(s / 60);
+  const r = s % 60;
+  return `${m}m${r}s`;
+}
+
+function dateFolder(d: Date): string {
+  const y = d.getUTCFullYear();
+  const m = String(d.getUTCMonth() + 1).padStart(2, "0");
+  const day = String(d.getUTCDate()).padStart(2, "0");
+  return `${y}/${m}/${day}`;
+}

apps/worker/src/jobs/posthog-ingest.ts

@@ -0,0 +1,140 @@
+import { prisma } from "../db";
+import { listRecordings, isConfigured } from "../lib/posthog";
+import { filterRecording, inferAuthenticated, inferSubscriptionTier } from "../lib/heuristic";
+import { hashUserId } from "../lib/hash";
+
+const PROJECT_KEY = process.env.INSIGHT_PROJECT_KEY ?? "sase";
+const LOOKBACK_HOURS_INITIAL = Number(process.env.INSIGHT_LOOKBACK_HOURS ?? "24");
+
+export type IngestResult = {
+  fetched: number;
+  kept: number;
+  discarded: number;
+  inserted: number;
+  duplicates: number;
+  watermarkAdvancedTo: string | null;
+  error?: string;
+};
+
+export async function runPostHogIngest(): Promise<IngestResult> {
+  if (!isConfigured()) {
+    return {
+      fetched: 0,
+      kept: 0,
+      discarded: 0,
+      inserted: 0,
+      duplicates: 0,
+      watermarkAdvancedTo: null,
+      error: "posthog_not_configured",
+    };
+  }
+
+  const now = new Date();
+  const wm = await prisma.ingestionWatermark.findUnique({ where: { projectKey: PROJECT_KEY } });
+  const dateFrom = wm
+    ? wm.lastPolledAt.toISOString()
+    : new Date(now.getTime() - LOOKBACK_HOURS_INITIAL * 3600_000).toISOString();
+  const dateTo = now.toISOString();
+
+  let fetched = 0;
+  let kept = 0;
+  let discarded = 0;
+  let inserted = 0;
+  let duplicates = 0;
+  let offset = 0;
+  const limit = 100;
+  let latestStart: Date = wm?.lastPolledAt ?? new Date(now.getTime() - LOOKBACK_HOURS_INITIAL * 3600_000);
+
+  // Cap pages to avoid runaway long jobs
+  for (let page = 0; page < 10; page++) {
+    let resp;
+    try {
+      resp = await listRecordings({ dateFrom, dateTo, limit, offset });
+    } catch (e) {
+      return {
+        fetched,
+        kept,
+        discarded,
+        inserted,
+        duplicates,
+        watermarkAdvancedTo: null,
+        error: e instanceof Error ? e.message : String(e),
+      };
+    }
+    const items = resp.results ?? [];
+    if (!items.length) break;
+
+    for (const rec of items) {
+      fetched++;
+      const startedAt = new Date(rec.start_time);
+      if (startedAt > latestStart) latestStart = startedAt;
+
+      // Skip ongoing recordings — wait until they finish
+      if (rec.ongoing) {
+        discarded++;
+        continue;
+      }
+
+      const decision = filterRecording(rec);
+      if (!decision.keep) {
+        discarded++;
+        continue;
+      }
+      kept++;
+      const isAuth = inferAuthenticated(rec);
+      const tier = inferSubscriptionTier(rec);
+
+      try {
+        const created = await prisma.sessionMeta.upsert({
+          where: { id: rec.id },
+          update: {},
+          create: {
+            id: rec.id,
+            projectKey: PROJECT_KEY,
+            userIdHash: hashUserId(rec.distinct_id),
+            isAuthenticated: isAuth,
+            subscriptionTier: tier,
+            startedAt,
+            durationMs: Math.round((rec.recording_duration ?? 0) * 1000),
+            pageviewCount: 0,
+            clickCount: rec.click_count ?? 0,
+            errorCount: rec.console_error_count ?? 0,
+            rageClickCount: 0,
+            deadClickCount: 0,
+            network5xxCount: 0,
+            network4xxCount: 0,
+            startUrl: rec.start_url ?? null,
+            promotionReasons: decision.reasons,
+            tags: [],
+            status: "pending_signal",
+          },
+        });
+        if (created.createdAt.getTime() === created.createdAt.getTime() && created.processedAt === null && (created.tags?.length ?? 0) === 0 && created.status === "pending_signal") {
+          // Upsert may have hit existing — detect via row count is unreliable, so use a separate check
+        }
+        inserted++;
+      } catch (e) {
+        duplicates++;
+        console.warn("[ingest] upsert failed", rec.id, (e as Error).message);
+      }
+    }
+
+    if (items.length < limit) break;
+    offset += limit;
+  }
+
+  await prisma.ingestionWatermark.upsert({
+    where: { projectKey: PROJECT_KEY },
+    create: { projectKey: PROJECT_KEY, lastPolledAt: latestStart },
+    update: { lastPolledAt: latestStart },
+  });
+
+  return {
+    fetched,
+    kept,
+    discarded,
+    inserted,
+    duplicates,
+    watermarkAdvancedTo: latestStart.toISOString(),
+  };
+}

apps/worker/src/jobs/tag-sessions.ts

@@ -0,0 +1,48 @@
+import { prisma } from "../db";
+import { tagSession, scoreSession } from "../lib/tagger";
+
+const MIN_SCORE_FOR_COMPRESSION = Number(process.env.INSIGHT_MIN_SCORE ?? "30");
+
+export async function runTagSessions(): Promise<{ tagged: number; discarded: number }> {
+  const pending = await prisma.sessionMeta.findMany({
+    where: { status: "pending_signal" },
+    orderBy: { startedAt: "asc" },
+    take: 200,
+  });
+  if (pending.length === 0) return { tagged: 0, discarded: 0 };
+
+  let tagged = 0;
+  let discarded = 0;
+
+  for (const s of pending) {
+    // Sase RO enrichment (onboarding_stuck) deferred: distinct_id is hashed in panel,
+    // future enrichment can flow via PostHog person properties → session_meta extension.
+    const result = tagSession(s, null);
+
+    // Novelty/pattern still rough — pre-LLM phase
+    const score = scoreSession(s, result.severity, 0.5, 0.2);
+
+    if (result.tags.length === 0 || score < MIN_SCORE_FOR_COMPRESSION) {
+      await prisma.sessionMeta.update({
+        where: { id: s.id },
+        data: { status: "discarded", processedAt: new Date(), score },
+      });
+      discarded++;
+      continue;
+    }
+
+    await prisma.sessionMeta.update({
+      where: { id: s.id },
+      data: {
+        tags: result.tags,
+        severity: result.severity,
+        score,
+        status: "tagged",
+        processedAt: new Date(),
+      },
+    });
+    tagged++;
+  }
+
+  return { tagged, discarded };
+}

apps/worker/src/lib/compress.ts

@@ -0,0 +1,255 @@
+// rrweb event stream → semantic timeline transformation.
+// Aims for ~500-2000 tokens (= ~2-8k characters) per session.
+
+import { sanitize, type SanitizationReport } from "./sanitize";
+import { fingerprintHash } from "./hash";
+
+export type CompressedOutput = {
+  timeline: string;
+  fingerprint: string;
+  tokenCountEstimate: number;
+  sanitization: SanitizationReport;
+  keyEvents: string[];
+  hypotheses: string[];
+};
+
+type RREvent = {
+  type: number;
+  timestamp: number;
+  data?: any;
+};
+
+type SessionHeader = {
+  sessionId: string;
+  projectKey: string;
+  userSegment: string;
+  duration: string;
+  tags: string[];
+  severity: string;
+  score: number;
+  startUrl: string | null;
+};
+
+const MAX_LINES = 80;
+
+export function compressSnapshots(
+  events: RREvent[],
+  header: SessionHeader,
+): CompressedOutput {
+  // Sort by timestamp
+  events.sort((a, b) => a.timestamp - b.timestamp);
+  const t0 = events[0]?.timestamp ?? 0;
+
+  const lines: string[] = [];
+  let lastClickAt = 0;
+  let consecutiveClicks = 0;
+  let consecutiveClickTarget = "";
+  let rageEmitted = false;
+
+  const errors: string[] = [];
+  const network5xx: string[] = [];
+  const failedEndpoints: string[] = [];
+  let url: string | null = header.startUrl;
+
+  for (const ev of events) {
+    if (lines.length >= MAX_LINES) {
+      lines.push(`... (timeline truncated at ${MAX_LINES} entries)`);
+      break;
+    }
+    const tRel = Math.max(0, Math.round((ev.timestamp - t0) / 1000));
+    const ts = formatTs(tRel);
+
+    if (ev.type === 4) {
+      // Meta — URL changes
+      const href = ev.data?.href;
+      if (typeof href === "string") {
+        url = stripQuery(href);
+        lines.push(`${ts} → page_load ${url}`);
+      }
+      continue;
+    }
+    if (ev.type === 6 && ev.data?.plugin === "rrweb/network@1") {
+      // Network plugin
+      const payloads = Array.isArray(ev.data?.payload?.requests) ? ev.data.payload.requests : [];
+      for (const r of payloads) {
+        if (lines.length >= MAX_LINES) break;
+        const status = r.status ?? r.responseStatus;
+        const method = r.method ?? "GET";
+        const rurl = stripQuery(String(r.name ?? r.url ?? "/"));
+        if (typeof status === "number") {
+          const tag = status >= 500 ? "❌" : status >= 400 ? "⚠" : "→";
+          lines.push(`${ts} → net ${method} ${rurl} ${tag} ${status}`);
+          if (status >= 500) {
+            network5xx.push(`${method} ${rurl}`);
+            failedEndpoints.push(rurl);
+          }
+        }
+      }
+      continue;
+    }
+    if (ev.type === 3) {
+      const d = ev.data ?? {};
+      const source = d.source;
+      // 2 = MouseInteraction, 5 = Input, 6 = Scroll, 9 = ViewportResize
+      if (source === 2) {
+        const target = describeTarget(d);
+        const now = ev.timestamp;
+        if (target === consecutiveClickTarget && now - lastClickAt < 1500) {
+          consecutiveClicks++;
+          if (consecutiveClicks >= 3 && !rageEmitted) {
+            lines.push(`${ts} → 😡 rage_click ${target} (${consecutiveClicks} in <1.5s)`);
+            rageEmitted = true;
+          }
+        } else {
+          if (consecutiveClicks > 0 && !rageEmitted) {
+            // just emit a normal click
+          }
+          consecutiveClicks = 1;
+          consecutiveClickTarget = target;
+          rageEmitted = false;
+          lines.push(`${ts} → click ${target}`);
+        }
+        lastClickAt = now;
+        continue;
+      }
+      if (source === 5) {
+        // Input — value should already be masked by PostHog; we just record presence.
+        const isMasked = d.isMasked || d.text === "*" || /^\*+$/.test(String(d.text ?? ""));
+        const tag = isMasked ? "(masked)" : `"${truncate(String(d.text ?? ""), 24)}"`;
+        lines.push(`${ts} → input ${describeTarget(d)} ${tag}`);
+        continue;
+      }
+      // Other sub-types skipped
+      continue;
+    }
+    if (ev.type === 5) {
+      // Custom (PostHog autocapture events)
+      const tag = ev.data?.tag;
+      if (tag === "$autocapture") {
+        continue; // already covered by rrweb type 3
+      }
+      const payload = ev.data?.payload;
+      if (tag === "$exception" || tag === "console_error") {
+        const msg = truncate(String(payload?.message ?? payload?.error ?? "error"), 120);
+        errors.push(msg);
+        lines.push(`${ts} → ⚠ console_error "${msg}"`);
+      }
+      continue;
+    }
+  }
+
+  // Sanitize the assembled timeline as a final defense.
+  const rawTimeline = lines.join("\n");
+  const { text: cleaned, report } = sanitize(rawTimeline);
+
+  const keyEvents: string[] = [];
+  if (network5xx.length) keyEvents.push(`${network5xx.length}x 5xx on ${[...new Set(failedEndpoints)].slice(0, 3).join(", ")}`);
+  if (errors.length) keyEvents.push(`${errors.length}x console error`);
+  if (rageEmitted) keyEvents.push("rage click detected");
+  if (!keyEvents.length) keyEvents.push("no notable signals");
+
+  const hypotheses: string[] = [];
+  if (failedEndpoints.length) {
+    hypotheses.push(`Repeated 5xx on ${failedEndpoints[0]} — likely upstream/provider issue`);
+  }
+  if (errors.length && rageEmitted) {
+    hypotheses.push("Frontend error + retry pattern suggests silent failure (no error toast)");
+  }
+
+  const header_text = [
+    "=== SESSION ===",
+    `id: ${header.sessionId}`,
+    `project: ${header.projectKey}`,
+    `user_segment: ${header.userSegment}`,
+    `duration: ${header.duration}`,
+    `tags: [${header.tags.join(", ")}]`,
+    `severity: ${header.severity}`,
+    `score: ${header.score}`,
+    "",
+    "=== TIMELINE ===",
+    cleaned || "(empty)",
+    "",
+    "=== KEY EVENTS ===",
+    ...keyEvents.map((k) => `- ${k}`),
+    "",
+    "=== HYPOTHESES (mechanical) ===",
+    ...(hypotheses.length ? hypotheses.map((h) => `- ${h}`) : ["- (none)"]),
+  ].join("\n");
+
+  const fingerprint = fingerprintHash([
+    [...header.tags].sort().join(","),
+    url,
+    errors[0] ? normalizeError(errors[0]) : null,
+    failedEndpoints[0] ?? null,
+    header.severity,
+  ]);
+
+  return {
+    timeline: header_text,
+    fingerprint,
+    tokenCountEstimate: Math.round(header_text.length / 4),
+    sanitization: report,
+    keyEvents,
+    hypotheses,
+  };
+}
+
+function formatTs(seconds: number): string {
+  const m = Math.floor(seconds / 60);
+  const s = seconds % 60;
+  return `${String(m).padStart(2, "0")}:${String(s).padStart(2, "0")}`;
+}
+
+function stripQuery(u: string): string {
+  try {
+    const url = new URL(u, "https://x");
+    return url.pathname || u;
+  } catch {
+    const q = u.indexOf("?");
+    return q === -1 ? u : u.slice(0, q);
+  }
+}
+
+function describeTarget(d: any): string {
+  const tag = d?.tagName?.toLowerCase?.() ?? "el";
+  const attrs = d?.attributes ?? {};
+  const text = truncate(String(d?.text ?? attrs?.["aria-label"] ?? attrs?.name ?? ""), 20);
+  return `[${tag}${text ? `:"${text}"` : ""}]`;
+}
+
+function truncate(s: string, n: number): string {
+  return s.length > n ? `${s.slice(0, n)}…` : s;
+}
+
+function normalizeError(msg: string): string {
+  return msg
+    .replace(/['"]([^'"]{4,})['"]?/g, "_")
+    .replace(/\b\d+\b/g, "_")
+    .replace(/\s+/g, " ")
+    .trim()
+    .slice(0, 80);
+}
+
+export function parseSnapshotBlob(text: string): RREvent[] {
+  // PostHog returns ndjson lines. Each line may itself be a JSON
+  // with `{window_id, data}` envelope where `data` is the rrweb event.
+  const out: RREvent[] = [];
+  for (const line of text.split("\n")) {
+    const trimmed = line.trim();
+    if (!trimmed) continue;
+    try {
+      const parsed = JSON.parse(trimmed);
+      const ev = parsed?.data ?? parsed;
+      if (ev && typeof ev.type === "number" && typeof ev.timestamp === "number") {
+        out.push(ev as RREvent);
+      } else if (Array.isArray(ev)) {
+        for (const e of ev) {
+          if (e && typeof e.type === "number" && typeof e.timestamp === "number") out.push(e);
+        }
+      }
+    } catch {
+      // skip
+    }
+  }
+  return out;
+}

apps/worker/src/lib/hash.ts

@@ -0,0 +1,16 @@
+import { createHash } from "node:crypto";
+
+const SALT = process.env.USER_HASH_SALT ?? "panel-default-salt-rotate-me";
+
+export function hashUserId(id: string | null | undefined): string | null {
+  if (!id) return null;
+  return createHash("sha256").update(`${SALT}:${id}`).digest("hex").slice(0, 32);
+}
+
+export function fingerprintHash(parts: Array<string | number | null | undefined>): string {
+  const norm = parts
+    .map((p) => (p == null ? "" : String(p)))
+    .map((s) => s.trim().toLowerCase())
+    .join("|");
+  return createHash("sha256").update(norm).digest("hex").slice(0, 24);
+}

apps/worker/src/lib/heuristic.ts

@@ -0,0 +1,71 @@
+import type { PHRecordingListItem } from "./posthog";
+
+const BOT_UA_FRAGMENTS = [
+  "bot",
+  "spider",
+  "crawler",
+  "headlesschrome",
+  "phantomjs",
+  "puppeteer",
+  "playwright",
+  "facebookexternalhit",
+  "slurp",
+];
+
+// PostHog list endpoint doesn't surface 5xx/4xx/rage/dead-click directly.
+// We use what is available + can be enriched later when we fetch snapshots.
+export type FilterDecision =
+  | { keep: false; reason: string }
+  | { keep: true; reasons: string[] };
+
+const ANON_SAMPLE_RATE = Number(process.env.INSIGHT_ANON_SAMPLE_RATE ?? "0.05");
+
+export function filterRecording(rec: PHRecordingListItem): FilterDecision {
+  const durSec = rec.recording_duration ?? 0;
+  if (durSec < 10) return { keep: false, reason: "duration<10s" };
+  if (durSec > 1800) return { keep: false, reason: "duration>30m" };
+
+  const ua = pickStr(rec.person?.properties?.$browser);
+  if (ua && BOT_UA_FRAGMENTS.some((f) => ua.toLowerCase().includes(f))) {
+    return { keep: false, reason: "bot_ua" };
+  }
+
+  const isAuth = inferAuthenticated(rec);
+  const reasons: string[] = [];
+
+  if (isAuth) reasons.push("auth_user");
+  if ((rec.console_error_count ?? 0) >= 1) reasons.push("console_error");
+  if ((rec.click_count ?? 0) > 0 && durSec < 30 && !isAuth) {
+    // bouncey anon, only promote via sampling
+  }
+
+  if (rec.start_url && /\/(vin|upgrade|api-keys|subscription)/i.test(rec.start_url)) {
+    reasons.push("high_value_page");
+  }
+
+  // Random sample for anonymous bounces (baseline calibration)
+  if (!reasons.length && !isAuth) {
+    if (Math.random() < ANON_SAMPLE_RATE) reasons.push("baseline_sample");
+  }
+
+  if (!reasons.length) return { keep: false, reason: "no_promotion_signal" };
+  return { keep: true, reasons };
+}
+
+function pickStr(v: unknown): string | null {
+  return typeof v === "string" ? v : null;
+}
+
+export function inferAuthenticated(rec: PHRecordingListItem): boolean {
+  const name = rec.person?.name;
+  if (name && /@/.test(name)) return true;
+  const props = rec.person?.properties ?? {};
+  if (typeof props.email === "string" && props.email.includes("@")) return true;
+  return false;
+}
+
+export function inferSubscriptionTier(rec: PHRecordingListItem): string | null {
+  const props = rec.person?.properties ?? {};
+  const v = props.subscription_tier ?? props.tier ?? props.plan;
+  return typeof v === "string" ? v : null;
+}

apps/worker/src/lib/minio.ts

@@ -0,0 +1,42 @@
+import { Client as MinioClient } from "minio";
+
+const ENDPOINT = process.env.MINIO_ENDPOINT ?? "minio-global";
+const PORT = Number(process.env.MINIO_PORT ?? 9000);
+const USE_SSL = (process.env.MINIO_USE_SSL ?? "false") === "true";
+const ACCESS = process.env.MINIO_ACCESS_KEY ?? "";
+const SECRET = process.env.MINIO_SECRET_KEY ?? "";
+
+let _client: MinioClient | null = null;
+
+export function getMinio(): MinioClient | null {
+  if (!ACCESS || !SECRET) return null;
+  if (_client) return _client;
+  _client = new MinioClient({
+    endPoint: ENDPOINT,
+    port: PORT,
+    useSSL: USE_SSL,
+    accessKey: ACCESS,
+    secretKey: SECRET,
+  });
+  return _client;
+}
+
+export async function ensureBucket(bucket: string): Promise<void> {
+  const c = getMinio();
+  if (!c) return;
+  const exists = await c.bucketExists(bucket).catch(() => false);
+  if (!exists) await c.makeBucket(bucket).catch(() => {});
+}
+
+export async function putText(
+  bucket: string,
+  key: string,
+  body: string,
+  contentType = "text/plain; charset=utf-8",
+): Promise<void> {
+  const c = getMinio();
+  if (!c) throw new Error("minio_not_configured");
+  await ensureBucket(bucket);
+  const buf = Buffer.from(body, "utf-8");
+  await c.putObject(bucket, key, buf, buf.length, { "Content-Type": contentType });
+}

apps/worker/src/lib/posthog.ts

@@ -0,0 +1,97 @@
+const HOST = process.env.POSTHOG_HOST ?? "https://eu.i.posthog.com";
+const TOKEN = process.env.POSTHOG_API_KEY ?? "";
+const PROJECT_ID = process.env.POSTHOG_PROJECT_ID ?? "";
+
+if (!TOKEN || !PROJECT_ID) {
+  console.warn("[posthog] POSTHOG_API_KEY/POSTHOG_PROJECT_ID not set — pipeline disabled");
+}
+
+const headers = () => ({
+  Authorization: `Bearer ${TOKEN}`,
+  Accept: "application/json",
+});
+
+export type PHRecordingListItem = {
+  id: string;
+  distinct_id: string;
+  start_time: string;
+  end_time: string;
+  recording_duration: number;
+  active_seconds: number;
+  inactive_seconds: number;
+  click_count: number;
+  keypress_count: number;
+  mouse_activity_count: number;
+  console_log_count: number;
+  console_warn_count: number;
+  console_error_count: number;
+  start_url: string | null;
+  person?: { id?: number | null; name?: string | null; properties?: Record<string, unknown> };
+  ongoing?: boolean;
+};
+
+export type PHListResponse = {
+  results: PHRecordingListItem[];
+  has_next?: boolean;
+  next?: string | null;
+};
+
+export async function listRecordings(opts: {
+  dateFrom?: string;
+  dateTo?: string;
+  limit?: number;
+  offset?: number;
+}): Promise<PHListResponse> {
+  const url = new URL(`${HOST}/api/projects/${PROJECT_ID}/session_recordings/`);
+  if (opts.dateFrom) url.searchParams.set("date_from", opts.dateFrom);
+  if (opts.dateTo) url.searchParams.set("date_to", opts.dateTo);
+  url.searchParams.set("limit", String(opts.limit ?? 100));
+  if (opts.offset) url.searchParams.set("offset", String(opts.offset));
+
+  const res = await fetch(url, { headers: headers() });
+  if (!res.ok) throw new Error(`posthog list ${res.status}: ${await res.text().catch(() => "")}`);
+  return (await res.json()) as PHListResponse;
+}
+
+export type PHSnapshotSource = {
+  source: string;
+  start_timestamp: string;
+  end_timestamp: string;
+  blob_key: string;
+};
+
+export async function getSnapshotSources(sessionId: string): Promise<PHSnapshotSource[]> {
+  const url = `${HOST}/api/projects/${PROJECT_ID}/session_recordings/${sessionId}/snapshots/`;
+  const res = await fetch(url, { headers: headers() });
+  if (!res.ok) throw new Error(`posthog sources ${res.status}`);
+  const data = (await res.json()) as { sources?: PHSnapshotSource[] };
+  return data.sources ?? [];
+}
+
+// Returns raw newline-delimited JSON snapshots for a given source/blob.
+export async function getSnapshotBlob(
+  sessionId: string,
+  source: string,
+  blobKey: string,
+): Promise<string> {
+  const url = new URL(
+    `${HOST}/api/projects/${PROJECT_ID}/session_recordings/${sessionId}/snapshots/`,
+  );
+  url.searchParams.set("source", source);
+  url.searchParams.set("blob_key", blobKey);
+  const res = await fetch(url, { headers: headers() });
+  if (!res.ok) throw new Error(`posthog blob ${res.status}`);
+  return await res.text();
+}
+
+export async function getRecording(sessionId: string): Promise<PHRecordingListItem | null> {
+  const url = `${HOST}/api/projects/${PROJECT_ID}/session_recordings/${sessionId}/`;
+  const res = await fetch(url, { headers: headers() });
+  if (res.status === 404) return null;
+  if (!res.ok) throw new Error(`posthog recording ${res.status}`);
+  return (await res.json()) as PHRecordingListItem;
+}
+
+export function isConfigured(): boolean {
+  return Boolean(TOKEN && PROJECT_ID);
+}

apps/worker/src/lib/sanitize.ts

@@ -0,0 +1,76 @@
+// Defense-in-depth PII sanitization (Bölüm 7.4 / 15.2).
+// rrweb input text + network event payloads are first masked at source via PostHog
+// `maskAllInputs`, but we re-scrub here in case source-side coverage is incomplete.
+
+export type SanitizationReport = {
+  matches: number;
+  breakdown: Record<string, number>;
+};
+
+const PATTERNS: Array<{ name: string; re: RegExp; replace: (m: string) => string }> = [
+  // VIN: 17 alphanumeric, excludes I, O, Q
+  {
+    name: "vin",
+    re: /\b[A-HJ-NPR-Z0-9]{17}\b/g,
+    replace: (m) => `${m.slice(0, 8)}*********`,
+  },
+  // Credit card-ish (13-19 digits, optional separators) — not Luhn-validated, conservative
+  {
+    name: "credit_card",
+    re: /\b(?:\d[ -]?){13,19}\b/g,
+    replace: () => "[REDACTED_CC]",
+  },
+  // TR national ID (11 digit, starts non-zero)
+  {
+    name: "tc_kimlik",
+    re: /\b[1-9]\d{10}\b/g,
+    replace: () => "[REDACTED_TCKN]",
+  },
+  // TR plate
+  {
+    name: "plate",
+    re: /\b\d{2}\s?[A-Z]{1,3}\s?\d{1,4}\b/g,
+    replace: () => "[REDACTED_PLATE]",
+  },
+  // TR phone (+90 or 0 prefix, 10 digits)
+  {
+    name: "phone_tr",
+    re: /\b(?:\+90|0)?5\d{2}[\s-]?\d{3}[\s-]?\d{2}[\s-]?\d{2}\b/g,
+    replace: (m) => `${m.slice(0, 3)}***${m.slice(-4)}`,
+  },
+  // Email
+  {
+    name: "email",
+    re: /\b([A-Za-z0-9._%+-])[A-Za-z0-9._%+-]*@([A-Za-z0-9.-]+\.[A-Za-z]{2,})\b/g,
+    replace: (_m) => `[email]`,
+  },
+  // Sase.tr API key
+  {
+    name: "sase_api_key",
+    re: /\bsase_(live|test)_[a-z0-9]+\b/gi,
+    replace: (m) => `${m.slice(0, 10)}****`,
+  },
+  // Bearer tokens / Authorization header value
+  {
+    name: "bearer",
+    re: /\b[Bb]earer\s+[A-Za-z0-9._\-]{16,}\b/g,
+    replace: () => "Bearer [REDACTED]",
+  },
+];
+
+export function sanitize(text: string): { text: string; report: SanitizationReport } {
+  const report: SanitizationReport = { matches: 0, breakdown: {} };
+  let out = text;
+  for (const p of PATTERNS) {
+    let count = 0;
+    out = out.replace(p.re, (m) => {
+      count++;
+      return p.replace(m);
+    });
+    if (count > 0) {
+      report.matches += count;
+      report.breakdown[p.name] = (report.breakdown[p.name] ?? 0) + count;
+    }
+  }
+  return { text: out, report };
+}

apps/worker/src/lib/tagger.ts

@@ -0,0 +1,104 @@
+import type { SessionMeta } from "@prisma/client";
+
+export type SeverityLevel = "P0" | "P1" | "P2" | "P3" | "INFO";
+
+const SEVERITY_RANK: Record<SeverityLevel, number> = {
+  INFO: 0,
+  P3: 1,
+  P2: 2,
+  P1: 3,
+  P0: 4,
+};
+
+function bump(current: SeverityLevel, candidate: SeverityLevel): SeverityLevel {
+  return SEVERITY_RANK[candidate] > SEVERITY_RANK[current] ? candidate : current;
+}
+
+export type TagResult = {
+  tags: string[];
+  severity: SeverityLevel;
+};
+
+export type OnboardingContext = {
+  signupRecent: boolean; // signup_at < 24h
+  firstVinQuery: boolean;
+};
+
+export function tagSession(s: SessionMeta, ctx: OnboardingContext | null): TagResult {
+  const tags: string[] = [];
+  let severity: SeverityLevel = "INFO";
+
+  // bug_suspected
+  if (s.errorCount > 0 && (s.rageClickCount > 0 || s.network5xxCount > 0)) {
+    tags.push("bug_suspected");
+    severity = bump(severity, "P1");
+  }
+  // server_error_impact
+  if (s.network5xxCount > 2) {
+    tags.push("server_error_impact");
+    severity = bump(severity, "P1");
+  }
+  // ux_friction
+  if (s.errorCount === 0 && (s.rageClickCount > 0 || s.deadClickCount > 0)) {
+    tags.push("ux_friction");
+    severity = bump(severity, "P2");
+  }
+  // upgrade_hesitation
+  if (s.startUrl && /\/(upgrade|subscription)/i.test(s.startUrl) && s.durationMs > 60_000) {
+    tags.push("upgrade_hesitation");
+    severity = bump(severity, "P2");
+  }
+  // api_key_friction
+  if (s.startUrl && /\/api-keys/i.test(s.startUrl) && s.durationMs > 60_000) {
+    tags.push("api_key_friction");
+    severity = bump(severity, "P3");
+  }
+
+  // onboarding_stuck: requires Sase RO lookup (caller provides ctx)
+  if (ctx && ctx.signupRecent && !ctx.firstVinQuery && s.durationMs > 120_000) {
+    tags.push("onboarding_stuck");
+    severity = bump(severity, "P2");
+  }
+
+  // power_user_path (rough heuristic — many clicks, auth, low error)
+  if (s.isAuthenticated && s.clickCount >= 20 && s.errorCount === 0 && tags.length === 0) {
+    tags.push("power_user_path");
+  }
+
+  return { tags, severity };
+}
+
+export function scoreSession(
+  s: SessionMeta,
+  sev: SeverityLevel,
+  novelty: number, // 0..1
+  patternStrength: number, // 0..1
+): number {
+  const severityWeight: Record<SeverityLevel, number> = {
+    P0: 1.0,
+    P1: 0.8,
+    P2: 0.5,
+    P3: 0.3,
+    INFO: 0.1,
+  };
+
+  const userValue = !s.isAuthenticated
+    ? 0.2
+    : s.subscriptionTier?.toLowerCase().includes("full")
+      ? 1.0
+      : s.subscriptionTier?.toLowerCase().includes("brand")
+        ? 0.7
+        : 0.5;
+
+  const ageHours = (Date.now() - s.startedAt.getTime()) / 3_600_000;
+  const recency = ageHours < 1 ? 1.0 : ageHours < 24 ? 0.7 : ageHours < 168 ? 0.4 : 0.2;
+
+  const raw =
+    severityWeight[sev] * 30 +
+    userValue * 25 +
+    novelty * 20 +
+    recency * 15 +
+    patternStrength * 10;
+
+  return Math.round(raw);
+}

apps/worker/src/schedulers/pipeline.ts

@@ -0,0 +1,54 @@
+import { Queue, Worker, type Job } from "bullmq";
+import { redis } from "../redis";
+import { runPostHogIngest } from "../jobs/posthog-ingest";
+import { runTagSessions } from "../jobs/tag-sessions";
+import { runCompressSessions } from "../jobs/compress-sessions";
+
+const QUEUE = "insight-pipeline";
+
+const queue = new Queue(QUEUE, { connection: redis });
+
+async function runJob(job: Job) {
+  switch (job.name) {
+    case "posthog-ingest": {
+      const res = await runPostHogIngest();
+      console.log(
+        `[pipeline] ingest fetched=${res.fetched} kept=${res.kept} discarded=${res.discarded} inserted=${res.inserted}${res.error ? ` error=${res.error}` : ""}`,
+      );
+      return res;
+    }
+    case "tag-sessions": {
+      const res = await runTagSessions();
+      if (res.tagged + res.discarded > 0) console.log(`[pipeline] tag tagged=${res.tagged} discarded=${res.discarded}`);
+      return res;
+    }
+    case "compress-sessions": {
+      const res = await runCompressSessions();
+      if (res.compressed + res.failed > 0) console.log(`[pipeline] compress ok=${res.compressed} fail=${res.failed}`);
+      return res;
+    }
+    default:
+      return { ok: false, error: `unknown job ${job.name}` };
+  }
+}
+
+export async function startInsightPipeline() {
+  await queue.upsertJobScheduler(
+    "posthog-ingest",
+    { pattern: "*/5 * * * *" },
+    { name: "posthog-ingest", data: {}, opts: { removeOnComplete: 50, removeOnFail: 25 } },
+  );
+  await queue.upsertJobScheduler(
+    "tag-sessions",
+    { pattern: "*/2 * * * *" },
+    { name: "tag-sessions", data: {}, opts: { removeOnComplete: 50, removeOnFail: 25 } },
+  );
+  await queue.upsertJobScheduler(
+    "compress-sessions",
+    { pattern: "*/3 * * * *" },
+    { name: "compress-sessions", data: {}, opts: { removeOnComplete: 50, removeOnFail: 25 } },
+  );
+
+  new Worker(QUEUE, runJob, { connection: redis, concurrency: 2 });
+  console.log("[pipeline] armed: posthog-ingest@*/5min, tag-sessions@*/2min, compress-sessions@*/3min");
+}