feat(insights): Phase B — PostHog person + cohort snapshots

Append-only history of person properties and cohort definitions, since
the free tier overwrites the live person / prunes data. A new row is
written only when the payload changes (hash compare), so the tables stay
a compact change-timeline rather than a daily full copy.

- `PosthogPersonSnapshot` (personId join key to PosthogEvent.personId,
  all distinct_ids, properties JSONB, propertiesHash).
- `PosthogCohortSnapshot` (cohortId, name, count, filters, stateHash).
- `listPersons` / `listCohorts` REST helpers (next-pagination) in posthog.ts.
- `stableHash` (sorted-key JSON hash) in hash.ts for change detection.
- `posthog-identity-archive` job wired @03:00 daily. Latest-hash lookup
  via one DISTINCT ON query, inserts only changed rows via createMany.

Verified against prod: 242 persons → 242 snapshots on first run, 0 on
immediate re-run (change detection). 0 cohorts currently → graceful no-op.

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

apps/web/prisma/schema.prisma

@@ -473,3 +473,38 @@ model PosthogEvent {
   @@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")
+}

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 };
+}

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);
+}

apps/worker/src/lib/posthog.ts

@@ -177,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,

apps/worker/src/schedulers/pipeline.ts

@@ -12,6 +12,7 @@ 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";
 
 const QUEUE = "insight-pipeline";
 
@@ -107,6 +108,15 @@ async function runJob(job: Job) {
       }
       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;
+    }
     default:
       return { ok: false, error: `unknown job ${job.name}` };
   }
@@ -168,6 +178,11 @@ export async function startInsightPipeline() {
     { 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 } },
+  );
 
   new Worker(QUEUE, runJob, {
     connection: redis,
@@ -176,6 +191,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, posthog-event-archive@*/15min, archive-recordings@*/6h",
+    "[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",
   );
 }