feat(archive): add groups + recordings to the PostHog archive

- groups: posthog-resource-archive now also snapshots group types + group
  instances (groups_types/ + groups/?group_type_index) into
  posthog_resource_snapshots (change-detected).
- recordings: new posthog-recording-archive job (@*/6h) captures complete
  metadata for EVERY recording into posthog_recordings (skipDuplicates over a
  rolling 90d window) + MinIO dump — a superset of the insight pipeline's
  promoted SessionMeta subset. rrweb blobs still archived by archive-recordings.
- /posthog-archive browser: Groups / Group Types tabs + a Recordings tab.

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

apps/worker/src/jobs/posthog-recording-archive.ts

@@ -0,0 +1,86 @@
+import { gzipSync } from "node:zlib";
+import { prisma } from "../db";
+import { listRecordings, isConfigured } from "../lib/posthog";
+import { putBuffer } from "../lib/minio";
+
+// Complete metadata archive of EVERY session recording — superset of the insight
+// pipeline's promoted SessionMeta rows. Pages the recording list over a rolling
+// window and inserts only new ids (skipDuplicates). rrweb blobs are archived
+// separately by archive-recordings for ingested sessions.
+const PROJECT_KEY = process.env.INSIGHT_PROJECT_KEY ?? "sase";
+const ARCHIVE_BUCKET = process.env.INSIGHT_ARCHIVE_BUCKET ?? "posthog-archive";
+const WINDOW_DAYS = Number(process.env.INSIGHT_RECORDING_META_WINDOW_DAYS ?? "90");
+const MAX_PAGES = Number(process.env.INSIGHT_RECORDING_META_MAX_PAGES ?? "40");
+const PAGE = 100;
+
+export type RecordingMetaArchiveResult = {
+  ok: boolean;
+  fetched: number;
+  inserted: number;
+  pages: number;
+  dumped: boolean;
+  reason?: string;
+};
+
+const int = (v: unknown): number | null => {
+  const n = Math.round(Number(v));
+  return Number.isFinite(n) ? n : null;
+};
+
+export async function runArchivePosthogRecordings(): Promise<RecordingMetaArchiveResult> {
+  if (!isConfigured()) {
+    return { ok: false, fetched: 0, inserted: 0, pages: 0, dumped: false, reason: "posthog_not_configured" };
+  }
+
+  const dateFrom = new Date(Date.now() - WINDOW_DAYS * 864e5).toISOString();
+  const dateTo = new Date(Date.now() - 3_600_000).toISOString(); // 1h buffer → finalized recordings only
+  let fetched = 0;
+  let inserted = 0;
+  let pages = 0;
+  const all: Array<Record<string, unknown>> = [];
+
+  for (let offset = 0; pages < MAX_PAGES; offset += PAGE, pages++) {
+    let resp;
+    try {
+      resp = await listRecordings({ dateFrom, dateTo, limit: PAGE, offset });
+    } catch (e) {
+      if (pages === 0) return { ok: false, fetched, inserted, pages, dumped: false, reason: (e as Error).message };
+      break; // mid-run page failure → keep what we have, retry next cycle
+    }
+    const items = resp.results ?? [];
+    if (items.length === 0) break;
+    fetched += items.length;
+    all.push(...(items as unknown as Array<Record<string, unknown>>));
+
+    const rows = items.map((r) => ({
+      id: r.id,
+      projectKey: PROJECT_KEY,
+      distinctId: r.distinct_id ?? null,
+      personName: r.person?.name ?? null,
+      startTime: r.start_time ? new Date(r.start_time) : null,
+      endTime: r.end_time ? new Date(r.end_time) : null,
+      durationSec: int(r.recording_duration),
+      activeSeconds: int(r.active_seconds),
+      clickCount: int(r.click_count),
+      keypressCount: int(r.keypress_count),
+      mouseActivityCount: int(r.mouse_activity_count),
+      consoleErrorCount: int(r.console_error_count),
+      startUrl: r.start_url ?? null,
+      ongoing: Boolean(r.ongoing),
+      properties: r as object,
+    }));
+    const res = await prisma.posthogRecording.createMany({ data: rows as never, skipDuplicates: true });
+    inserted += res.count;
+
+    if (!resp.has_next && !resp.next) break;
+  }
+
+  let dumped = false;
+  if (all.length) {
+    const day = new Date().toISOString().slice(0, 10);
+    await putBuffer(ARCHIVE_BUCKET, `recordings/${day}.json.gz`, gzipSync(Buffer.from(JSON.stringify(all))), "application/gzip");
+    dumped = true;
+  }
+
+  return { ok: true, fetched, inserted, pages, dumped };
+}

