feat(db): per-day status/audit for the replay archive (Postgres, no deletion) (#423)

* feat(db): per-day status/audit for the replay archive (Postgres, no deletion)

Track the session-replay CH->Azure-Blob archive per day so we can see what's
archived/verified/failed and later verify by hand before any deletion. The job
records; it never drops a CH partition.

- Prisma model ReplayArchiveDay (@@Map replay_archive_days) + migration. One row
  per day (= the CH partition unit toYYYYMMDD(started_at) and the verifyDay
  unit). deletedAt column reserved for a future delete phase; never written here.
- replay-archive-day.service.ts: reconcileArchiveDay / markArchiving /
  markVerified / markVerifyFailed / listArchiveDays.
- archive-replay-chunks.ts wiring:
  * reconcileAllStatus() at start upserts EVERY settled day from counts (so days
    archived before this existed are recorded too).
  * per processed day: archiving -> archived+verifiedAt (verify passed) /
    verify_failed+reason.
  * all status writes are best-effort (status() wrapper): a Postgres hiccup or a
    pod without DATABASE_URL logs a WARN and NEVER fails an archive run.
- status:replay CLI prints the table (day, status, ch/blob, drift, verifiedAt,
  error) + a summary. Read-only.

No DROP PARTITION, no DELETE_AFTER_DAYS, no delete gate — observability only.
Rollout: run the migration first, then deploy; add DATABASE_URL to the archive
cron/backfill pods (openpanel repo manifests) so they can write.

* fix(db): address CodeRabbit — export-throw + unreadable-count status writes

- archiveProject signals failure by THROWING (never returns false), so the old
  `if (!archiveProject)` branch was dead code and an export failure left the day
  stuck at status='archiving'. Wrap the project loop in try/catch: record the
  failure (without clobbering counts) and return false so the day retries and
  the other days still run.
- verifyDay: reject unreadable/sentinel counts (idxN=-1, srcN=-2) before the
  completeness check — otherwise `idxN < srcN` was false and a day with no
  readable counts was marked archived with negative counts.
- markVerifyFailed: counts are now optional so the export-throw path doesn't
  overwrite a real blob count with 0.

Author: ayushjhanwar-png

packages/db/package.json

@@ -11,6 +11,7 @@
     "migrate:deploy": "pnpm migrate:deploy:db && pnpm migrate:deploy:code",
     "materialize:analyze": "pnpm with-env jiti ./src/cli/materialize.ts",
     "archive:replay": "pnpm with-env jiti ./src/cli/archive-replay-chunks.ts",
+    "status:replay": "pnpm with-env jiti ./src/cli/replay-archive-status.ts",
     "duplicate-events": "pnpm with-env jiti ./scripts/find-duplicate-events.ts",
     "typecheck": "tsc --noEmit",
     "with-env": "dotenv -e ../../.env -c --"

packages/db/prisma/migrations/20260818120000_add_replay_archive_days/migration.sql

@@ -0,0 +1,18 @@
+-- CreateTable
+CREATE TABLE "replay_archive_days" (
+    "day" DATE NOT NULL,
+    "status" TEXT NOT NULL,
+    "chChunks" BIGINT,
+    "blobChunks" BIGINT,
+    "driftChunks" INTEGER,
+    "sessions" INTEGER,
+    "archivedAt" TIMESTAMP(3),
+    "verifiedAt" TIMESTAMP(3),
+    "verifyError" TEXT,
+    "lastRunAt" TIMESTAMP(3) NOT NULL,
+    "deletedAt" TIMESTAMP(3),
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updatedAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+
+    CONSTRAINT "replay_archive_days_pkey" PRIMARY KEY ("day")
+);

packages/db/prisma/schema.prisma

@@ -569,3 +569,29 @@ model Cohort {
   @@map("cohorts")
   @@index([projectId])
 }
+
+/// Per-day lifecycle record for the session-replay CH→Azure-Blob archive.
+/// Populated by src/cli/archive-replay-chunks.ts; observational + audit only —
+/// the archive job never drops CH partitions. Deletion (when built) will
+/// re-verify freshly at delete time and set deletedAt; the job does not.
+/// Granularity is per-day because a CH partition is toYYYYMMDD(started_at)
+/// (one partition per day, all projects) — the same unit as verifyDay and the
+/// eventual DROP PARTITION.
+model ReplayArchiveDay {
+  day         DateTime  @id @db.Date // the archived day (UTC)
+  /// pending | archiving | archived | verify_failed
+  status      String
+  chChunks    BigInt? // CH source chunk count at last check
+  blobChunks  BigInt? // archive-index chunk count at last check
+  driftChunks Int? // blobChunks - chChunks (benign blob-superset, if > 0)
+  sessions    Int? // archived session count for the day
+  archivedAt  DateTime? // first time the day was count-complete (index >= ch)
+  verifiedAt  DateTime? // last time the full verify passed (count + blob sample)
+  verifyError String? // reason when status = verify_failed
+  lastRunAt   DateTime // last time the archive job touched this day
+  deletedAt   DateTime? // reserved for the future delete phase; job never sets it
+  createdAt   DateTime  @default(now())
+  updatedAt   DateTime  @default(now()) @updatedAt
+
+  @@map("replay_archive_days")
+}

packages/db/src/cli/archive-replay-chunks.ts

@@ -32,6 +32,12 @@
  */
 import type { ClickHouseSettings } from '@clickhouse/client';
 import { ch, chQuery } from '../clickhouse/client';
+import {
+  markArchiving,
+  markVerified,
+  markVerifyFailed,
+  reconcileArchiveDay,
+} from '../services/replay-archive-day.service';
 
 const CONTAINER = process.env.REPLAY_ARCHIVE_CONTAINER || 'clickhouse-export';
 const CONN = process.env.AZURE_BLOB_CONNECTION_STRING || '';
@@ -415,7 +421,9 @@ async function archiveProject(day: DayPlan, p: ProjectPlan): Promise<boolean> {
   return true;
 }
 
-async function verifyDay(day: DayPlan): Promise<boolean> {
+type VerifyResult = { ok: boolean; idxN: number; srcN: number };
+
+async function verifyDay(day: DayPlan): Promise<VerifyResult> {
   // Gate: everything we indexed for this day must equal the source. Both count
   // physical day-N chunks (populateSliceIndex filters started_at to the day), so
   // they match even for midnight-crossing sessions. This is a cheap aggregate —
@@ -431,6 +439,13 @@ async function verifyDay(day: DayPlan): Promise<boolean> {
   );
   const idxN = Number(idx?.n ?? -1);
   const srcN = Number(src?.n ?? -2);
+  // Unreadable counts (missing row / non-numeric) must NOT slip through as
+  // verified — the sentinels (-1, -2) would otherwise satisfy `idxN < srcN ===
+  // false` and mark the day archived with negative counts. Treat as failed.
+  if (!Number.isFinite(idxN) || !Number.isFinite(srcN) || idxN < 0 || srcN < 0) {
+    log(`  verify day: unreadable counts index=${idxN} ch=${srcN} — failing`);
+    return { ok: false, idxN, srcN };
+  }
   // Completeness is one-directional: the archive must hold AT LEAST every source
   // chunk. index < ch => some chunks were never written to a blob => INCOMPLETE,
   // fail (loud, blocks deletion of this day). index >= ch => nothing missing;
@@ -442,7 +457,7 @@ async function verifyDay(day: DayPlan): Promise<boolean> {
     log(
       `  verify day: index=${idxN} ch=${srcN} INCOMPLETE — ${srcN - idxN} chunk(s) not archived`,
     );
-    return false;
+    return { ok: false, idxN, srcN };
   }
   // Independent spot-check: actually read a few of the smallest per-session
   // blobs back and confirm their row counts. Guards against an indexed-but-
@@ -457,13 +472,26 @@ async function verifyDay(day: DayPlan): Promise<boolean> {
     const got = await countBlob(s.blob_path);
     if (got !== Number(s.chunks)) {
       log(`  verify day: sample ${s.blob_path} blob=${got} idx=${s.chunks} MISMATCH`);
-      return false;
+      return { ok: false, idxN, srcN };
     }
   }
   log(
     `  verify day: index=${idxN} ch=${srcN} OK${idxN > srcN ? ` (+${idxN - srcN} blob-superset drift)` : ''} (+${samples.length} blob samples)`,
   );
-  return true;
+  return { ok: true, idxN, srcN };
+}
+
+/**
+ * Best-effort status write. Postgres status is observability, not the archive's
+ * job — a DB hiccup (or a pod without DATABASE_URL) must NEVER fail an archive
+ * run, so every write is swallowed here with a loud log.
+ */
+async function status(fn: () => Promise<void>): Promise<void> {
+  try {
+    await fn();
+  } catch (err) {
+    log(`  WARN: status write failed (non-fatal): ${String(err)}`);
+  }
 }
 
 async function archiveDay(day: DayPlan): Promise<boolean> {
@@ -479,23 +507,79 @@ async function archiveDay(day: DayPlan): Promise<boolean> {
     }
     return true;
   }
-  for (const p of projects) {
-    if (!(await archiveProject(day, p))) return false;
+  await status(() => markArchiving(day.date));
+  // archiveProject signals failure by THROWING (exportSlice / populateSliceIndex),
+  // never by returning false — so catch it here: record the failure (without
+  // clobbering counts) and return false so this day is retried and the OTHER
+  // days still run, instead of the throw aborting the whole batch.
+  try {
+    for (const p of projects) {
+      await archiveProject(day, p);
+    }
+  } catch (err) {
+    log(`  ${day.date} export failed: ${String(err)}`);
+    await status(() =>
+      markVerifyFailed(day.date, `export failed: ${String(err)}`),
+    );
+    return false;
   }
-  if (!(await verifyDay(day))) {
+  const v = await verifyDay(day);
+  if (!v.ok) {
     log(`  STOP: ${day.date} count mismatch — leaving unarchived for retry`);
+    await status(() =>
+      markVerifyFailed(day.date, `verify failed: index=${v.idxN} ch=${v.srcN}`, {
+        chChunks: v.srcN,
+        blobChunks: v.idxN,
+      }),
+    );
     return false;
   }
+  const sessions = projects.reduce((n, p) => n + p.sessions, 0);
+  await status(() => markVerified(day.date, v.srcN, v.idxN, sessions));
   log(`  ${day.date} done`);
   return true;
 }
 
+/**
+ * Upsert a status row for EVERY settled day from counts alone, so the table is
+ * complete — including days already fully archived (which the archiver skips and
+ * would otherwise never record). Count-only: `blob >= ch` → archived, else
+ * pending; verifiedAt is left for the per-day verify path. Best-effort.
+ */
+async function reconcileAllStatus(): Promise<void> {
+  const cutoff = new Date();
+  cutoff.setUTCDate(cutoff.getUTCDate() - SETTLE_DAYS);
+  const cutoffStr = cutoff.toISOString().slice(0, 10);
+  const parts = await chQuery<{ partition: string; rows: string }>(
+    `SELECT partition, sum(rows) AS rows
+       FROM system.parts
+      WHERE table = '${TABLE}' AND database = currentDatabase() AND active
+      GROUP BY partition`,
+    LIGHT_SETTINGS,
+  );
+  const indexed = await chQuery<{ dt: string; chunks: string }>(
+    `SELECT toString(dt) AS dt, sum(chunks) AS chunks FROM ${INDEX} FINAL GROUP BY dt`,
+    LIGHT_SETTINGS,
+  );
+  const idxMap = new Map(indexed.map((r) => [r.dt, Number(r.chunks)]));
+  for (const p of parts) {
+    const date = partitionToDate(p.partition);
+    if (!date || date < MIN_DAY || date > cutoffStr) continue;
+    const rows = Number(p.rows);
+    if (rows === 0) continue;
+    await reconcileArchiveDay(date, rows, idxMap.get(date) ?? 0);
+  }
+}
+
 async function main(): Promise<number> {
   if (!CONN) throw new Error('AZURE_BLOB_CONNECTION_STRING is required');
   log(
     `start container=${CONTAINER} settleDays=${SETTLE_DAYS} sessionsPerBatch=${TARGET_SESSIONS_PER_BATCH} maxDays=${MAX_DAYS_PER_RUN} blockSize=${MAX_BLOCK_SIZE} rearchive=${REARCHIVE} dryRun=${DRY_RUN}`,
   );
 
+  // Refresh the per-day status table for all settled days (observability only).
+  await status(() => reconcileAllStatus());
+
   const plans = await planDays();
   if (plans.length === 0) {
     log('nothing to archive — all settled days already indexed');

packages/db/src/cli/replay-archive-status.ts

@@ -0,0 +1,62 @@
+/**
+ * Print the session-replay archive status table (Postgres `replay_archive_days`).
+ * Read-only — no ClickHouse, no Blob, nothing dropped.
+ *
+ *   pnpm --filter @openpanel/db status:replay
+ */
+import { listArchiveDays } from '../services/replay-archive-day.service';
+
+const ts = (d: Date | null): string =>
+  d ? new Date(d).toISOString().slice(0, 16).replace('T', ' ') : '—';
+const ymd = (d: Date): string => new Date(d).toISOString().slice(0, 10);
+
+async function main(): Promise<void> {
+  const days = await listArchiveDays();
+  if (days.length === 0) {
+    // eslint-disable-next-line no-console
+    console.log('replay_archive_days is empty — run archive:replay first.');
+    return;
+  }
+
+  const cols = [
+    'day'.padEnd(10),
+    'status'.padEnd(13),
+    'ch'.padStart(10),
+    'blob'.padStart(10),
+    'drift'.padStart(6),
+    'sessions'.padStart(8),
+    'verifiedAt'.padEnd(16),
+    'error',
+  ].join('  ');
+  // eslint-disable-next-line no-console
+  console.log(cols);
+  for (const r of days) {
+    // eslint-disable-next-line no-console
+    console.log(
+      [
+        ymd(r.day).padEnd(10),
+        r.status.padEnd(13),
+        (r.chChunks?.toString() ?? '—').padStart(10),
+        (r.blobChunks?.toString() ?? '—').padStart(10),
+        String(r.driftChunks ?? 0).padStart(6),
+        String(r.sessions ?? '—').padStart(8),
+        ts(r.verifiedAt).padEnd(16),
+        r.verifyError ?? '',
+      ].join('  '),
+    );
+  }
+
+  const by = (s: string): number => days.filter((d) => d.status === s).length;
+  // eslint-disable-next-line no-console
+  console.log(
+    `\n${days.length} day(s): archived=${by('archived')} pending=${by('pending')} archiving=${by('archiving')} verify_failed=${by('verify_failed')}`,
+  );
+}
+
+main()
+  .then(() => process.exit(0))
+  .catch((err) => {
+    // eslint-disable-next-line no-console
+    console.error('[status:replay]', err);
+    process.exit(1);
+  });