feat(insights): permanent archive of PostHog events + raw recordings

Phase A of the external observability archive. PostHog Cloud free tier
deletes data on a rolling window (events ~1yr, session recordings ~30d);
this mirrors both into our own infra permanently (hot Postgres + cold
MinIO gzip JSONL).

PostHog raw event archive:
- New `PosthogEvent` model (uuid PK → dedup, full properties JSONB,
  hot columns promoted + indexed).
- `posthog-event-archive` job: watermark-paged HogQL pull (cursor in the
  existing IngestionWatermark via a "sase:events" stream key), createMany
  + skipDuplicates for idempotency, 365d backfill. Daily closed-day cold
  dump to MinIO `posthog-archive/{project}/YYYY/MM/DD.jsonl.gz`.
- `hogqlQuery` helper added to posthog.ts. Timestamp cursor uses
  parseDateTimeBestEffort() — ClickHouse 500s on a raw ISO8601 literal
  (verified live against the project).

Raw rrweb recording preservation (separate from compress, which only
covers scored sessions and caps blobs):
- `SessionMeta.rrwebArchivedAt` / `rrwebArchiveKey`.
- `archive-recordings` job: every recording within 25d (margin before
  30d deletion), full blob range (no 12-cap), gzip → MinIO
  `rrweb-archive/{project}/YYYY/MM/DD/{sessionId}.jsonl.gz`.

- `putBuffer` gzip helper in minio.ts.
- Both jobs wired into pipeline.ts (event-archive@*/15min, recordings@*/6h).
- retention.ts untouched → new table + buckets persist forever (the goal).

Schema applies via the existing `prisma db push` on web start (additive:
new table + nullable columns).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

apps/worker/src/jobs/archive-recordings.ts

@@ -0,0 +1,89 @@
+import { gzipSync } from "node:zlib";
+import { prisma } from "../db";
+import { getSnapshotSources, getSnapshotBlob, isConfigured } from "../lib/posthog";
+import { putBuffer } from "../lib/minio";
+
+// Archive every recording's raw rrweb before PostHog's ~30-day deletion. Runs over
+// ALL recordings (incl. discarded ones) at full fidelity — unlike compress, which
+// only handles scored sessions and caps blobs for token budget.
+const RRWEB_BUCKET = process.env.RRWEB_ARCHIVE_BUCKET ?? "rrweb-archive";
+const BATCH = Number(process.env.INSIGHT_RECORDING_ARCHIVE_BATCH ?? "100");
+const MAX_AGE_DAYS = Number(process.env.INSIGHT_RECORDING_ARCHIVE_MAX_AGE_DAYS ?? "25");
+
+export type RecordingArchiveResult = {
+  scanned: number;
+  archived: number;
+  failed: number;
+  skipped: number;
+};
+
+export async function runArchiveRecordings(): Promise<RecordingArchiveResult> {
+  if (!isConfigured()) return { scanned: 0, archived: 0, failed: 0, skipped: 0 };
+
+  const floor = new Date(Date.now() - MAX_AGE_DAYS * 24 * 3600_000);
+  const pending = await prisma.sessionMeta.findMany({
+    where: { rrwebArchivedAt: null, startedAt: { gt: floor } },
+    orderBy: { startedAt: "asc" },
+    take: BATCH,
+  });
+  if (pending.length === 0) return { scanned: 0, archived: 0, failed: 0, skipped: 0 };
+
+  let archived = 0;
+  let failed = 0;
+  let skipped = 0;
+
+  for (const s of pending) {
+    try {
+      const sources = await getSnapshotSources(s.id);
+      if (sources.length === 0) {
+        // No snapshots (too short / already purged) — mark done so we stop retrying.
+        await prisma.sessionMeta.update({ where: { id: s.id }, data: { rrwebArchivedAt: new Date() } });
+        skipped++;
+        continue;
+      }
+
+      // Fetch the FULL blob range per source (no 12-cap) and concatenate raw JSONL.
+      const bySource = new Map<string, string[]>();
+      for (const src of sources) {
+        if (!bySource.has(src.source)) bySource.set(src.source, []);
+        bySource.get(src.source)!.push(src.blob_key);
+      }
+      const parts: string[] = [];
+      for (const [source, keys] of bySource) {
+        const sorted = [...keys].sort((a, b) => Number(a) - Number(b));
+        if (!sorted.length) continue;
+        const blob = await getSnapshotBlob(s.id, source, sorted[0], sorted[sorted.length - 1]);
+        if (blob) parts.push(blob);
+      }
+      const raw = parts.join("\n").trim();
+      if (!raw) {
+        await prisma.sessionMeta.update({ where: { id: s.id }, data: { rrwebArchivedAt: new Date() } });
+        skipped++;
+        continue;
+      }
+
+      const key = `${s.projectKey}/${dateFolder(s.startedAt)}/${s.id}.jsonl.gz`;
+      const gz = gzipSync(Buffer.from(raw, "utf-8"));
+      await putBuffer(RRWEB_BUCKET, key, gz, "application/gzip");
+
+      await prisma.sessionMeta.update({
+        where: { id: s.id },
+        data: { rrwebArchivedAt: new Date(), rrwebArchiveKey: key },
+      });
+      archived++;
+    } catch (e) {
+      // Transient — leave rrwebArchivedAt null so it retries next run.
+      console.warn("[archive-recordings] failed", s.id, (e as Error).message);
+      failed++;
+    }
+  }
+
+  return { scanned: pending.length, archived, failed, skipped };
+}
+
+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-event-archive.ts

