fix(documents): record archive integrity checks in their own ledger so the nightly control advances again (#2108)

The 03:00 WORM verification cron stamped last_integrity_check_at on
document_attachments. enforce_period_lock_documents() fires on any UPDATE of a
row whose journal entry sits in a closed or locked period, without checking
whether the entry link actually changed, so a read-only integrity stamp was
rejected. The queue orders last_integrity_check_at ASC NULLS FIRST, so the
rejected rows re-sorted to the head every night and the batch became
permanently 200/200 blocked. Both call sites discarded the update error, so
nothing logged and nothing alerted.

Prod state: 34 557 current-version documents, 24 083 never checked, last
successful stamp 2026-08-31 03:00, nightly successes already decayed to
single digits.

Migration 017's enforcement triggers are legally required and never-touch, so
this does not narrow the trigger. The verification outcome moves to its own
document_integrity_checks table and the cron stops writing document_attachments
altogether, which takes the trigger off the write path. The legacy column stays
in place. Failures are now counted, logged and reported in the route's summary:
the silence is why this went unnoticed for weeks.


Claude-Session: https://claude.ai/code/session_016ifKg6Ec67A39oxfGPU1yc

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

app/api/documents/verify/cron/__tests__/route.test.ts

@@ -1,12 +1,17 @@
 /**
  * Tests for the nightly document integrity-verify cron.
  *
- * Covers the two production defects fixed in this route:
- * - the run must fit its budget (maxDuration 300 + batch default 200), and
+ * Covers the production defects fixed in this route:
+ * - the run must fit its budget (maxDuration 300 + batch default 200),
  * - a document whose storage object cannot be downloaded must surface as an
- *   audit incident (INTEGRITY_FAILURE / DOCUMENT_OBJECT_MISSING) AND get its
- *   last_integrity_check_at stamped so it stops head-blocking the
- *   nulls-first queue every night.
+ *   audit incident (INTEGRITY_FAILURE / DOCUMENT_OBJECT_MISSING) AND get a
+ *   ledger row so it stops head-blocking the queue,
+ * - the queue and the stamp both live in document_integrity_checks: the route
+ *   must never write to document_attachments, whose UPDATE trigger
+ *   (enforce_period_lock_documents, migration 017) rejects every document
+ *   linked to a closed/locked period and wedged the cron at 200/200 rejected,
+ * - and a rejected ledger or audit write must be counted, logged and reported
+ *   rather than discarded, which is why the stall went unnoticed for weeks.
  */
 import { describe, it, expect, vi, beforeEach } from 'vitest'
 import { createHash } from 'node:crypto'
@@ -24,39 +29,36 @@ interface MockDoc {
   storage_path: string
   sha256_hash: string
   file_name: string
+  last_checked_at: string | null
 }
 
 const state = {
   documents: [] as MockDoc[],
   fetchError: null as { message: string } | null,
   downloadResults: new Map<string, { data: unknown; error: { message: string } | null }>(),
-  updates: [] as Array<{ values: Record<string, unknown>; id: string }>,
+  rpcCalls: [] as Array<{ fn: string; args: Record<string, unknown> }>,
+  ledgerInserts: [] as Array<Record<string, unknown>>,
+  ledgerInsertError: null as { message: string } | null,
   auditInserts: [] as Array<Record<string, unknown>>,
   auditInsertError: null as { message: string } | null,
-  limitCalls: [] as number[],
 }
 
 function makeMockClient() {
   return {
+    rpc: (fn: string, args: Record<string, unknown>) => {
+      state.rpcCalls.push({ fn, args })
+      return Promise.resolve({
+        data: state.fetchError ? null : state.documents,
+        error: state.fetchError,
+      })
+    },
     from: (table: string) => {
-      if (table === 'document_attachments') {
+      if (table === 'document_integrity_checks') {
         return {
-          select: () => ({
-            eq: () => ({
-              order: () => ({
-                limit: (n: number) => {
-                  state.limitCalls.push(n)
-                  return Promise.resolve({ data: state.documents, error: state.fetchError })
-                },
-              }),
-            }),
-          }),
-          update: (values: Record<string, unknown>) => ({
-            eq: (_column: string, id: string) => {
-              state.updates.push({ values, id })
-              return Promise.resolve({ error: null })
-            },
-          }),
+          insert: (row: Record<string, unknown>) => {
+            state.ledgerInserts.push(row)
+            return Promise.resolve({ error: state.ledgerInsertError })
+          },
         }
       }
       if (table === 'audit_log') {
@@ -67,6 +69,12 @@ function makeMockClient() {
           },
         }
       }
+      if (table === 'document_attachments') {
+        // The whole point of the fix: a write here runs through
+        // enforce_period_lock_documents() and is rejected for every document
+        // linked to a closed/locked period.
+        throw new Error('the verify cron must never write to document_attachments')
+      }
       throw new Error(`unexpected table: ${table}`)
     },
     storage: {
@@ -101,6 +109,7 @@ function makeDoc(overrides: Partial<MockDoc> = {}): MockDoc {
     storage_path: 'user-1/company-1/inbox/file.pdf',
     sha256_hash: 'deadbeef',
     file_name: 'file.pdf',
+    last_checked_at: null,
     ...overrides,
   }
 }
@@ -124,10 +133,11 @@ beforeEach(() => {
   state.documents = []
   state.fetchError = null
   state.downloadResults.clear()
-  state.updates = []
+  state.rpcCalls = []
+  state.ledgerInserts = []
+  state.ledgerInsertError = null
   state.auditInserts = []
   state.auditInsertError = null
-  state.limitCalls = []
 })
 
 describe('GET /api/documents/verify/cron', () => {
@@ -139,23 +149,28 @@ describe('GET /api/documents/verify/cron', () => {
     const response = await GET(cronRequest())
 
     expect(response.status).toBe(401)
-    expect(state.limitCalls).toHaveLength(0)
+    expect(state.rpcCalls).toHaveLength(0)
   })
 
   it('declares a 300s function budget', () => {
     expect(maxDuration).toBe(300)
   })
 
-  it('requests a batch of 200 by default and honors the env override', async () => {
+  it('draws the batch from the ledger queue, 200 by default, env override honored', async () => {
     await GET(cronRequest())
-    expect(state.limitCalls).toEqual([200])
+    expect(state.rpcCalls).toEqual([
+      { fn: 'next_documents_for_integrity_check', args: { p_limit: 200 } },
+    ])
 
     process.env.DOCUMENT_VERIFY_BATCH_SIZE = '50'
     await GET(cronRequest())
-    expect(state.limitCalls).toEqual([200, 50])
+    expect(state.rpcCalls[1]).toEqual({
+      fn: 'next_documents_for_integrity_check',
+      args: { p_limit: 50 },
+    })
   })
 
-  it('stamps last_integrity_check_at on a successful verification', async () => {
+  it('appends a passed ledger row on a successful verification', async () => {
     const doc = makeDoc()
     const hash = registerObject(doc.storage_path, '%PDF-1.4 demo content')
     state.documents = [{ ...doc, sha256_hash: hash }]
@@ -168,15 +183,24 @@ describe('GET /api/documents/verify/cron', () => {
       verified: 1,
       failures: 0,
       missingObjects: 0,
+      writeFailures: 0,
       errors: 0,
     })
-    expect(state.updates).toHaveLength(1)
-    expect(state.updates[0].id).toBe(doc.id)
-    expect(state.updates[0].values.last_integrity_check_at).toEqual(expect.any(String))
+    expect(state.ledgerInserts).toHaveLength(1)
+    expect(state.ledgerInserts[0]).toMatchObject({
+      document_id: doc.id,
+      company_id: doc.company_id,
+      result: 'passed',
+      expected_sha256: hash,
+      computed_sha256: hash,
+      storage_path: doc.storage_path,
+      detail: null,
+    })
+    expect(state.ledgerInserts[0].checked_at).toEqual(expect.any(String))
     expect(state.auditInserts).toHaveLength(0)
   })
 
-  it('writes an INTEGRITY_FAILURE audit row and still stamps on hash mismatch', async () => {
+  it('writes an INTEGRITY_FAILURE audit row and a hash_mismatch ledger row on mismatch', async () => {
     const doc = makeDoc({ sha256_hash: 'not-the-real-hash' })
     registerObject(doc.storage_path, 'tampered content')
     state.documents = [doc]
@@ -186,10 +210,16 @@ describe('GET /api/documents/verify/cron', () => {
 
     expect(json.failures).toBe(1)
     expect(json.missingObjects).toBe(0)
-    expect(state.updates).toHaveLength(1)
+    expect(json.writeFailures).toBe(0)
     expect(state.auditInserts).toHaveLength(1)
     expect(state.auditInserts[0].action).toBe('INTEGRITY_FAILURE')
     expect(String(state.auditInserts[0].description)).not.toContain('DOCUMENT_OBJECT_MISSING')
+    expect(state.ledgerInserts).toHaveLength(1)
+    expect(state.ledgerInserts[0]).toMatchObject({
+      document_id: doc.id,
+      result: 'hash_mismatch',
+      expected_sha256: 'not-the-real-hash',
+    })
   })
 
   it('verifies via the company-scoped fallback key when a concurrent backfill re-homed the object', async () => {
@@ -208,17 +238,18 @@ describe('GET /api/documents/verify/cron', () => {
     const json = await response.json()
 
     expect(state.auditInserts).toHaveLength(0)
-    expect(state.updates.map((u) => u.id)).toEqual(['doc-repointed'])
+    expect(state.ledgerInserts.map((row) => row.document_id)).toEqual(['doc-repointed'])
     expect(json).toEqual({
       processed: 1,
       verified: 1,
       failures: 0,
       missingObjects: 0,
+      writeFailures: 0,
       errors: 0,
     })
   })
 
-  it('surfaces a missing storage object as an audit incident AND stamps the check', async () => {
+  it('surfaces a missing storage object as an audit incident AND an object_missing ledger row', async () => {
     const missing = makeDoc({ id: 'doc-missing', storage_path: 'user-1/company-1/gone.pdf' })
     const healthy = makeDoc({ id: 'doc-healthy', storage_path: 'user-1/company-1/ok.pdf' })
     const healthyHash = registerObject(healthy.storage_path, 'healthy content')
@@ -235,20 +266,28 @@ describe('GET /api/documents/verify/cron', () => {
     expect(String(audit.description)).toContain('DOCUMENT_OBJECT_MISSING')
     expect(audit.new_state).toMatchObject({ reason: 'DOCUMENT_OBJECT_MISSING' })
 
-    // Both documents are stamped: the failing one must stop head-blocking
-    // the nulls-first queue, and the healthy one was verified.
-    expect(state.updates.map((u) => u.id).sort()).toEqual(['doc-healthy', 'doc-missing'])
+    // Both documents get a ledger row: the failing one must stop head-blocking
+    // the queue, and the healthy one was verified.
+    expect(state.ledgerInserts.map((row) => row.document_id).sort()).toEqual([
+      'doc-healthy',
+      'doc-missing',
+    ])
+    expect(state.ledgerInserts.find((row) => row.document_id === 'doc-missing')).toMatchObject({
+      result: 'object_missing',
+      computed_sha256: null,
+    })
 
     expect(json).toEqual({
       processed: 2,
       verified: 1,
       failures: 0,
       missingObjects: 1,
+      writeFailures: 0,
       errors: 1,
     })
   })
 
-  it('does not stamp a missing object when the audit insert fails, so it retries next run', async () => {
+  it('does not write a ledger row when the audit insert fails, so it retries next run', async () => {
     const missing = makeDoc({ id: 'doc-missing', storage_path: 'user-1/company-1/gone.pdf' })
     state.documents = [missing]
     state.auditInsertError = { message: 'insert blocked' }
@@ -257,12 +296,49 @@ describe('GET /api/documents/verify/cron', () => {
     const json = await response.json()
 
     expect(state.auditInserts).toHaveLength(1)
-    expect(state.updates).toHaveLength(0)
+    expect(state.ledgerInserts).toHaveLength(0)
     expect(json.missingObjects).toBe(0)
+    expect(json.writeFailures).toBe(1)
     expect(json.errors).toBe(1)
   })
 
-  it('returns an error envelope when the document fetch fails', async () => {
+  it('reports a rejected ledger write instead of swallowing it', async () => {
+    const doc = makeDoc()
+    const hash = registerObject(doc.storage_path, 'healthy content')
+    state.documents = [{ ...doc, sha256_hash: hash }]
+    state.ledgerInsertError = { message: 'permission denied for table document_integrity_checks' }
+
+    const response = await GET(cronRequest())
+    const json = await response.json()
+
+    expect(state.ledgerInserts).toHaveLength(1)
+    expect(json).toEqual({
+      processed: 1,
+      verified: 0,
+      failures: 0,
+      missingObjects: 0,
+      writeFailures: 1,
+      errors: 1,
+    })
+  })
+
+  it('counts a rejected audit write on a hash mismatch as a write failure', async () => {
+    const doc = makeDoc({ sha256_hash: 'not-the-real-hash' })
+    registerObject(doc.storage_path, 'tampered content')
+    state.documents = [doc]
+    state.auditInsertError = { message: 'insert blocked' }
+
+    const response = await GET(cronRequest())
+    const json = await response.json()
+
+    expect(state.auditInserts).toHaveLength(1)
+    expect(state.ledgerInserts).toHaveLength(0)
+    expect(json.failures).toBe(0)
+    expect(json.writeFailures).toBe(1)
+    expect(json.errors).toBe(1)
+  })
+
+  it('returns an error envelope when the queue fetch fails', async () => {
     state.fetchError = { message: 'db down' }
 
     const response = await GET(cronRequest())

app/api/documents/verify/cron/route.ts

@@ -7,21 +7,71 @@ import { errorResponse, errorResponseFromCode } from '@/lib/errors/get-structure
 /**
  * GET /api/documents/verify/cron: nightly 03:00 UTC (schedule in vercel.json).
  * Spot-checks WORM archive integrity by recomputing SHA-256 for the next
- * batch of documents and writing INTEGRITY_FAILURE rows to the audit log
- * for any mismatches. Documents whose storage object cannot be downloaded
- * get an INTEGRITY_FAILURE row marked DOCUMENT_OBJECT_MISSING and are still
- * stamped as checked so they stop head-blocking the nulls-first queue.
+ * batch of documents and appending the outcome to document_integrity_checks.
+ * Hash mismatches and unreadable storage objects additionally get an
+ * INTEGRITY_FAILURE row in the audit log, which is the durable incident
+ * surface.
+ *
+ * Both the queue and the stamp live in document_integrity_checks and NOT in
+ * document_attachments.last_integrity_check_at (now legacy): any UPDATE of a
+ * document linked to an entry in a closed/locked period is rejected by
+ * enforce_period_lock_documents() (migration 017, legally required and never
+ * touched), which had this cron wedged at 200 of 200 rejected per night with
+ * both write errors discarded. See
+ * supabase/migrations/20260901130000_document_integrity_checks.sql.
  */
 
 // Vercel function budget; verification is sequential, see batch size below.
 export const maxDuration = 300
 
-// Measured ~0.8s per document (download + hash + stamp), so 200 documents
-// finish in ~160s with headroom inside the 300s budget. The previous default
-// of 500 hit the platform timeout around item ~250 every night, so the tail
-// of the queue was never reached.
+// Measured ~0.8s per document (download + hash + ledger append), so 200
+// documents finish in ~160s with headroom inside the 300s budget. The previous
+// default of 500 hit the platform timeout around item ~250 every night, so the
+// tail of the queue was never reached.
 const DEFAULT_VERIFY_BATCH_SIZE = 200
 
+type ServiceClient = ReturnType<typeof createServiceRoleClient>
+
+type IntegrityResult = 'passed' | 'hash_mismatch' | 'object_missing'
+
+/** One row of public.next_documents_for_integrity_check(). */
+interface QueuedDocument {
+  id: string
+  user_id: string
+  company_id: string
+  storage_path: string
+  sha256_hash: string
+  file_name: string
+  last_checked_at: string | null
+}
+
+/**
+ * Append the outcome of one check to the integrity ledger. The write error is
+ * returned, never swallowed: a lost ledger row leaves the document at the head
+ * of the queue, and discarding exactly this error is why the control stayed
+ * dead for weeks without anyone noticing.
+ */
+async function recordCheck(
+  supabase: ServiceClient,
+  doc: QueuedDocument,
+  result: IntegrityResult,
+  computedHash: string | null,
+  detail: string | null
+): Promise<{ error: { message: string } | null }> {
+  const { error } = await supabase.from('document_integrity_checks').insert({
+    company_id: doc.company_id,
+    document_id: doc.id,
+    checked_at: new Date().toISOString(),
+    expected_sha256: doc.sha256_hash,
+    computed_sha256: computedHash,
+    storage_path: doc.storage_path,
+    result,
+    detail,
+  })
+
+  return { error }
+}
+
 export const GET = withCronContext('cron.documents_verify', async (_request, ctx) => {
   const supabaseUrl = process.env.NEXT_PUBLIC_SUPABASE_URL
   const supabaseServiceKey = process.env.SUPABASE_SERVICE_ROLE_KEY
@@ -35,25 +85,33 @@ export const GET = withCronContext('cron.documents_verify', async (_request, ctx
 
   const supabase = createServiceRoleClient(supabaseUrl, supabaseServiceKey)
 
-  const { data: documents, error: fetchError } = await supabase
-    .from('document_attachments')
-    .select('id, user_id, company_id, storage_path, sha256_hash, file_name')
-    .eq('is_current_version', true)
-    .order('last_integrity_check_at', { ascending: true, nullsFirst: true })
-    .limit(parseInt(process.env.DOCUMENT_VERIFY_BATCH_SIZE || '', 10) || DEFAULT_VERIFY_BATCH_SIZE)
+  const batchSize =
+    parseInt(process.env.DOCUMENT_VERIFY_BATCH_SIZE || '', 10) || DEFAULT_VERIFY_BATCH_SIZE
+
+  // Least-recently-checked first, never-checked ahead of those, tie-broken on
+  // created_at so the drain is a deterministic FIFO rather than heap order.
+  const { data, error: fetchError } = await supabase.rpc('next_documents_for_integrity_check', {
+    p_limit: batchSize,
+  })
 
   if (fetchError) {
     ctx.log.error('failed to fetch documents for verify', fetchError)
     return errorResponse(fetchError, ctx.log, { requestId: ctx.requestId })
   }
 
-  if (!documents || documents.length === 0) {
+  const documents = (data ?? []) as QueuedDocument[]
+
+  if (documents.length === 0) {
     return NextResponse.json({ message: 'No documents to verify', processed: 0 })
   }
 
   let verified = 0
   let failures = 0
   let missingObjects = 0
+  // Ledger and audit writes the database rejected. Counted, logged and shipped
+  // in the response so a broken write path shows up on the first run instead
+  // of weeks later.
+  let writeFailures = 0
 
   const summary = await ctx.forEach('document', documents, async (doc, itemCtx) => {
     // Dual-layout download: the batch is snapshotted up front, and a
@@ -89,18 +147,31 @@ export const GET = withCronContext('cron.documents_verify', async (_request, ctx
       })
 
       if (auditError) {
-        // Leave last_integrity_check_at untouched so the document is retried
-        // (and the incident write re-attempted) on the next run.
+        // Append no ledger row, so the document keeps its place at the head of
+        // the queue and the incident write is re-attempted on the next run.
+        writeFailures++
+        itemCtx.log.error('audit insert failed for missing object', new Error(auditError.message), {
+          documentId: doc.id,
+        })
         throw new Error(`audit insert failed for missing object: ${auditError.message}`)
       }
 
-      // Stamp the check so the row stops sorting to the head of the
-      // nulls-first queue every night; the audit row above is the durable
-      // incident surface.
-      await supabase
-        .from('document_attachments')
-        .update({ last_integrity_check_at: new Date().toISOString() })
-        .eq('id', doc.id)
+      const { error: ledgerError } = await recordCheck(
+        supabase,
+        doc,
+        'object_missing',
+        null,
+        `DOCUMENT_OBJECT_MISSING: ${reason}`
+      )
+
+      if (ledgerError) {
+        writeFailures++
+        itemCtx.log.error('integrity ledger write failed', new Error(ledgerError.message), {
+          documentId: doc.id,
+          result: 'object_missing',
+        })
+        throw new Error(`integrity ledger write failed: ${ledgerError.message}`)
+      }
 
       missingObjects++
       itemCtx.log.error('document object missing', new Error(reason), {
@@ -118,13 +189,11 @@ export const GET = withCronContext('cron.documents_verify', async (_request, ctx
 
     const isValid = computedHash === doc.sha256_hash
 
-    await supabase
-      .from('document_attachments')
-      .update({ last_integrity_check_at: new Date().toISOString() })
-      .eq('id', doc.id)
-
     if (!isValid) {
-      await supabase.from('audit_log').insert({
+      // Audit row before ledger row: if the ledger row landed first and the
+      // audit write then failed, the document would leave the queue with the
+      // incident lost. This order re-checks it tomorrow instead.
+      const { error: auditError } = await supabase.from('audit_log').insert({
         user_id: doc.user_id,
         company_id: doc.company_id,
         action: 'INTEGRITY_FAILURE',
@@ -135,6 +204,33 @@ export const GET = withCronContext('cron.documents_verify', async (_request, ctx
         new_state: { computed_hash: computedHash },
       })
 
+      if (auditError) {
+        writeFailures++
+        itemCtx.log.error('audit insert failed for hash mismatch', new Error(auditError.message), {
+          documentId: doc.id,
+        })
+        throw new Error(`audit insert failed for hash mismatch: ${auditError.message}`)
+      }
+    }
+
+    const { error: ledgerError } = await recordCheck(
+      supabase,
+      doc,
+      isValid ? 'passed' : 'hash_mismatch',
+      computedHash,
+      isValid ? null : `stored hash ${doc.sha256_hash}, computed hash ${computedHash}`
+    )
+
+    if (ledgerError) {
+      writeFailures++
+      itemCtx.log.error('integrity ledger write failed', new Error(ledgerError.message), {
+        documentId: doc.id,
+        result: isValid ? 'passed' : 'hash_mismatch',
+      })
+      throw new Error(`integrity ledger write failed: ${ledgerError.message}`)
+    }
+
+    if (!isValid) {
       itemCtx.log.error('integrity failure', new Error('hash_mismatch'), {
         documentId: doc.id,
         fileName: doc.file_name,
@@ -152,14 +248,25 @@ export const GET = withCronContext('cron.documents_verify', async (_request, ctx
     verified,
     failures,
     missingObjects,
+    writeFailures,
     downloadErrors: summary.failed,
   })
 
+  if (writeFailures > 0) {
+    // Loud on its own line: a rejected ledger write means the queue does not
+    // advance, which is exactly the silent stall this cron just came out of.
+    ctx.log.error('integrity ledger writes rejected', new Error('integrity_write_failed'), {
+      processed: summary.total,
+      writeFailures,
+    })
+  }
+
   return NextResponse.json({
     processed: summary.total,
     verified,
     failures,
     missingObjects,
+    writeFailures,
     errors: summary.failed,
   })
 })