feat(archive): capture all PostHog resource modules to DB + MinIO

Extends archival beyond events/persons/cohorts. New generic resource
archiver (posthog-resource-archive@*/6h) snapshots surveys, feature flags,
experiments, dashboards, insights, annotations, and actions into
posthog_resource_snapshots — change-detected (hash over non-volatile config),
lossless `data` payload — and dumps a full per-type daily copy to MinIO
(resources/<type>/<date>.json.gz). listResource() pages any PostHog REST
resource. Survey responses + flag calls are already captured as events.

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

apps/web/prisma/schema.prisma

@@ -538,6 +538,29 @@ model PosthogCohortSnapshot {
   @@map("posthog_cohort_snapshots")
 }
 
+// Generic change-history of every PostHog "resource" module beyond events:
+// surveys, feature_flags, experiments, dashboards, insights, annotations,
+// actions. One row is appended only when a resource's config changed
+// (stateHash compare); the full payload is kept lossless in `data`, and a full
+// per-type daily snapshot is also dumped to MinIO. Survey responses + flag
+// calls are events (already in posthog_events) — this captures the definitions.
+model PosthogResourceSnapshot {
+  id           String    @id @default(cuid())
+  projectKey   String
+  resourceType String // survey | feature_flag | experiment | dashboard | insight | annotation | action
+  resourceId   String // PostHog resource id (stringified)
+  name         String?
+  stateHash    String // change-detection over the non-volatile config
+  data         Json // full resource payload, lossless
+  capturedAt   DateTime  @default(now())
+  dumpedAt     DateTime?
+
+  @@index([projectKey, resourceType, capturedAt])
+  @@index([resourceType, resourceId, capturedAt])
+  @@index([dumpedAt])
+  @@map("posthog_resource_snapshots")
+}
+
 // ---------- Phase C: Sentry archive ----------
 
 // Sentry issue (group) aggregate state. Upserted to the latest snapshot — the

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

