fix(ingest): rolling lookback + ongoing-aware watermark to prevent missed sessions

Bug: previous logic advanced watermark to the latest session start_time including
ongoing sessions. PostHog session_recordings filters by start_time, so once a
session was 'seen' as ongoing the watermark moved past its start time and the
session was never re-fetched after it ended. Today 4 auth sessions on
/dashboard/vehicles/* and /dashboard/search (07:17-07:40 UTC) were lost this way.

Fix:
1. ROLLING_LOOKBACK_MINUTES (default 60): every cycle queries date_from =
   min(watermark, now - 60min). Sessions that just finished get re-fetched
   regardless of watermark drift. Upsert dedupes.
2. Track earliestOngoingStart; cap watermark to (earliestOngoingStart - 1s)
   so subsequent cycles re-read that range.

Also added GET/DELETE /api/insights/watermark for manual reset (used to
trigger 24h backfill after this deploy).

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

apps/web/src/app/api/insights/watermark/route.ts

@@ -0,0 +1,24 @@
+import { NextResponse } from "next/server";
+import { auth } from "@/lib/auth";
+import { headers } from "next/headers";
+import { prisma } from "@/lib/db";
+
+export const dynamic = "force-dynamic";
+
+export async function DELETE(req: Request) {
+  const session = await auth.api.getSession({ headers: await headers() });
+  if (!session) return NextResponse.json({ ok: false, error: "unauthenticated" }, { status: 401 });
+
+  const url = new URL(req.url);
+  const projectKey = url.searchParams.get("project") ?? "sase";
+  const deleted = await prisma.ingestionWatermark.deleteMany({ where: { projectKey } });
+  return NextResponse.json({ ok: true, projectKey, deleted: deleted.count });
+}
+
+export async function GET() {
+  const session = await auth.api.getSession({ headers: await headers() });
+  if (!session) return NextResponse.json({ ok: false, error: "unauthenticated" }, { status: 401 });
+
+  const rows = await prisma.ingestionWatermark.findMany();
+  return NextResponse.json({ ok: true, rows });
+}

apps/worker/src/jobs/posthog-ingest.ts

@@ -5,6 +5,9 @@ import { hashUserId } from "../lib/hash";
 
 const PROJECT_KEY = process.env.INSIGHT_PROJECT_KEY ?? "sase";
 const LOOKBACK_HOURS_INITIAL = Number(process.env.INSIGHT_LOOKBACK_HOURS ?? "24");
+// Always re-scan the last N minutes regardless of watermark, so sessions that
+// were "ongoing" on an earlier cycle get re-fetched once they finish.
+const ROLLING_LOOKBACK_MINUTES = Number(process.env.INSIGHT_ROLLING_LOOKBACK_MIN ?? "60");
 
 export type IngestResult = {
   fetched: number;
@@ -31,8 +34,12 @@ export async function runPostHogIngest(): Promise<IngestResult> {
 
   const now = new Date();
   const wm = await prisma.ingestionWatermark.findUnique({ where: { projectKey: PROJECT_KEY } });
+  // Watermark gives an upper bound, but we always look back ROLLING_LOOKBACK_MINUTES
+  // so previously-ongoing sessions get re-fetched once they finish (PostHog filters by
+  // start_time, which is fixed before the session ends — strict watermark misses these).
+  const rollingFloor = new Date(now.getTime() - ROLLING_LOOKBACK_MINUTES * 60_000);
   const dateFrom = wm
-    ? wm.lastPolledAt.toISOString()
+    ? new Date(Math.min(wm.lastPolledAt.getTime(), rollingFloor.getTime())).toISOString()
     : new Date(now.getTime() - LOOKBACK_HOURS_INITIAL * 3600_000).toISOString();
   const dateTo = now.toISOString();
 
@@ -44,6 +51,7 @@ export async function runPostHogIngest(): Promise<IngestResult> {
   let offset = 0;
   const limit = 100;
   let latestStart: Date = wm?.lastPolledAt ?? new Date(now.getTime() - LOOKBACK_HOURS_INITIAL * 3600_000);
+  let earliestOngoingStart: Date | null = null;
 
   // Cap pages to avoid runaway long jobs
   for (let page = 0; page < 10; page++) {
@@ -67,14 +75,19 @@ export async function runPostHogIngest(): Promise<IngestResult> {
     for (const rec of items) {
       fetched++;
       const startedAt = new Date(rec.start_time);
-      if (startedAt > latestStart) latestStart = startedAt;
 
-      // Skip ongoing recordings — wait until they finish
+      // Skip ongoing recordings — wait until they finish. Track the earliest
+      // ongoing start so we don't advance the watermark past it.
       if (rec.ongoing) {
+        if (!earliestOngoingStart || startedAt < earliestOngoingStart) {
+          earliestOngoingStart = startedAt;
+        }
         discarded++;
         continue;
       }
 
+      if (startedAt > latestStart) latestStart = startedAt;
+
       const decision = filterRecording(rec);
       if (!decision.keep) {
         discarded++;
@@ -127,10 +140,16 @@ export async function runPostHogIngest(): Promise<IngestResult> {
     offset += limit;
   }
 
+  // If there were ongoing sessions in this batch, cap watermark just before the
+  // earliest ongoing start so the next cycle re-fetches them once they finish.
+  let watermarkAt = latestStart;
+  if (earliestOngoingStart && watermarkAt >= earliestOngoingStart) {
+    watermarkAt = new Date(earliestOngoingStart.getTime() - 1000);
+  }
   await prisma.ingestionWatermark.upsert({
     where: { projectKey: PROJECT_KEY },
-    create: { projectKey: PROJECT_KEY, lastPolledAt: latestStart },
-    update: { lastPolledAt: latestStart },
+    create: { projectKey: PROJECT_KEY, lastPolledAt: watermarkAt },
+    update: { lastPolledAt: watermarkAt },
   });
 
   return {
@@ -139,6 +158,6 @@ export async function runPostHogIngest(): Promise<IngestResult> {
     discarded,
     inserted,
     duplicates,
-    watermarkAdvancedTo: latestStart.toISOString(),
+    watermarkAdvancedTo: watermarkAt.toISOString(),
   };
 }