@@ -0,0 +1,202 @@
+import { gzipSync } from "node:zlib";
+import { prisma } from "../db";
+import { hogqlQuery, isConfigured } from "../lib/posthog";
+import { putBuffer } from "../lib/minio";
+
+const PROJECT_KEY = process.env.INSIGHT_PROJECT_KEY ?? "sase";
+// Separate watermark stream so it never collides with the recording-ingest watermark (PROJECT_KEY).
+const WM_KEY = `${PROJECT_KEY}:events`;
+const BACKFILL_DAYS = Number(process.env.INSIGHT_EVENT_BACKFILL_DAYS ?? "365");
+const BATCH_SIZE = Number(process.env.INSIGHT_EVENT_BATCH ?? "500");
+const MAX_PAGES = Number(process.env.INSIGHT_EVENT_MAX_PAGES ?? "20");
+const MAX_DUMP_DAYS = Number(process.env.INSIGHT_EVENT_MAX_DUMP_DAYS ?? "120");
+const ARCHIVE_BUCKET = process.env.INSIGHT_ARCHIVE_BUCKET ?? "posthog-archive";
+
+export type EventArchiveResult = {
+  fetched: number;
+  inserted: number;
+  duplicates: number;
+  pages: number;
+  cursorAdvancedTo: string | null;
+  dumped: number;
+  error?: string;
+};
+
+type EventRow = {
+  uuid: string;
+  projectKey: string;
+  event: string;
+  distinctId: string;
+  personId: string | null;
+  sessionId: string | null;
+  timestamp: Date;
+  properties: unknown;
+};
+
+// Coerce HogQL's properties cell (object or JSON string) into a plain object.
+function parseProps(v: unknown): unknown {
+  if (v && typeof v === "object") return v;
+  if (typeof v === "string" && v) {
+    try {
+      return JSON.parse(v);
+    } catch {
+      return {};
+    }
+  }
+  return {};
+}
+
+function emptyOrNull(v: unknown): string | null {
+  const s = v == null ? "" : String(v);
+  return s ? s : null;
+}
+
+export async function runArchivePosthogEvents(): Promise<EventArchiveResult> {
+  if (!isConfigured()) {
+    return { fetched: 0, inserted: 0, duplicates: 0, pages: 0, cursorAdvancedTo: null, dumped: 0, error: "posthog_not_configured" };
+  }
+
+  const now = new Date();
+  const wm = await prisma.ingestionWatermark.findUnique({ where: { projectKey: WM_KEY } });
+  let cursor = wm?.posthogCursor
+    ? new Date(wm.posthogCursor)
+    : new Date(now.getTime() - BACKFILL_DAYS * 24 * 3600_000);
+
+  let fetched = 0;
+  let inserted = 0;
+  let duplicates = 0;
+  let pages = 0;
+  let error: string | undefined;
+
+  for (let page = 0; page < MAX_PAGES; page++) {
+    let resp;
+    try {
+      // properties.$session_id is the 5th column; full properties is the 7th.
+      // parseDateTimeBestEffort is required — ClickHouse can't compare timestamp
+      // against a raw ISO8601 string literal (the trailing Z / ms 500s the query).
+      resp = await hogqlQuery(
+        `SELECT uuid, event, distinct_id, person_id, properties.$session_id, timestamp, properties
+         FROM events
+         WHERE timestamp >= parseDateTimeBestEffort('${cursor.toISOString()}')
+         ORDER BY timestamp ASC
+         LIMIT ${BATCH_SIZE}`,
+      );
+    } catch (e) {
+      error = (e as Error).message;
+      break;
+    }
+    pages++;
+    const rows = resp.results ?? [];
+    if (rows.length === 0) break;
+
+    const batch: EventRow[] = [];
+    let maxTs = cursor;
+    for (const r of rows) {
+      const uuid = String(r[0] ?? "");
+      if (!uuid) continue;
+      const ts = new Date(String(r[5]));
+      if (ts > maxTs) maxTs = ts;
+      batch.push({
+        uuid,
+        projectKey: PROJECT_KEY,
+        event: String(r[1] ?? ""),
+        distinctId: String(r[2] ?? ""),
+        personId: emptyOrNull(r[3]),
+        sessionId: emptyOrNull(r[4]),
+        timestamp: ts,
+        properties: parseProps(r[6]) as object,
+      });
+    }
+    fetched += batch.length;
+
+    if (batch.length) {
+      const res = await prisma.posthogEvent.createMany({
+        data: batch as never,
+        skipDuplicates: true,
+      });
+      inserted += res.count;
+      duplicates += batch.length - res.count;
+    }
+
+    // Advance cursor to the batch's max timestamp. Using >= on the next query
+    // re-reads same-second events, but uuid PK + skipDuplicates absorbs them.
+    if (maxTs > cursor) {
+      cursor = maxTs;
+    } else {
+      // Safety valve: a full all-duplicate batch with no time progress would
+      // otherwise loop forever. Nudge cursor past it.
+      cursor = new Date(cursor.getTime() + 1);
+    }
+
+    if (rows.length < BATCH_SIZE) break;
+  }
+
+  // Persist cursor.
+  await prisma.ingestionWatermark.upsert({
+    where: { projectKey: WM_KEY },
+    create: { projectKey: WM_KEY, lastPolledAt: now, posthogCursor: cursor.toISOString() },
+    update: { lastPolledAt: now, posthogCursor: cursor.toISOString() },
+  });
+
+  // Cold dump: write closed (before-today, UTC) days to MinIO as gzip JSONL.
+  let dumped = 0;
+  try {
+    dumped = await dumpClosedDays(now);
+  } catch (e) {
+    error = error ?? `dump_failed: ${(e as Error).message}`;
+  }
+
+  return { fetched, inserted, duplicates, pages, cursorAdvancedTo: cursor.toISOString(), dumped, error };
+}
+
+// Groups undumped rows (timestamp < start-of-today-UTC) by UTC day, writes the
+// full day as one gzip JSONL object, then marks those rows dumped. Re-running is
+// deterministic: each closed day maps to exactly one object that is overwritten.
+async function dumpClosedDays(now: Date): Promise<number> {
+  const todayStart = new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), now.getUTCDate()));
+
+  const undumped = await prisma.posthogEvent.findMany({
+    where: { projectKey: PROJECT_KEY, dumpedAt: null, timestamp: { lt: todayStart } },
+    select: { timestamp: true },
+    orderBy: { timestamp: "asc" },
+  });
+  if (undumped.length === 0) return 0;
+
+  const days = Array.from(new Set(undumped.map((r) => r.timestamp.toISOString().slice(0, 10)))).slice(0, MAX_DUMP_DAYS);
+
+  let dumped = 0;
+  for (const day of days) {
+    const dayStart = new Date(`${day}T00:00:00.000Z`);
+    const dayEnd = new Date(dayStart.getTime() + 24 * 3600_000);
+    const rows = await prisma.posthogEvent.findMany({
+      where: { projectKey: PROJECT_KEY, timestamp: { gte: dayStart, lt: dayEnd } },
+      orderBy: { timestamp: "asc" },
+    });
+    if (rows.length === 0) continue;
+
+    const jsonl = rows
+      .map((r) =>
+        JSON.stringify({
+          uuid: r.uuid,
+          event: r.event,
+          distinctId: r.distinctId,
+          personId: r.personId,
+          sessionId: r.sessionId,
+          timestamp: r.timestamp.toISOString(),
+          properties: r.properties,
+        }),
+      )
+      .join("\n");
+    const gz = gzipSync(Buffer.from(jsonl, "utf-8"));
+    const [y, m, d] = day.split("-");
+    const key = `${PROJECT_KEY}/${y}/${m}/${d}.jsonl.gz`;
+    await putBuffer(ARCHIVE_BUCKET, key, gz, "application/gzip");
+
+    const marked = await prisma.posthogEvent.updateMany({
+      where: { projectKey: PROJECT_KEY, dumpedAt: null, timestamp: { gte: dayStart, lt: dayEnd } },
+      data: { dumpedAt: new Date() },
+    });
+    dumped += marked.count;
+  }
+  return dumped;
+}