@@ -0,0 +1,91 @@
+import { gzipSync } from "node:zlib";
+import { prisma } from "../db";
+import { listResource, isConfigured } from "../lib/posthog";
+import { stableHash } from "../lib/hash";
+import { putBuffer } from "../lib/minio";
+
+const PROJECT_KEY = process.env.INSIGHT_PROJECT_KEY ?? "sase";
+const ARCHIVE_BUCKET = process.env.INSIGHT_ARCHIVE_BUCKET ?? "posthog-archive";
+
+// Top-level fields PostHog mutates on every fetch (server timestamps / cached
+// results) — excluded from the change-detection hash so we only append a
+// snapshot on a real config change. The full payload is still stored losslessly.
+const VOLATILE = new Set([
+  "updated_at", "last_modified_at", "last_refresh", "last_used_at", "last_modified_by",
+  "is_cached", "result", "results", "next_allowed_client_refresh", "last_viewed_at",
+  "filters_hash", "effective_restriction_level", "effective_privilege_level", "_create_in_folder",
+]);
+
+// Every PostHog resource module beyond the event stream. Survey responses and
+// flag-call events are already captured as events (posthog_events); this archives
+// the definitions/config + their daily full state.
+const RESOURCES: Array<{ type: string; path: string; nameField?: string }> = [
+  { type: "survey", path: "surveys", nameField: "name" },
+  { type: "feature_flag", path: "feature_flags", nameField: "key" },
+  { type: "experiment", path: "experiments", nameField: "name" },
+  { type: "dashboard", path: "dashboards", nameField: "name" },
+  { type: "annotation", path: "annotations", nameField: "content" },
+  { type: "insight", path: "insights", nameField: "name" },
+  { type: "action", path: "actions", nameField: "name" },
+];
+
+function hashConfig(item: Record<string, unknown>): string {
+  const clean: Record<string, unknown> = {};
+  for (const [k, v] of Object.entries(item)) if (!VOLATILE.has(k)) clean[k] = v;
+  return stableHash(clean);
+}
+
+export type ResourceArchiveResult = {
+  ok: boolean;
+  perType: Record<string, { scanned: number; snapshots: number; dumped: boolean; error?: string }>;
+  reason?: string;
+};
+
+export async function runArchivePosthogResources(): Promise<ResourceArchiveResult> {
+  if (!isConfigured()) return { ok: false, perType: {}, reason: "posthog_not_configured" };
+  const perType: ResourceArchiveResult["perType"] = {};
+  const day = new Date().toISOString().slice(0, 10);
+
+  for (const r of RESOURCES) {
+    const stat = { scanned: 0, snapshots: 0, dumped: false } as ResourceArchiveResult["perType"][string];
+    perType[r.type] = stat;
+    try {
+      const items = await listResource(r.path);
+      stat.scanned = items.length;
+
+      // Latest stored hash per resourceId (one query via DISTINCT ON).
+      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" = ${r.type}
+        ORDER BY "resourceId", "capturedAt" DESC
+      `;
+      const latestHash = new Map(latest.map((x) => [x.resourceId, x.stateHash]));
+
+      const rows = items
+        .map((it) => ({
+          projectKey: PROJECT_KEY,
+          resourceType: r.type,
+          resourceId: String(it.id ?? it.short_id ?? ""),
+          name: r.nameField ? String(it[r.nameField] ?? "").slice(0, 300) : null,
+          stateHash: hashConfig(it),
+          data: it as object,
+        }))
+        .filter((row) => row.resourceId && latestHash.get(row.resourceId) !== row.stateHash);
+
+      if (rows.length) {
+        const res = await prisma.posthogResourceSnapshot.createMany({ data: rows as never });
+        stat.snapshots = res.count;
+      }
+
+      // Full lossless dump to MinIO — one object per type per day (idempotent).
+      const gz = gzipSync(Buffer.from(JSON.stringify(items)));
+      await putBuffer(ARCHIVE_BUCKET, `resources/${r.type}/${day}.json.gz`, gz, "application/gzip");
+      stat.dumped = true;
+    } catch (e) {
+      stat.error = (e as Error).message;
+    }
+  }
+
+  return { ok: true, perType };
+}

apps/worker/src/lib/posthog.ts

@@ -117,6 +117,30 @@ export async function hogqlQuery(query: string): Promise<HogQLResponse> {
   return { columns: data.columns ?? [], results: data.results ?? [] };
 }
 
+// Generic paginated reader for PostHog REST resources (surveys, feature_flags,
+// experiments, dashboards, insights, annotations, actions). Follows `next`.
+export async function listResource(
+  path: string,
+  opts: { maxPages?: number; pageSize?: number } = {},
+): Promise<Array<Record<string, unknown>>> {
+  const maxPages = opts.maxPages ?? 50;
+  const pageSize = opts.pageSize ?? 100;
+  const out: Array<Record<string, unknown>> = [];
+  let url: string | null = `${HOST}/api/projects/${PROJECT_ID}/${path}/?limit=${pageSize}`;
+  let pages = 0;
+  while (url && pages < maxPages) {
+    const res: Response = await fetch(url, { headers: headers() });
+    if (!res.ok) {
+      throw new Error(`posthog ${path} ${res.status}: ${(await res.text().catch(() => "")).slice(0, 160)}`);
+    }
+    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

@@ -15,6 +15,7 @@ import { runSaseEveningBrief } from "../jobs/sase-evening-brief";
 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 { runArchiveSentry } from "../jobs/sentry-archive";
 
 const QUEUE = "insight-pipeline";
@@ -134,6 +135,17 @@ async function runJob(job: Job) {
       }
       return res;
     }
+    case "posthog-resource-archive": {
+      const res = await runArchivePosthogResources();
+      const snaps = Object.values(res.perType).reduce((a, t) => a + t.snapshots, 0);
+      const errs = Object.entries(res.perType).filter(([, t]) => t.error).map(([k]) => k);
+      if (snaps > 0 || errs.length || !res.ok) {
+        console.log(
+          `[pipeline] resource-archive types=${Object.keys(res.perType).length} snapshots=${snaps}${errs.length ? ` errors=[${errs.join(",")}]` : ""}${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) {
@@ -222,6 +234,11 @@ export async function startInsightPipeline() {
     { pattern: "0 3 * * *" },
     { name: "posthog-identity-archive", data: {}, opts: { removeOnComplete: 30, removeOnFail: 15 } },
   );
+  await queue.upsertJobScheduler(
+    "posthog-resource-archive",
+    { pattern: "15 */6 * * *" },
+    { name: "posthog-resource-archive", data: {}, opts: { removeOnComplete: 30, removeOnFail: 15 } },
+  );
   await queue.upsertJobScheduler(
     "sentry-archive",
     { pattern: "0 * * * *" },
@@ -235,7 +252,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, 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, sentry-archive@hourly",
   );
   return worker;
 }