From ab86dbdf5987892996569616412277a07e46724b Mon Sep 17 00:00:00 2001 From: Semih Date: Wed, 27 May 2026 20:47:20 +0300 Subject: [PATCH] =?UTF-8?q?feat(insights):=20Phase=20B=20=E2=80=94=20PostH?= =?UTF-8?q?og=20person=20+=20cohort=20snapshots?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- apps/web/prisma/schema.prisma | 35 +++++++ .../src/jobs/posthog-identity-archive.ts | 97 +++++++++++++++++++ apps/worker/src/lib/hash.ts | 14 +++ apps/worker/src/lib/posthog.ts | 49 ++++++++++ apps/worker/src/schedulers/pipeline.ts | 17 +++- 5 files changed, 211 insertions(+), 1 deletion(-) create mode 100644 apps/worker/src/jobs/posthog-identity-archive.ts diff --git a/apps/web/prisma/schema.prisma b/apps/web/prisma/schema.prisma index d84b1da..5b28c98 100644 --- a/apps/web/prisma/schema.prisma +++ b/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") +} diff --git a/apps/worker/src/jobs/posthog-identity-archive.ts b/apps/worker/src/jobs/posthog-identity-archive.ts new file mode 100644 index 0000000..8632419 --- /dev/null +++ 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 { + 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>` + 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>` + 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 }; +} diff --git a/apps/worker/src/lib/hash.ts b/apps/worker/src/lib/hash.ts index 02dd887..c7a0f63 100644 --- a/apps/worker/src/lib/hash.ts +++ b/apps/worker/src/lib/hash.ts @@ -14,3 +14,17 @@ export function fingerprintHash(parts: Array .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; + 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); +} diff --git a/apps/worker/src/lib/posthog.ts b/apps/worker/src/lib/posthog.ts index b06a2a8..fe180cf 100644 --- a/apps/worker/src/lib/posthog.ts +++ b/apps/worker/src/lib/posthog.ts @@ -177,6 +177,55 @@ export async function getPerson(distinctId: string): Promise { 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; +}; + +// Lists all persons via the REST endpoint, following `next` pagination. +export async function listPersons(opts?: { limit?: number; maxPages?: number }): Promise { + 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 { + 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, diff --git a/apps/worker/src/schedulers/pipeline.ts b/apps/worker/src/schedulers/pipeline.ts index 2f808ca..f1e388f 100644 --- a/apps/worker/src/schedulers/pipeline.ts +++ b/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", ); }