feat(insights): Phase C — Sentry issues + events archive

Mirrors Sentry into our DB before the free tier prunes events (~30d).

- `SentryIssue` (aggregate state, upserted to latest) + `SentryEvent`
  (raw occurrences, lossless full payload JSONB, dedup by eventId, cold
  dump to MinIO `sentry-archive/{project}/YYYY/MM/DD.jsonl.gz`).
- `lib/sentry.ts`: read-only client, Link-header cursor pagination,
  listIssuesPage / listEventsPage. EU-region aware (SENTRY_API_BASE).
- `sentry-archive` job @hourly: issues upsert + events newest-first with
  skipDuplicates, stops once a page is all-duplicates (caught up).
- Config via env: SENTRY_AUTH_TOKEN / SENTRY_ORG / SENTRY_PROJECT /
  SENTRY_API_BASE.

Verified live against otolog/python (EU): a test event archived on run 1,
0 inserts / 1 duplicate on run 2 (dedup), issue upserted idempotently.

Note: Sase API currently sends to an inaccessible org's DSN (hardcoded
fallback); repointing SENTRY_DSN to otolog/python is a separate deploy step.

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

apps/web/prisma/schema.prisma

@@ -508,3 +508,51 @@ model PosthogCohortSnapshot {
   @@index([projectKey, capturedAt])
   @@map("posthog_cohort_snapshots")
 }
+
+// ---------- Phase C: Sentry archive ----------
+
+// Sentry issue (group) aggregate state. Upserted to the latest snapshot — the
+// raw history lives in SentryEvent; the issue row is the current grouping/metadata.
+model SentryIssue {
+  id           String   @id          // Sentry issue/group id
+  projectKey   String                // "sase"
+  shortId      String?
+  title        String?
+  culprit      String?
+  level        String?
+  status       String?
+  type         String?
+  count        Int?                  // total events as of lastSyncedAt
+  userCount    Int?
+  firstSeen    DateTime?
+  lastSeen     DateTime?
+  permalink    String?
+  metadata     Json?
+  ingestedAt   DateTime @default(now())
+  lastSyncedAt DateTime @default(now())
+
+  @@index([projectKey, lastSeen])
+  @@map("sentry_issues")
+}
+
+// Individual Sentry events (the at-risk raw occurrences; free tier prunes ~30d).
+// Lossless full payload as JSONB; also cold-dumped to MinIO.
+model SentryEvent {
+  eventId     String   @id          // Sentry event id (32 hex)
+  issueId     String?               // parent issue/group id
+  projectKey  String
+  title       String?
+  message     String?
+  level       String?
+  platform    String?
+  timestamp   DateTime               // event occurrence time
+  tags        Json?
+  payload     Json                   // full event payload — lossless
+  ingestedAt  DateTime @default(now())
+  dumpedAt    DateTime?
+
+  @@index([projectKey, timestamp])
+  @@index([issueId, timestamp])
+  @@index([dumpedAt])
+  @@map("sentry_events")
+}

apps/worker/src/jobs/sentry-archive.ts

