Move standard.site documents from the Fly proxy into D1

CLAUDE.md

@@ -218,7 +218,8 @@ AT Protocol (Bluesky PDS) + Fly.io Feed Proxy + Jetstream Firehose
 - **Framework:** SvelteKit 2.x with Svelte 5 runes
 - **Runtime:** Cloudflare Pages
 - **Database:** D1 (SQLite) - reads the same database as the backend
-- **Features:** Ops panel (cron liveness, firehose lag, proxy cache health) with 30-day trend
+- **Features:** Ops panel (cron liveness, firehose lag, document-stream lag + ingest saturation,
+  proxy cache health) with 30-day trend
   sparklines, system metrics, user management, feed health monitoring, search/sort/pagination
 - **Pages:** Dashboard (ops + metrics + trends), Users (list + detail), Feeds (health + error tracking).
   Feed health is the crawler's own verdict from `feeds.error_count` / `feeds.crawl_stale`, not an
@@ -234,6 +235,11 @@ AT Protocol (Bluesky PDS) + Fly.io Feed Proxy + Jetstream Firehose
    frontend refreshes with one `GET /api/v2/timeline` query that joins subscriptions and read
    state. Reads never touch Fly — see `docs/plans/D1_FEED_TIMELINE.md`
 4. **Social:** Jetstream firehose → D1 shares table → frontend polls for updates
+5. **Documents:** the JetstreamPoller DO drains `site.standard.document` (filtered to
+   subscribed authors) into D1, backfilling an author's back catalogue from their PDS at
+   subscribe time; `/api/v2/documents/batch` serves from D1 behind the
+   `documents_v2_enabled` gate, from the Fly proxy until it's flipped — see
+   `docs/plans/DOCUMENTS_TO_D1.md`
 
 ## AT Protocol Integration
 

admin/src/lib/metrics/ops.test.ts

@@ -28,6 +28,12 @@ const status = (overrides: Partial<OpsStatus> = {}): OpsStatus => ({
       processed: 12,
       errors: 0,
       alarmScheduled: true,
+      documentsLagMs: 30_000,
+      documentsProcessed: 3,
+      documentsErrors: 0,
+      documentsCapStreak: 0,
+      documentsAuthors: 42,
+      documentsIngestPaused: false,
     },
   },
   proxy: {
@@ -126,9 +132,45 @@ describe('ops tiles', () => {
     expect(fresh(null)).toBe('warning');
   });
 
+  it('grades the document stream separately from subscriptions', () => {
+    const s = status();
+    s.poller!.value.documentsLagMs = 20 * MINUTE;
+    const metrics = opsMetricsFrom(s, NOW);
+    expect(tile(metrics, 'Document Stream Lag').status).toBe('error');
+    // One stream stuck while the other is fine is exactly what a shared number
+    // would hide.
+    expect(tile(metrics, 'Firehose Lag').status).toBe('healthy');
+  });
+
+  it('escalates as capped cycles pile up, and says so when ingest is paused', () => {
+    const streak = (n: number) => {
+      const s = status();
+      s.poller!.value.documentsCapStreak = n;
+      return tile(opsMetricsFrom(s, NOW), 'Document Ingest');
+    };
+    // A burst draining across a couple of cycles is the design working.
+    expect(streak(0).value).toBe('Draining');
+    expect(streak(1).status).toBe('healthy');
+    expect(streak(4).status).toBe('warning');
+    expect(streak(12).status).toBe('error');
+
+    const paused = status();
+    paused.poller!.value.documentsIngestPaused = true;
+    const metrics = opsMetricsFrom(paused, NOW);
+    expect(tile(metrics, 'Document Ingest').value).toBe('Disabled');
+    // A paused stream's lag climbs by construction; don't render it as a stall.
+    expect(tile(metrics, 'Document Stream Lag').value).toBe('Paused');
+  });
+
+  it('says "no data" for a backend that predates the document stream', () => {
+    const s = status();
+    delete s.poller!.value.documentsLagMs;
+    expect(tile(opsMetricsFrom(s, NOW), 'Document Stream Lag').value).toBe('No data');
+  });
+
   it('renders a full set of tiles before the cron has ever run', () => {
     const metrics = opsMetricsFrom({ cron: null, poller: null, proxy: null }, NOW);
-    expect(metrics).toHaveLength(7);
+    expect(metrics).toHaveLength(9);
     expect(metrics.every((m) => m.status === 'error')).toBe(true);
   });
 });

