Merge pull request 'feat(insights): permanent archive of PostHog (events+recordings+identity) + Sentry (Phase A+B+C)' (#2) from feat/observability-archive into main

2026-05-27 19:13:31 +00:00
parent fb33692452 928728dc66
commit 7eca679ab8
10 changed files with 939 additions and 1 deletions
--- a/apps/web/prisma/schema.prisma
+++ b/apps/web/prisma/schema.prisma
@@ -142,6 +142,10 @@ model SessionMeta {
  processedAt       DateTime?
  createdAt         DateTime @default(now())

+  // Raw rrweb recording preservation (archived to MinIO before PostHog's ~30-day deletion).
+  rrwebArchivedAt   DateTime?
+  rrwebArchiveKey   String?
+
  compressed        CompressedSession?

  @@index([status, createdAt])
@@ -445,3 +449,110 @@ model SaseUserNote {
  @@index([saseUserId, pinned, createdAt])
  @@map("sase_user_notes")
 }
+
+// ---------- Phase 9: External Observability Archive ----------
+
+// PostHog raw event stream mirrored to our DB permanently (free tier deletes
+// events on a rolling window). Lossless: full properties kept as JSONB, with
+// hot fields promoted to indexed columns. Also cold-dumped to MinIO as gzip JSONL.
+model PosthogEvent {
+  uuid        String   @id            // PostHog event uuid → natural dedup key
+  projectKey  String                  // "sase"
+  event       String
+  distinctId  String
+  personId    String?
+  sessionId   String?                 // $session_id — joins to SessionMeta.id
+  timestamp   DateTime                // PostHog event time
+  properties  Json                    // full properties — lossless
+  ingestedAt  DateTime @default(now())
+  dumpedAt    DateTime?               // set once written to MinIO cold partition
+
+  @@index([projectKey, timestamp])
+  @@index([event, timestamp])
+  @@index([distinctId, timestamp])
+  @@index([dumpedAt])
+  @@map("posthog_events")
+}
+
+// Append-only history of PostHog person properties. Free tier overwrites the
+// live person; a new row is written only when properties change (propertiesHash),
+// so this is a compact timeline of how each person evolved.
+model PosthogPersonSnapshot {
+  id             String   @id @default(cuid())
+  projectKey     String
+  personId       String                  // PostHog person uuid (persons.id) — joins PosthogEvent.personId
+  distinctIds    String[]                // all distinct_ids merged into this person
+  propertiesHash String                  // change-detection key
+  properties     Json
+  capturedAt     DateTime @default(now())
+
+  @@index([personId, capturedAt])
+  @@index([projectKey, capturedAt])
+  @@map("posthog_person_snapshots")
+}
+
+// Append-only history of cohort definitions + membership counts. New row only
+// when count or filters change (stateHash).
+model PosthogCohortSnapshot {
+  id          String   @id @default(cuid())
+  projectKey  String
+  cohortId    Int
+  name        String
+  count       Int?
+  isStatic    Boolean  @default(false)
+  stateHash   String                     // hash of count + filters → change-detection
+  filters     Json
+  capturedAt  DateTime @default(now())
+
+  @@index([cohortId, capturedAt])
+  @@index([projectKey, capturedAt])
+  @@map("posthog_cohort_snapshots")
+}
+
+// ---------- Phase C: Sentry archive ----------
+
+// Sentry issue (group) aggregate state. Upserted to the latest snapshot — the
+// raw history lives in SentryEvent; the issue row is the current grouping/metadata.
+model SentryIssue {
+  id           String   @id          // Sentry issue/group id
+  projectKey   String                // "sase"
+  shortId      String?
+  title        String?
+  culprit      String?
+  level        String?
+  status       String?
+  type         String?
+  count        Int?                  // total events as of lastSyncedAt
+  userCount    Int?
+  firstSeen    DateTime?
+  lastSeen     DateTime?
+  permalink    String?
+  metadata     Json?
+  ingestedAt   DateTime @default(now())
+  lastSyncedAt DateTime @default(now())
+
+  @@index([projectKey, lastSeen])
+  @@map("sentry_issues")
+}
+
+// Individual Sentry events (the at-risk raw occurrences; free tier prunes ~30d).
+// Lossless full payload as JSONB; also cold-dumped to MinIO.
+model SentryEvent {
+  eventId     String   @id          // Sentry event id (32 hex)
+  issueId     String?               // parent issue/group id
+  projectKey  String
+  title       String?
+  message     String?
+  level       String?
+  platform    String?
+  timestamp   DateTime               // event occurrence time
+  tags        Json?
+  payload     Json                   // full event payload — lossless
+  ingestedAt  DateTime @default(now())
+  dumpedAt    DateTime?
+
+  @@index([projectKey, timestamp])
+  @@index([issueId, timestamp])
+  @@index([dumpedAt])
+  @@map("sentry_events")
+}
--- a/apps/worker/src/jobs/archive-recordings.ts
+++ b/apps/worker/src/jobs/archive-recordings.ts
@@ -0,0 +1,114 @@
+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 ?? "60");
+const MAX_AGE_DAYS = Number(process.env.INSIGHT_RECORDING_ARCHIVE_MAX_AGE_DAYS ?? "25");
+// PostHog's snapshot API is aggressively rate-limited. Pace requests and bail out
+// of the run on sustained 429s — unarchived rows keep rrwebArchivedAt=null and are
+// retried next cycle (the 25d window vs */6h cadence gives ample slack).
+const RATE_DELAY_MS = Number(process.env.INSIGHT_RECORDING_ARCHIVE_DELAY_MS ?? "500");
+// blob_v2 rejects wide blob-key ranges (a 23-key span 400s, a 5-key span is fine),
+// so fetch in small contiguous chunks and concatenate.
+const BLOB_CHUNK = Number(process.env.INSIGHT_RECORDING_BLOB_CHUNK ?? "10");
+const CHUNK_DELAY_MS = Number(process.env.INSIGHT_RECORDING_CHUNK_DELAY_MS ?? "150");
+
+export type RecordingArchiveResult = {
+  scanned: number;
+  archived: number;
+  failed: number;
+  skipped: number;
+  rateLimited: boolean;
+};
+
+const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
+const isRateLimit = (e: unknown) => (e as Error)?.message?.includes(" 429");
+
+export async function runArchiveRecordings(): Promise<RecordingArchiveResult> {
+  if (!isConfigured()) return { scanned: 0, archived: 0, failed: 0, skipped: 0, rateLimited: false };
+
+  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, rateLimited: false };
+
+  let archived = 0;
+  let failed = 0;
+  let skipped = 0;
+  let rateLimited = false;
+  let processed = 0;
+
+  for (const s of pending) {
+    if (processed > 0) await sleep(RATE_DELAY_MS);
+    processed++;
+    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 ALL blobs per source (no 12-cap), chunked to dodge blob_v2's range limit.
+      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));
+        for (let i = 0; i < sorted.length; i += BLOB_CHUNK) {
+          if (i > 0) await sleep(CHUNK_DELAY_MS);
+          const lo = sorted[i];
+          const hi = sorted[Math.min(i + BLOB_CHUNK - 1, sorted.length - 1)];
+          const blob = await getSnapshotBlob(s.id, source, lo, hi);
+          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) {
+      if (isRateLimit(e)) {
+        // Rate limited — stop hammering. Remaining rows retry next cycle.
+        rateLimited = true;
+        break;
+      }
+      // Other transient errors — 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, rateLimited };
+}
+
+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}`;
+}
--- a/apps/worker/src/jobs/posthog-event-archive.ts
+++ b/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;
+}
--- a/apps/worker/src/jobs/posthog-identity-archive.ts
+++ b/apps/worker/src/jobs/posthog-identity-archive.ts
@@ -0,0 +1,97 @@
+import { prisma } from "../db";
+import { listPersons, listCohorts, isConfigured } from "../lib/posthog";
+import { stableHash } from "../lib/hash";
+
+const PROJECT_KEY = process.env.INSIGHT_PROJECT_KEY ?? "sase";
+
+export type IdentityArchiveResult = {
+  personsScanned: number;
+  personSnapshots: number;
+  cohortsScanned: number;
+  cohortSnapshots: number;
+  error?: string;
+};
+
+// Daily append-only snapshot of PostHog persons + cohorts. A new row is written
+// only when the payload changed since the last snapshot (hash compare), so the
+// table stays a compact change history rather than a daily full copy.
+export async function runArchiveIdentity(): Promise<IdentityArchiveResult> {
+  if (!isConfigured()) {
+    return { personsScanned: 0, personSnapshots: 0, cohortsScanned: 0, cohortSnapshots: 0, error: "posthog_not_configured" };
+  }
+
+  let personsScanned = 0;
+  let personSnapshots = 0;
+  let cohortsScanned = 0;
+  let cohortSnapshots = 0;
+  let error: string | undefined;
+
+  // ── Persons ──
+  try {
+    const persons = await listPersons();
+    personsScanned = persons.length;
+
+    // Latest stored hash per person (one query via DISTINCT ON).
+    const latest = await prisma.$queryRaw<Array<{ personId: string; propertiesHash: string }>>`
+      SELECT DISTINCT ON ("personId") "personId", "propertiesHash"
+      FROM posthog_person_snapshots
+      WHERE "projectKey" = ${PROJECT_KEY}
+      ORDER BY "personId", "capturedAt" DESC
+    `;
+    const latestHash = new Map(latest.map((r) => [r.personId, r.propertiesHash]));
+
+    const toInsert = persons
+      .filter((p) => p.id)
+      .map((p) => ({
+        projectKey: PROJECT_KEY,
+        personId: p.id,
+        distinctIds: p.distinct_ids ?? [],
+        propertiesHash: stableHash(p.properties ?? {}),
+        properties: (p.properties ?? {}) as object,
+      }))
+      .filter((row) => latestHash.get(row.personId) !== row.propertiesHash);
+
+    if (toInsert.length) {
+      const res = await prisma.posthogPersonSnapshot.createMany({ data: toInsert as never });
+      personSnapshots = res.count;
+    }
+  } catch (e) {
+    error = `persons_failed: ${(e as Error).message}`;
+  }
+
+  // ── Cohorts ──
+  try {
+    const cohorts = await listCohorts();
+    cohortsScanned = cohorts.length;
+
+    const latest = await prisma.$queryRaw<Array<{ cohortId: number; stateHash: string }>>`
+      SELECT DISTINCT ON ("cohortId") "cohortId", "stateHash"
+      FROM posthog_cohort_snapshots
+      WHERE "projectKey" = ${PROJECT_KEY}
+      ORDER BY "cohortId", "capturedAt" DESC
+    `;
+    const latestHash = new Map(latest.map((r) => [r.cohortId, r.stateHash]));
+
+    const toInsert = cohorts.map((c) => {
+      const filters = c.filters ?? c.groups ?? {};
+      return {
+        projectKey: PROJECT_KEY,
+        cohortId: c.id,
+        name: c.name ?? "",
+        count: c.count ?? null,
+        isStatic: Boolean(c.is_static),
+        stateHash: stableHash({ count: c.count ?? null, filters }),
+        filters: filters as object,
+      };
+    }).filter((row) => latestHash.get(row.cohortId) !== row.stateHash);
+
+    if (toInsert.length) {
+      const res = await prisma.posthogCohortSnapshot.createMany({ data: toInsert as never });
+      cohortSnapshots = res.count;
+    }
+  } catch (e) {
+    error = error ?? `cohorts_failed: ${(e as Error).message}`;
+  }
+
+  return { personsScanned, personSnapshots, cohortsScanned, cohortSnapshots, error };
+}
--- a/apps/worker/src/jobs/sentry-archive.ts
+++ b/apps/worker/src/jobs/sentry-archive.ts
@@ -0,0 +1,173 @@
+import { gzipSync } from "node:zlib";
+import { prisma } from "../db";
+import { isConfigured, listIssuesPage, listEventsPage, type SentryEventRaw } from "../lib/sentry";
+import { putBuffer } from "../lib/minio";
+
+const PROJECT_KEY = process.env.INSIGHT_PROJECT_KEY ?? "sase";
+const ISSUE_MAX_PAGES = Number(process.env.SENTRY_ISSUE_MAX_PAGES ?? "20");
+const EVENT_MAX_PAGES = Number(process.env.SENTRY_EVENT_MAX_PAGES ?? "40");
+const MAX_DUMP_DAYS = Number(process.env.SENTRY_MAX_DUMP_DAYS ?? "120");
+const ARCHIVE_BUCKET = process.env.SENTRY_ARCHIVE_BUCKET ?? "sentry-archive";
+
+export type SentryArchiveResult = {
+  issuesSynced: number;
+  eventsFetched: number;
+  eventsInserted: number;
+  eventsDuplicate: number;
+  dumped: number;
+  error?: string;
+};
+
+function toDate(v: unknown): Date | null {
+  if (!v) return null;
+  const d = new Date(String(v));
+  return Number.isNaN(d.getTime()) ? null : d;
+}
+
+export async function runArchiveSentry(): Promise<SentryArchiveResult> {
+  if (!isConfigured()) {
+    return { issuesSynced: 0, eventsFetched: 0, eventsInserted: 0, eventsDuplicate: 0, dumped: 0, error: "sentry_not_configured" };
+  }
+
+  let issuesSynced = 0;
+  let eventsFetched = 0;
+  let eventsInserted = 0;
+  let eventsDuplicate = 0;
+  let error: string | undefined;
+
+  // ── Issues: upsert latest aggregate state ──
+  try {
+    let cursor: string | undefined;
+    for (let page = 0; page < ISSUE_MAX_PAGES; page++) {
+      const { data, nextCursor } = await listIssuesPage(cursor);
+      if (!data.length) break;
+      for (const i of data) {
+        if (!i.id) continue;
+        const row = {
+          projectKey: PROJECT_KEY,
+          shortId: i.shortId ?? null,
+          title: i.title ?? null,
+          culprit: i.culprit ?? null,
+          level: i.level ?? null,
+          status: i.status ?? null,
+          type: i.type ?? null,
+          count: i.count != null ? Number(i.count) : null,
+          userCount: i.userCount ?? null,
+          firstSeen: toDate(i.firstSeen),
+          lastSeen: toDate(i.lastSeen),
+          permalink: i.permalink ?? null,
+          metadata: (i.metadata ?? {}) as object,
+          lastSyncedAt: new Date(),
+        };
+        await prisma.sentryIssue.upsert({
+          where: { id: i.id },
+          create: { id: i.id, ...row },
+          update: row,
+        });
+        issuesSynced++;
+      }
+      if (!nextCursor) break;
+      cursor = nextCursor;
+    }
+  } catch (e) {
+    error = `issues_failed: ${(e as Error).message}`;
+  }
+
+  // ── Events: newest-first, dedup by eventId, stop once caught up ──
+  try {
+    let cursor: string | undefined;
+    for (let page = 0; page < EVENT_MAX_PAGES; page++) {
+      const { data, nextCursor } = await listEventsPage(cursor);
+      if (!data.length) break;
+
+      const batch = data
+        .map((e: SentryEventRaw) => {
+          const eventId = String(e.eventID ?? e.id ?? "");
+          const ts = toDate(e.dateCreated ?? e.dateReceived);
+          if (!eventId || !ts) return null;
+          return {
+            eventId,
+            issueId: e.groupID ? String(e.groupID) : null,
+            projectKey: PROJECT_KEY,
+            title: e.title ?? null,
+            message: typeof e.message === "string" ? e.message : null,
+            level: typeof e.level === "string" ? (e.level as string) : null,
+            platform: e.platform ?? null,
+            timestamp: ts,
+            tags: (e.tags ?? []) as object,
+            payload: e as object,
+          };
+        })
+        .filter((r): r is NonNullable<typeof r> => r !== null);
+
+      eventsFetched += batch.length;
+      if (batch.length) {
+        const res = await prisma.sentryEvent.createMany({ data: batch as never, skipDuplicates: true });
+        eventsInserted += res.count;
+        eventsDuplicate += batch.length - res.count;
+        // A full page with no new rows means we've reached already-archived events.
+        if (res.count === 0) break;
+      }
+      if (!nextCursor) break;
+      cursor = nextCursor;
+    }
+  } catch (e) {
+    error = error ?? `events_failed: ${(e as Error).message}`;
+  }
+
+  // ── Cold dump closed days to MinIO ──
+  let dumped = 0;
+  try {
+    dumped = await dumpClosedDays(new Date());
+  } catch (e) {
+    error = error ?? `dump_failed: ${(e as Error).message}`;
+  }
+
+  return { issuesSynced, eventsFetched, eventsInserted, eventsDuplicate, dumped, error };
+}
+
+async function dumpClosedDays(now: Date): Promise<number> {
+  const todayStart = new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), now.getUTCDate()));
+  const undumped = await prisma.sentryEvent.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.sentryEvent.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({
+          eventId: r.eventId,
+          issueId: r.issueId,
+          title: r.title,
+          message: r.message,
+          level: r.level,
+          platform: r.platform,
+          timestamp: r.timestamp.toISOString(),
+          tags: r.tags,
+          payload: r.payload,
+        }),
+      )
+      .join("\n");
+    const gz = gzipSync(Buffer.from(jsonl, "utf-8"));
+    const [y, m, d] = day.split("-");
+    await putBuffer(ARCHIVE_BUCKET, `${PROJECT_KEY}/${y}/${m}/${d}.jsonl.gz`, gz, "application/gzip");
+    const marked = await prisma.sentryEvent.updateMany({
+      where: { projectKey: PROJECT_KEY, dumpedAt: null, timestamp: { gte: dayStart, lt: dayEnd } },
+      data: { dumpedAt: new Date() },
+    });
+    dumped += marked.count;
+  }
+  return dumped;
+}
--- a/apps/worker/src/lib/hash.ts
+++ b/apps/worker/src/lib/hash.ts
@@ -14,3 +14,17 @@ export function fingerprintHash(parts: Array<string | number | null | undefined>
    .join("|");
  return createHash("sha256").update(norm).digest("hex").slice(0, 24);
 }
+
+// Deterministic hash of an arbitrary JSON value (object keys sorted) — used for
+// change detection so re-serialized-but-equal payloads don't produce false diffs.
+function stableStringify(v: unknown): string {
+  if (v === null || typeof v !== "object") return JSON.stringify(v) ?? "null";
+  if (Array.isArray(v)) return `[${v.map(stableStringify).join(",")}]`;
+  const obj = v as Record<string, unknown>;
+  const keys = Object.keys(obj).sort();
+  return `{${keys.map((k) => `${JSON.stringify(k)}:${stableStringify(obj[k])}`).join(",")}}`;
+}
+
+export function stableHash(value: unknown): string {
+  return createHash("sha256").update(stableStringify(value)).digest("hex").slice(0, 32);
+}
--- a/apps/worker/src/lib/minio.ts
+++ b/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");
--- a/apps/worker/src/lib/posthog.ts
+++ b/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 = {
@@ -159,6 +177,55 @@ export async function getPerson(distinctId: string): Promise<PHPerson | null> {
  return data.results?.[0] ?? null;
 }

+// ---------- Person & cohort listing (archival snapshots) ----------
+
+export type PHPersonFull = {
+  id: string; // PostHog person uuid (== events.person_id)
+  name?: string;
+  distinct_ids: string[];
+  properties: Record<string, unknown>;
+};
+
+// Lists all persons via the REST endpoint, following `next` pagination.
+export async function listPersons(opts?: { limit?: number; maxPages?: number }): Promise<PHPersonFull[]> {
+  const limit = opts?.limit ?? 100;
+  const maxPages = opts?.maxPages ?? 50;
+  const out: PHPersonFull[] = [];
+  let next: string | null = `${HOST}/api/projects/${PROJECT_ID}/persons/?limit=${limit}`;
+  for (let page = 0; page < maxPages && next; page++) {
+    const res = await fetch(next, { headers: headers() });
+    if (!res.ok) throw new Error(`posthog persons ${res.status}`);
+    const data = (await res.json()) as { results?: PHPersonFull[]; next?: string | null };
+    out.push(...(data.results ?? []));
+    next = data.next ?? null;
+  }
+  return out;
+}
+
+export type PHCohort = {
+  id: number;
+  name: string;
+  count?: number | null;
+  is_static?: boolean;
+  filters?: unknown;
+  groups?: unknown;
+};
+
+// Lists cohort definitions (with membership counts), following `next` pagination.
+export async function listCohorts(opts?: { maxPages?: number }): Promise<PHCohort[]> {
+  const maxPages = opts?.maxPages ?? 20;
+  const out: PHCohort[] = [];
+  let next: string | null = `${HOST}/api/projects/${PROJECT_ID}/cohorts/`;
+  for (let page = 0; page < maxPages && next; page++) {
+    const res = await fetch(next, { headers: headers() });
+    if (!res.ok) throw new Error(`posthog cohorts ${res.status}`);
+    const data = (await res.json()) as { results?: PHCohort[]; next?: string | null };
+    out.push(...(data.results ?? []));
+    next = data.next ?? null;
+  }
+  return out;
+}
+
 export async function getGroup(
  groupType: string,
  groupKey: string,
--- a/apps/worker/src/lib/sentry.ts
+++ b/apps/worker/src/lib/sentry.ts
@@ -0,0 +1,88 @@
+// Sentry API client (read-only) for the archival pipeline. Uses a User Auth
+// Token with org:read / project:read / event:read. EU-region orgs use the
+// de.sentry.io API base.
+const TOKEN = process.env.SENTRY_AUTH_TOKEN ?? "";
+const ORG = process.env.SENTRY_ORG ?? "";
+const PROJECT = process.env.SENTRY_PROJECT ?? "";
+const API_BASE = (process.env.SENTRY_API_BASE ?? "https://sentry.io/api/0").replace(/\/$/, "");
+
+export function isConfigured(): boolean {
+  return Boolean(TOKEN && ORG && PROJECT);
+}
+
+const headers = () => ({ Authorization: `Bearer ${TOKEN}`, Accept: "application/json" });
+
+// Sentry paginates via a Link header: <url>; rel="next"; results="true"; cursor="...".
+// Returns the next cursor only when results="true".
+function parseNextCursor(link: string | null): string | null {
+  if (!link) return null;
+  for (const part of link.split(",")) {
+    if (/rel="next"/.test(part) && /results="true"/.test(part)) {
+      const m = part.match(/cursor="([^"]+)"/);
+      if (m) return m[1];
+    }
+  }
+  return null;
+}
+
+type Page<T> = { data: T[]; nextCursor: string | null };
+
+async function getPage<T>(path: string, params: Record<string, string>): Promise<Page<T>> {
+  const url = new URL(`${API_BASE}${path}`);
+  for (const [k, v] of Object.entries(params)) if (v) url.searchParams.set(k, v);
+  const res = await fetch(url, { headers: headers() });
+  if (!res.ok) throw new Error(`sentry ${path} ${res.status}: ${await res.text().catch(() => "")}`);
+  const data = (await res.json()) as T[];
+  return { data, nextCursor: parseNextCursor(res.headers.get("link")) };
+}
+
+export type SentryIssueRaw = {
+  id: string;
+  shortId?: string;
+  title?: string;
+  culprit?: string;
+  level?: string;
+  status?: string;
+  type?: string;
+  count?: string | number;
+  userCount?: number;
+  firstSeen?: string;
+  lastSeen?: string;
+  permalink?: string;
+  metadata?: Record<string, unknown>;
+};
+
+export type SentryEventRaw = {
+  id?: string;
+  eventID?: string;
+  groupID?: string;
+  title?: string;
+  message?: string;
+  "event.type"?: string;
+  platform?: string;
+  dateCreated?: string;
+  dateReceived?: string;
+  tags?: unknown;
+  [k: string]: unknown;
+};
+
+// One page of org issues (newest activity first). statsPeriod bounds the window.
+export function listIssuesPage(cursor?: string, statsPeriod = "90d"): Promise<Page<SentryIssueRaw>> {
+  return getPage<SentryIssueRaw>(`/organizations/${ORG}/issues/`, {
+    project: "",
+    query: "",
+    statsPeriod,
+    limit: "100",
+    ...(cursor ? { cursor } : {}),
+  });
+}
+
+// One page of project events, full payloads, newest first.
+export function listEventsPage(cursor?: string): Promise<Page<SentryEventRaw>> {
+  return getPage<SentryEventRaw>(`/projects/${ORG}/${PROJECT}/events/`, {
+    full: "true",
+    ...(cursor ? { cursor } : {}),
+  });
+}
+
+export const sentryConfig = { ORG, PROJECT, API_BASE };
--- a/apps/worker/src/schedulers/pipeline.ts
+++ b/apps/worker/src/schedulers/pipeline.ts
@@ -10,6 +10,10 @@ 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";
+import { runArchiveIdentity } from "../jobs/posthog-identity-archive";
+import { runArchiveSentry } from "../jobs/sentry-archive";

 const QUEUE = "insight-pipeline";

@@ -87,6 +91,42 @@ 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}${res.rateLimited ? " rate_limited" : ""}`,
+        );
+      }
+      return res;
+    }
+    case "posthog-identity-archive": {
+      const res = await runArchiveIdentity();
+      if (res.personSnapshots > 0 || res.cohortSnapshots > 0 || res.error) {
+        console.log(
+          `[pipeline] identity-archive persons=${res.personsScanned}/${res.personSnapshots} cohorts=${res.cohortsScanned}/${res.cohortSnapshots}${res.error ? ` error=${res.error}` : ""}`,
+        );
+      }
+      return res;
+    }
+    case "sentry-archive": {
+      const res = await runArchiveSentry();
+      if (res.eventsInserted > 0 || res.issuesSynced > 0 || res.dumped > 0 || res.error) {
+        console.log(
+          `[pipeline] sentry-archive issues=${res.issuesSynced} events=${res.eventsFetched}/${res.eventsInserted} dup=${res.eventsDuplicate} dumped=${res.dumped}${res.error ? ` error=${res.error}` : ""}`,
+        );
+      }
+      return res;
+    }
    default:
      return { ok: false, error: `unknown job ${job.name}` };
  }
@@ -138,6 +178,26 @@ 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 } },
+  );
+  await queue.upsertJobScheduler(
+    "posthog-identity-archive",
+    { pattern: "0 3 * * *" },
+    { name: "posthog-identity-archive", data: {}, opts: { removeOnComplete: 30, removeOnFail: 15 } },
+  );
+  await queue.upsertJobScheduler(
+    "sentry-archive",
+    { pattern: "0 * * * *" },
+    { name: "sentry-archive", data: {}, opts: { removeOnComplete: 30, removeOnFail: 15 } },
+  );

  new Worker(QUEUE, runJob, {
    connection: redis,
@@ -146,6 +206,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, posthog-identity-archive@03:00, sentry-archive@hourly",
  );
 }