@@ -0,0 +1,173 @@
+import { gzipSync } from "node:zlib";
+import { prisma } from "../db";
+import { isConfigured, listIssuesPage, listEventsPage, type SentryEventRaw } from "../lib/sentry";
+import { putBuffer } from "../lib/minio";
+
+const PROJECT_KEY = process.env.INSIGHT_PROJECT_KEY ?? "sase";
+const ISSUE_MAX_PAGES = Number(process.env.SENTRY_ISSUE_MAX_PAGES ?? "20");
+const EVENT_MAX_PAGES = Number(process.env.SENTRY_EVENT_MAX_PAGES ?? "40");
+const MAX_DUMP_DAYS = Number(process.env.SENTRY_MAX_DUMP_DAYS ?? "120");
+const ARCHIVE_BUCKET = process.env.SENTRY_ARCHIVE_BUCKET ?? "sentry-archive";
+
+export type SentryArchiveResult = {
+  issuesSynced: number;
+  eventsFetched: number;
+  eventsInserted: number;
+  eventsDuplicate: number;
+  dumped: number;
+  error?: string;
+};
+
+function toDate(v: unknown): Date | null {
+  if (!v) return null;
+  const d = new Date(String(v));
+  return Number.isNaN(d.getTime()) ? null : d;
+}
+
+export async function runArchiveSentry(): Promise<SentryArchiveResult> {
+  if (!isConfigured()) {
+    return { issuesSynced: 0, eventsFetched: 0, eventsInserted: 0, eventsDuplicate: 0, dumped: 0, error: "sentry_not_configured" };
+  }
+
+  let issuesSynced = 0;
+  let eventsFetched = 0;
+  let eventsInserted = 0;
+  let eventsDuplicate = 0;
+  let error: string | undefined;
+
+  // ── Issues: upsert latest aggregate state ──
+  try {
+    let cursor: string | undefined;
+    for (let page = 0; page < ISSUE_MAX_PAGES; page++) {
+      const { data, nextCursor } = await listIssuesPage(cursor);
+      if (!data.length) break;
+      for (const i of data) {
+        if (!i.id) continue;
+        const row = {
+          projectKey: PROJECT_KEY,
+          shortId: i.shortId ?? null,
+          title: i.title ?? null,
+          culprit: i.culprit ?? null,
+          level: i.level ?? null,
+          status: i.status ?? null,
+          type: i.type ?? null,
+          count: i.count != null ? Number(i.count) : null,
+          userCount: i.userCount ?? null,
+          firstSeen: toDate(i.firstSeen),
+          lastSeen: toDate(i.lastSeen),
+          permalink: i.permalink ?? null,
+          metadata: (i.metadata ?? {}) as object,
+          lastSyncedAt: new Date(),
+        };
+        await prisma.sentryIssue.upsert({
+          where: { id: i.id },
+          create: { id: i.id, ...row },
+          update: row,
+        });
+        issuesSynced++;
+      }
+      if (!nextCursor) break;
+      cursor = nextCursor;
+    }
+  } catch (e) {
+    error = `issues_failed: ${(e as Error).message}`;
+  }
+
+  // ── Events: newest-first, dedup by eventId, stop once caught up ──
+  try {
+    let cursor: string | undefined;
+    for (let page = 0; page < EVENT_MAX_PAGES; page++) {
+      const { data, nextCursor } = await listEventsPage(cursor);
+      if (!data.length) break;
+
+      const batch = data
+        .map((e: SentryEventRaw) => {
+          const eventId = String(e.eventID ?? e.id ?? "");
+          const ts = toDate(e.dateCreated ?? e.dateReceived);
+          if (!eventId || !ts) return null;
+          return {
+            eventId,
+            issueId: e.groupID ? String(e.groupID) : null,
+            projectKey: PROJECT_KEY,
+            title: e.title ?? null,
+            message: typeof e.message === "string" ? e.message : null,
+            level: typeof e.level === "string" ? (e.level as string) : null,
+            platform: e.platform ?? null,
+            timestamp: ts,
+            tags: (e.tags ?? []) as object,
+            payload: e as object,
+          };
+        })
+        .filter((r): r is NonNullable<typeof r> => r !== null);
+
+      eventsFetched += batch.length;
+      if (batch.length) {
+        const res = await prisma.sentryEvent.createMany({ data: batch as never, skipDuplicates: true });
+        eventsInserted += res.count;
+        eventsDuplicate += batch.length - res.count;
+        // A full page with no new rows means we've reached already-archived events.
+        if (res.count === 0) break;
+      }
+      if (!nextCursor) break;
+      cursor = nextCursor;
+    }
+  } catch (e) {
+    error = error ?? `events_failed: ${(e as Error).message}`;
+  }
+
+  // ── Cold dump closed days to MinIO ──
+  let dumped = 0;
+  try {
+    dumped = await dumpClosedDays(new Date());
+  } catch (e) {
+    error = error ?? `dump_failed: ${(e as Error).message}`;
+  }
+
+  return { issuesSynced, eventsFetched, eventsInserted, eventsDuplicate, dumped, error };
+}
+
+async function dumpClosedDays(now: Date): Promise<number> {
+  const todayStart = new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), now.getUTCDate()));
+  const undumped = await prisma.sentryEvent.findMany({
+    where: { projectKey: PROJECT_KEY, dumpedAt: null, timestamp: { lt: todayStart } },
+    select: { timestamp: true },
+    orderBy: { timestamp: "asc" },
+  });
+  if (undumped.length === 0) return 0;
+
+  const days = Array.from(new Set(undumped.map((r) => r.timestamp.toISOString().slice(0, 10)))).slice(0, MAX_DUMP_DAYS);
+  let dumped = 0;
+  for (const day of days) {
+    const dayStart = new Date(`${day}T00:00:00.000Z`);
+    const dayEnd = new Date(dayStart.getTime() + 24 * 3600_000);
+    const rows = await prisma.sentryEvent.findMany({
+      where: { projectKey: PROJECT_KEY, timestamp: { gte: dayStart, lt: dayEnd } },
+      orderBy: { timestamp: "asc" },
+    });
+    if (rows.length === 0) continue;
+    const jsonl = rows
+      .map((r) =>
+        JSON.stringify({
+          eventId: r.eventId,
+          issueId: r.issueId,
+          title: r.title,
+          message: r.message,
+          level: r.level,
+          platform: r.platform,
+          timestamp: r.timestamp.toISOString(),
+          tags: r.tags,
+          payload: r.payload,
+        }),
+      )
+      .join("\n");
+    const gz = gzipSync(Buffer.from(jsonl, "utf-8"));
+    const [y, m, d] = day.split("-");
+    await putBuffer(ARCHIVE_BUCKET, `${PROJECT_KEY}/${y}/${m}/${d}.jsonl.gz`, gz, "application/gzip");
+    const marked = await prisma.sentryEvent.updateMany({
+      where: { projectKey: PROJECT_KEY, dumpedAt: null, timestamp: { gte: dayStart, lt: dayEnd } },
+      data: { dumpedAt: new Date() },
+    });
+    dumped += marked.count;
+  }
+  return dumped;
+}