apps/worker/src/jobs/posthog-resource-archive.ts

@@ -1,6 +1,6 @@
 import { gzipSync } from "node:zlib";
 import { prisma } from "../db";
-import { listResource, isConfigured } from "../lib/posthog";
+import { listResource, listGroupTypes, listGroups, isConfigured } from "../lib/posthog";
 import { stableHash } from "../lib/hash";
 import { putBuffer } from "../lib/minio";
 
@@ -35,6 +35,23 @@ function hashConfig(item: Record<string, unknown>): string {
   return stableHash(clean);
 }
 
+type SnapRow = { projectKey: string; resourceType: string; resourceId: string; name: string | null; stateHash: string; data: object };
+
+// Insert only rows whose config changed since the latest stored snapshot.
+async function persistChanged(type: string, rows: SnapRow[]): Promise<number> {
+  const latest = await prisma.$queryRaw<Array<{ resourceId: string; stateHash: string }>>`
+    SELECT DISTINCT ON ("resourceId") "resourceId", "stateHash"
+    FROM posthog_resource_snapshots
+    WHERE "projectKey" = ${PROJECT_KEY} AND "resourceType" = ${type}
+    ORDER BY "resourceId", "capturedAt" DESC
+  `;
+  const latestHash = new Map(latest.map((x) => [x.resourceId, x.stateHash]));
+  const changed = rows.filter((row) => row.resourceId && latestHash.get(row.resourceId) !== row.stateHash);
+  if (!changed.length) return 0;
+  const res = await prisma.posthogResourceSnapshot.createMany({ data: changed as never });
+  return res.count;
+}
+
 export type ResourceArchiveResult = {
   ok: boolean;
   perType: Record<string, { scanned: number; snapshots: number; dumped: boolean; error?: string }>;
@@ -87,5 +104,53 @@ export async function runArchivePosthogResources(): Promise<ResourceArchiveResul
     }
   }
 
+  // ── Groups: group types + instances (separate API shape) ──
+  try {
+    const types = await listGroupTypes();
+    const typeRows: SnapRow[] = types
+      .map((t) => ({
+        projectKey: PROJECT_KEY,
+        resourceType: "group_type",
+        resourceId: String(t.group_type_index ?? t.group_type ?? ""),
+        name: String(t.name_singular ?? t.group_type ?? ""),
+        stateHash: hashConfig(t),
+        data: t as object,
+      }))
+      .filter((row) => row.resourceId);
+    perType["group_type"] = { scanned: typeRows.length, snapshots: await persistChanged("group_type", typeRows), dumped: false };
+    if (types.length) {
+      await putBuffer(ARCHIVE_BUCKET, `resources/group_type/${day}.json.gz`, gzipSync(Buffer.from(JSON.stringify(types))), "application/gzip");
+      perType["group_type"].dumped = true;
+    }
+
+    let allGroups: Array<Record<string, unknown>> = [];
+    const groupRows: SnapRow[] = [];
+    for (const t of types) {
+      const idx = Number(t.group_type_index ?? -1);
+      if (idx < 0) continue;
+      const groups = await listGroups(idx);
+      allGroups = allGroups.concat(groups);
+      for (const g of groups) {
+        const key = String(g.group_key ?? "");
+        if (!key) continue;
+        groupRows.push({
+          projectKey: PROJECT_KEY,
+          resourceType: "group",
+          resourceId: `${idx}:${key}`,
+          name: key,
+          stateHash: hashConfig(g),
+          data: g as object,
+        });
+      }
+    }
+    perType["group"] = { scanned: groupRows.length, snapshots: await persistChanged("group", groupRows), dumped: false };
+    if (allGroups.length) {
+      await putBuffer(ARCHIVE_BUCKET, `resources/group/${day}.json.gz`, gzipSync(Buffer.from(JSON.stringify(allGroups))), "application/gzip");
+      perType["group"].dumped = true;
+    }
+  } catch (e) {
+    perType["group"] = { scanned: 0, snapshots: 0, dumped: false, error: (e as Error).message };
+  }
+
   return { ok: true, perType };
 }

