sp/apps/worker/src/jobs/archive-recordings.ts

import { gzipSync } from "node:zlib";
import { prisma } from "../db";
import { getSnapshotSources, getSnapshotBlob, isConfigured } from "../lib/posthog";
import { putBuffer } from "../lib/minio";

// Archive every recording's raw rrweb before PostHog's ~30-day deletion. Runs over
// ALL recordings (incl. discarded ones) at full fidelity — unlike compress, which
// only handles scored sessions and caps blobs for token budget.
const RRWEB_BUCKET = process.env.RRWEB_ARCHIVE_BUCKET ?? "rrweb-archive";
const BATCH = Number(process.env.INSIGHT_RECORDING_ARCHIVE_BATCH ?? "60");
const MAX_AGE_DAYS = Number(process.env.INSIGHT_RECORDING_ARCHIVE_MAX_AGE_DAYS ?? "25");
// PostHog's snapshot API is aggressively rate-limited. Pace requests and bail out
// of the run on sustained 429s — unarchived rows keep rrwebArchivedAt=null and are
// retried next cycle (the 25d window vs */6h cadence gives ample slack).
const RATE_DELAY_MS = Number(process.env.INSIGHT_RECORDING_ARCHIVE_DELAY_MS ?? "500");
// blob_v2 rejects wide blob-key ranges (a 23-key span 400s, a 5-key span is fine),
// so fetch in small contiguous chunks and concatenate.
const BLOB_CHUNK = Number(process.env.INSIGHT_RECORDING_BLOB_CHUNK ?? "10");
const CHUNK_DELAY_MS = Number(process.env.INSIGHT_RECORDING_CHUNK_DELAY_MS ?? "150");

export type RecordingArchiveResult = {
  scanned: number;
  archived: number;
  failed: number;
  skipped: number;
  rateLimited: boolean;
};

const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
const isRateLimit = (e: unknown) => (e as Error)?.message?.includes(" 429");

export async function runArchiveRecordings(): Promise<RecordingArchiveResult> {
  if (!isConfigured()) return { scanned: 0, archived: 0, failed: 0, skipped: 0, rateLimited: false };

  const floor = new Date(Date.now() - MAX_AGE_DAYS * 24 * 3600_000);
  const pending = await prisma.sessionMeta.findMany({
    where: { rrwebArchivedAt: null, startedAt: { gt: floor } },
    orderBy: { startedAt: "asc" },
    take: BATCH,
  });
  if (pending.length === 0) return { scanned: 0, archived: 0, failed: 0, skipped: 0, rateLimited: false };

  let archived = 0;
  let failed = 0;
  let skipped = 0;
  let rateLimited = false;
  let processed = 0;

  for (const s of pending) {
    if (processed > 0) await sleep(RATE_DELAY_MS);
    processed++;
    try {
      const sources = await getSnapshotSources(s.id);
      if (sources.length === 0) {
        // No snapshots (too short / already purged) — mark done so we stop retrying.
        await prisma.sessionMeta.update({ where: { id: s.id }, data: { rrwebArchivedAt: new Date() } });
        skipped++;
        continue;
      }

      // Fetch ALL blobs per source (no 12-cap), chunked to dodge blob_v2's range limit.
      const bySource = new Map<string, string[]>();
      for (const src of sources) {
        if (!bySource.has(src.source)) bySource.set(src.source, []);
        bySource.get(src.source)!.push(src.blob_key);
      }
      const parts: string[] = [];
      for (const [source, keys] of bySource) {
        const sorted = [...keys].sort((a, b) => Number(a) - Number(b));
        for (let i = 0; i < sorted.length; i += BLOB_CHUNK) {
          if (i > 0) await sleep(CHUNK_DELAY_MS);
          const lo = sorted[i];
          const hi = sorted[Math.min(i + BLOB_CHUNK - 1, sorted.length - 1)];
          const blob = await getSnapshotBlob(s.id, source, lo, hi);
          if (blob) parts.push(blob);
        }
      }
      const raw = parts.join("\n").trim();
      if (!raw) {
        await prisma.sessionMeta.update({ where: { id: s.id }, data: { rrwebArchivedAt: new Date() } });
        skipped++;
        continue;
      }

      const key = `${s.projectKey}/${dateFolder(s.startedAt)}/${s.id}.jsonl.gz`;
      const gz = gzipSync(Buffer.from(raw, "utf-8"));
      await putBuffer(RRWEB_BUCKET, key, gz, "application/gzip");

      await prisma.sessionMeta.update({
        where: { id: s.id },
        data: { rrwebArchivedAt: new Date(), rrwebArchiveKey: key },
      });
      archived++;
    } catch (e) {
      if (isRateLimit(e)) {
        // Rate limited — stop hammering. Remaining rows retry next cycle.
        rateLimited = true;
        break;
      }
      // Other transient errors — leave rrwebArchivedAt null so it retries next run.
      console.warn("[archive-recordings] failed", s.id, (e as Error).message);
      failed++;
    }
  }

  return { scanned: pending.length, archived, failed, skipped, rateLimited };
}

function dateFolder(d: Date): string {
  const y = d.getUTCFullYear();
  const m = String(d.getUTCMonth() + 1).padStart(2, "0");
  const day = String(d.getUTCDate()).padStart(2, "0");
  return `${y}/${m}/${day}`;
}