apps/worker/src/lib/sentry.ts

@@ -0,0 +1,88 @@
+// Sentry API client (read-only) for the archival pipeline. Uses a User Auth
+// Token with org:read / project:read / event:read. EU-region orgs use the
+// de.sentry.io API base.
+const TOKEN = process.env.SENTRY_AUTH_TOKEN ?? "";
+const ORG = process.env.SENTRY_ORG ?? "";
+const PROJECT = process.env.SENTRY_PROJECT ?? "";
+const API_BASE = (process.env.SENTRY_API_BASE ?? "https://sentry.io/api/0").replace(/\/$/, "");
+
+export function isConfigured(): boolean {
+  return Boolean(TOKEN && ORG && PROJECT);
+}
+
+const headers = () => ({ Authorization: `Bearer ${TOKEN}`, Accept: "application/json" });
+
+// Sentry paginates via a Link header: <url>; rel="next"; results="true"; cursor="...".
+// Returns the next cursor only when results="true".
+function parseNextCursor(link: string | null): string | null {
+  if (!link) return null;
+  for (const part of link.split(",")) {
+    if (/rel="next"/.test(part) && /results="true"/.test(part)) {
+      const m = part.match(/cursor="([^"]+)"/);
+      if (m) return m[1];
+    }
+  }
+  return null;
+}
+
+type Page<T> = { data: T[]; nextCursor: string | null };
+
+async function getPage<T>(path: string, params: Record<string, string>): Promise<Page<T>> {
+  const url = new URL(`${API_BASE}${path}`);
+  for (const [k, v] of Object.entries(params)) if (v) url.searchParams.set(k, v);
+  const res = await fetch(url, { headers: headers() });
+  if (!res.ok) throw new Error(`sentry ${path} ${res.status}: ${await res.text().catch(() => "")}`);
+  const data = (await res.json()) as T[];
+  return { data, nextCursor: parseNextCursor(res.headers.get("link")) };
+}
+
+export type SentryIssueRaw = {
+  id: string;
+  shortId?: string;
+  title?: string;
+  culprit?: string;
+  level?: string;
+  status?: string;
+  type?: string;
+  count?: string | number;
+  userCount?: number;
+  firstSeen?: string;
+  lastSeen?: string;
+  permalink?: string;
+  metadata?: Record<string, unknown>;
+};
+
+export type SentryEventRaw = {
+  id?: string;
+  eventID?: string;
+  groupID?: string;
+  title?: string;
+  message?: string;
+  "event.type"?: string;
+  platform?: string;
+  dateCreated?: string;
+  dateReceived?: string;
+  tags?: unknown;
+  [k: string]: unknown;
+};
+
+// One page of org issues (newest activity first). statsPeriod bounds the window.
+export function listIssuesPage(cursor?: string, statsPeriod = "90d"): Promise<Page<SentryIssueRaw>> {
+  return getPage<SentryIssueRaw>(`/organizations/${ORG}/issues/`, {
+    project: "",
+    query: "",
+    statsPeriod,
+    limit: "100",
+    ...(cursor ? { cursor } : {}),
+  });
+}
+
+// One page of project events, full payloads, newest first.
+export function listEventsPage(cursor?: string): Promise<Page<SentryEventRaw>> {
+  return getPage<SentryEventRaw>(`/projects/${ORG}/${PROJECT}/events/`, {
+    full: "true",
+    ...(cursor ? { cursor } : {}),
+  });
+}
+
+export const sentryConfig = { ORG, PROJECT, API_BASE };

apps/worker/src/schedulers/pipeline.ts

@@ -13,6 +13,7 @@ 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";
+import { runArchiveSentry } from "../jobs/sentry-archive";
 
 const QUEUE = "insight-pipeline";
 
@@ -117,6 +118,15 @@ async function runJob(job: Job) {
       }
       return res;
     }
+    case "sentry-archive": {
+      const res = await runArchiveSentry();
+      if (res.eventsInserted > 0 || res.issuesSynced > 0 || res.dumped > 0 || res.error) {
+        console.log(
+          `[pipeline] sentry-archive issues=${res.issuesSynced} events=${res.eventsFetched}/${res.eventsInserted} dup=${res.eventsDuplicate} dumped=${res.dumped}${res.error ? ` error=${res.error}` : ""}`,
+        );
+      }
+      return res;
+    }
     default:
       return { ok: false, error: `unknown job ${job.name}` };
   }
@@ -183,6 +193,11 @@ export async function startInsightPipeline() {
     { pattern: "0 3 * * *" },
     { name: "posthog-identity-archive", data: {}, opts: { removeOnComplete: 30, removeOnFail: 15 } },
   );
+  await queue.upsertJobScheduler(
+    "sentry-archive",
+    { pattern: "0 * * * *" },
+    { name: "sentry-archive", data: {}, opts: { removeOnComplete: 30, removeOnFail: 15 } },
+  );
 
   new Worker(QUEUE, runJob, {
     connection: redis,
@@ -191,6 +206,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, posthog-identity-archive@03:00",
+    "[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, sentry-archive@hourly",
   );
 }