apps/worker/src/lib/posthog.ts

@@ -141,6 +141,36 @@ export async function listResource(
   return out;
 }
 
+// Group types (e.g. "company") + group instances. Groups use a separate API
+// shape (group_type_index), so they can't go through listResource.
+export async function listGroupTypes(): Promise<Array<Record<string, unknown>>> {
+  const res = await fetch(`${HOST}/api/projects/${PROJECT_ID}/groups_types/`, { headers: headers() });
+  if (!res.ok) throw new Error(`posthog groups_types ${res.status}: ${(await res.text().catch(() => "")).slice(0, 120)}`);
+  const data = (await res.json()) as unknown;
+  if (Array.isArray(data)) return data as Array<Record<string, unknown>>;
+  const r = (data as { results?: Array<Record<string, unknown>> }).results;
+  return Array.isArray(r) ? r : [];
+}
+
+export async function listGroups(
+  groupTypeIndex: number,
+  opts: { maxPages?: number } = {},
+): Promise<Array<Record<string, unknown>>> {
+  const maxPages = opts.maxPages ?? 50;
+  const out: Array<Record<string, unknown>> = [];
+  let url: string | null = `${HOST}/api/projects/${PROJECT_ID}/groups/?group_type_index=${groupTypeIndex}&limit=100`;
+  let pages = 0;
+  while (url && pages < maxPages) {
+    const res: Response = await fetch(url, { headers: headers() });
+    if (!res.ok) throw new Error(`posthog groups ${res.status}: ${(await res.text().catch(() => "")).slice(0, 120)}`);
+    const data = (await res.json()) as { results?: Array<Record<string, unknown>>; next?: string | null };
+    if (Array.isArray(data.results)) out.push(...data.results);
+    url = data.next ?? null;
+    pages++;
+  }
+  return out;
+}
+
 // ---------- Custom events ----------
 
 export type PHCustomEvent = {

apps/worker/src/schedulers/pipeline.ts

@@ -16,6 +16,7 @@ import { runArchivePosthogEvents } from "../jobs/posthog-event-archive";
 import { runArchiveRecordings } from "../jobs/archive-recordings";
 import { runArchiveIdentity } from "../jobs/posthog-identity-archive";
 import { runArchivePosthogResources } from "../jobs/posthog-resource-archive";
+import { runArchivePosthogRecordings } from "../jobs/posthog-recording-archive";
 import { runArchiveSentry } from "../jobs/sentry-archive";
 
 const QUEUE = "insight-pipeline";
@@ -146,6 +147,15 @@ async function runJob(job: Job) {
       }
       return res;
     }
+    case "posthog-recording-archive": {
+      const res = await runArchivePosthogRecordings();
+      if (res.inserted > 0 || !res.ok) {
+        console.log(
+          `[pipeline] recording-archive fetched=${res.fetched} inserted=${res.inserted} pages=${res.pages}${res.reason ? ` reason=${res.reason}` : ""}`,
+        );
+      }
+      return res;
+    }
     case "sentry-archive": {
       const res = await runArchiveSentry();
       if (res.eventsInserted > 0 || res.issuesSynced > 0 || res.dumped > 0 || res.error) {
@@ -239,6 +249,11 @@ export async function startInsightPipeline() {
     { pattern: "15 */6 * * *" },
     { name: "posthog-resource-archive", data: {}, opts: { removeOnComplete: 30, removeOnFail: 15 } },
   );
+  await queue.upsertJobScheduler(
+    "posthog-recording-archive",
+    { pattern: "45 */6 * * *" },
+    { name: "posthog-recording-archive", data: {}, opts: { removeOnComplete: 30, removeOnFail: 15 } },
+  );
   await queue.upsertJobScheduler(
     "sentry-archive",
     { pattern: "0 * * * *" },
@@ -252,7 +267,7 @@ 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, catalog-gap-detect@*/6h, sase-evening-brief@17:00UTC, posthog-event-archive@*/15min, archive-recordings@*/6h, posthog-identity-archive@03:00, posthog-resource-archive@*/6h, sentry-archive@hourly",
+    "[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, catalog-gap-detect@*/6h, sase-evening-brief@17:00UTC, posthog-event-archive@*/15min, archive-recordings@*/6h, posthog-identity-archive@03:00, posthog-resource-archive@*/6h, posthog-recording-archive@*/6h, sentry-archive@hourly",
   );
   return worker;
 }