admin/src/lib/metrics/ops.ts

@@ -67,7 +67,22 @@ function cronMetric(status: OpsStatus, now: number): MetricValue {
   };
 }
 
-const POLLER_LABELS = ['Firehose Lag', 'Last Poll', 'Poll Errors (last cycle)'] as const;
+const POLLER_LABELS = [
+  'Firehose Lag',
+  'Last Poll',
+  'Poll Errors (last cycle)',
+  'Document Stream Lag',
+  'Document Ingest',
+] as const;
+
+/**
+ * Consecutive capped document cycles before the tile goes red. Mirrors the
+ * backend's `DOCUMENT_CAP_SATURATION_ALERT_STREAK`: one or two capped cycles is a
+ * burst draining exactly as designed, ten in a row is a flood that wants the
+ * `documents_ingest_enabled` switch.
+ */
+const CAP_STREAK_ERROR = 10;
+const CAP_STREAK_WARN = 3;
 
 function pollerMetrics(status: OpsStatus, now: number): MetricValue[] {
   const poller = status.poller;
@@ -100,6 +115,50 @@ function pollerMetrics(status: OpsStatus, now: number): MetricValue[] {
           status: pollAge > POLL_STALE_MS ? 'error' : 'healthy',
         };
 
+  const {
+    documentsLagMs,
+    documentsCapStreak = 0,
+    documentsAuthors = 0,
+    documentsIngestPaused = false,
+  } = poller.value;
+
+  // The two streams fail independently — one stuck while the other is healthy is
+  // exactly what a single "firehose lag" number hides — so documents get their own
+  // tile, graded on the same thresholds.
+  const documentsLag: MetricValue =
+    documentsLagMs === undefined
+      ? unknown(POLLER_LABELS[3])
+      : documentsIngestPaused
+        ? { label: POLLER_LABELS[3], value: 'Paused', status: 'warning' }
+        : documentsLagMs === null
+          ? { label: POLLER_LABELS[3], value: '—', status: 'warning' }
+          : {
+              label: POLLER_LABELS[3],
+              value: formatAge(documentsLagMs),
+              status:
+                documentsLagMs >= LAG_ERROR_MS
+                  ? 'error'
+                  : documentsLagMs >= LAG_WARN_MS
+                    ? 'warning'
+                    : 'healthy',
+            };
+
+  // Cap saturation: how many cycles in a row the drain stopped early. This is the
+  // number an operator reads before deciding a burst has become a flood.
+  const documentIngest: MetricValue = documentsIngestPaused
+    ? { label: POLLER_LABELS[4], value: 'Disabled', status: 'error' }
+    : {
+        label: POLLER_LABELS[4],
+        value: documentsCapStreak === 0 ? 'Draining' : `${documentsCapStreak} capped cycles`,
+        unit: `${documentsAuthors} authors`,
+        status:
+          documentsCapStreak >= CAP_STREAK_ERROR
+            ? 'error'
+            : documentsCapStreak >= CAP_STREAK_WARN
+              ? 'warning'
+              : 'healthy',
+      };
+
   return [
     lag,
     lastPoll,
@@ -108,6 +167,8 @@ function pollerMetrics(status: OpsStatus, now: number): MetricValue[] {
       value: errors,
       status: errors > 0 ? 'warning' : 'healthy',
     },
+    documentsLag,
+    documentIngest,
   ];
 }
 

admin/src/lib/queries/ops.ts

@@ -15,6 +15,17 @@ export interface PollerStatusValue {
   processed: number;
   errors: number;
   alarmScheduled: boolean;
+  // The document stream (site.standard.document + reader collections), which the
+  // poller drains alongside subscriptions. Optional: a backend that predates it
+  // writes rows without these, and the tiles say "no data" rather than "0".
+  documentsLagMs?: number | null;
+  documentsProcessed?: number;
+  documentsErrors?: number;
+  /** Consecutive cycles that stopped on the per-cycle apply cap. */
+  documentsCapStreak?: number;
+  documentsAuthors?: number;
+  /** `documents_ingest_enabled` is off — a deliberate pause, not a stall. */
+  documentsIngestPaused?: boolean;
 }
 
 export interface CronLastRunValue {