apps/worker/src/lib/minio.ts

@@ -41,6 +41,18 @@ export async function putText(
   await c.putObject(bucket, key, buf, buf.length, { "Content-Type": contentType });
 }
 
+export async function putBuffer(
+  bucket: string,
+  key: string,
+  buf: Buffer,
+  contentType = "application/octet-stream",
+): Promise<void> {
+  const c = getMinio();
+  if (!c) throw new Error("minio_not_configured");
+  await ensureBucket(bucket);
+  await c.putObject(bucket, key, buf, buf.length, { "Content-Type": contentType });
+}
+
 export async function getText(bucket: string, key: string): Promise<string> {
   const c = getMinio();
   if (!c) throw new Error("minio_not_configured");

apps/worker/src/lib/posthog.ts

@@ -99,6 +99,24 @@ export function isConfigured(): boolean {
   return Boolean(TOKEN && PROJECT_ID);
 }
 
+// ---------- HogQL query API ----------
+
+export type HogQLResponse = { columns: string[]; results: unknown[][] };
+
+// Runs a HogQL query against the project and returns columns + row arrays.
+// Used by the archival pipeline to page the full event stream.
+export async function hogqlQuery(query: string): Promise<HogQLResponse> {
+  const url = `${HOST}/api/projects/${PROJECT_ID}/query/`;
+  const res = await fetch(url, {
+    method: "POST",
+    headers: { ...headers(), "Content-Type": "application/json" },
+    body: JSON.stringify({ query: { kind: "HogQLQuery", query } }),
+  });
+  if (!res.ok) throw new Error(`posthog hogql ${res.status}: ${await res.text().catch(() => "")}`);
+  const data = (await res.json()) as { columns?: string[]; results?: unknown[][] };
+  return { columns: data.columns ?? [], results: data.results ?? [] };
+}
+
 // ---------- Custom events ----------
 
 export type PHCustomEvent = {

apps/worker/src/schedulers/pipeline.ts

@@ -10,6 +10,8 @@ import { runRetention } from "../jobs/retention";
 import { runEvalSet } from "../jobs/eval-run";
 import { runDailyBrief } from "../jobs/daily-brief";
 import { runVinAnomalyDetect } from "../jobs/vin-anomaly";
+import { runArchivePosthogEvents } from "../jobs/posthog-event-archive";
+import { runArchiveRecordings } from "../jobs/archive-recordings";
 
 const QUEUE = "insight-pipeline";
 
@@ -87,6 +89,24 @@ async function runJob(job: Job) {
       }
       return res;
     }
+    case "posthog-event-archive": {
+      const res = await runArchivePosthogEvents();
+      if (res.fetched > 0 || res.dumped > 0 || res.error) {
+        console.log(
+          `[pipeline] event-archive fetched=${res.fetched} inserted=${res.inserted} dup=${res.duplicates} pages=${res.pages} dumped=${res.dumped}${res.error ? ` error=${res.error}` : ""}`,
+        );
+      }
+      return res;
+    }
+    case "archive-recordings": {
+      const res = await runArchiveRecordings();
+      if (res.scanned > 0) {
+        console.log(
+          `[pipeline] archive-recordings scanned=${res.scanned} archived=${res.archived} skipped=${res.skipped} failed=${res.failed}`,
+        );
+      }
+      return res;
+    }
     default:
       return { ok: false, error: `unknown job ${job.name}` };
   }
@@ -138,6 +158,16 @@ export async function startInsightPipeline() {
     { pattern: "*/5 * * * *" },
     { name: "vin-anomaly-detect", data: {}, opts: { removeOnComplete: 30, removeOnFail: 15 } },
   );
+  await queue.upsertJobScheduler(
+    "posthog-event-archive",
+    { pattern: "*/15 * * * *" },
+    { name: "posthog-event-archive", data: {}, opts: { removeOnComplete: 50, removeOnFail: 25 } },
+  );
+  await queue.upsertJobScheduler(
+    "archive-recordings",
+    { pattern: "0 */6 * * *" },
+    { name: "archive-recordings", data: {}, opts: { removeOnComplete: 30, removeOnFail: 15 } },
+  );
 
   new Worker(QUEUE, runJob, {
     connection: redis,
@@ -146,6 +176,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, vin-anomaly-detect@*/5min",
+    "[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, posthog-event-archive@*/15min, archive-recordings@*/6h",
   );
 }