fix(reads): page the nine remaining unbounded sweeps and the digest RPC (#548)

PostgREST caps every response at the API "Max rows" limit and returns the
capped result as an ordinary 200, so truncation is indistinguishable from a
complete read. #533 shipped `fetchAllRows` but wired it to only two call
sites; nine reads were left unbounded.

Two of them failed as wrong user-visible behaviour rather than a wrong number:

  - The digest's `notification_preferences` read. `resolveChannels(undefined)`
    treats a missing row as the schema default `{ inApp: true, email: true }`,
    so every student the truncated read dropped got emailed — including the
    ones who had turned email off. A short read could not mean "send more
    email"; now the read is chunked and paged, and a failure skips the tenant
    instead of falling through with an empty preferences map.
  - The digest's idempotency read. One notification row is written per
    recipient, so past the cap it truncated and a cron retry re-sent the day's
    digests, inverting the module's documented "re-runs never double-send"
    guarantee exactly when a retry happens.

`get_daily_digest_candidates` could not be fixed from the caller side: it is a
SETOF function with no ORDER BY, so `.range()` was not a stable window and
there was no exact count to verify against. It now takes a keyset cursor on
(tenant_id, user_id) — UNIQUE via tenant_users_unique — with ORDER BY and a
clamped LIMIT. Keyset rather than OFFSET because the candidate set is computed
live and shifts under a multi-second sweep. The caller stops on an EMPTY page,
never a short one: a short page is ambiguous between "end of set" and "the row
cap clamped us", and reading it as "done" is the original bug in a new place.

Also adds `fetchAllRowsIn` for `.in()` lookups. Those have a second ceiling on
the REQUEST side — a few hundred uuids overflow the gateway's URL limit — and
response paging cannot help with a request that is never made.

Verified by forcing the cap, not by a green run on a small dataset:

  - Live, against real PostgREST with the row cap lowered to 5 and 37
    transactions / 23 payouts / 37 candidates seeded past it
    (scripts/qa-548-cap-forcing.mts). Unpaged reads come back with 5 rows while
    count reports 37; every paged read returns all 37 and the sums are exact.
    The opted-out student at index 30 is invisible to the unpaged read and
    found by the paged one.
  - In unit tests, via a fake PostgREST that clamps every response to 7 rows
    over a 40-student fixture. Reverting the fix fails 6 of them with exactly
    the expected numbers (7 of 40 sent; the opted-out student emailed).

`tests/unit/unbounded-read-guard.test.ts` statically scans the tree for reads
against growth tables that never bound themselves, so the next unbounded sweep
fails in review. Pre-existing sites are recorded in KNOWN_UNBOUNDED as a
ratchet, split into `scoped` (bounded by a user or a course) and `gap` (real
platform-wide sweeps still outstanding under #540); a stale entry also fails,
so the list cannot rot.

`rollover_all_leagues`' tenant loop is untouched — it is PL/pgSQL and never
crosses PostgREST, so it is not subject to the cap.

Gates: typecheck, lint, build, test:unit 446/446 (was 411).

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0185GrowMKwTrLQE1it2HNGN

Author: guillermoscript

app/[locale]/dashboard/admin/payouts/page.tsx

@@ -4,6 +4,7 @@ import { getTranslations } from 'next-intl/server'
 import { format } from 'date-fns'
 import { es, enUS } from 'date-fns/locale'
 import { getCurrentTenantId, getCurrentUserId } from '@/lib/supabase/tenant'
+import { fetchAllRows } from '@/lib/supabase/fetch-all-rows'
 import { formatByCurrency, formatMoney, sumByCurrency } from '@/lib/payments/format-money'
 import { AdminBreadcrumb } from '@/components/admin/admin-breadcrumb'
 import { Card, CardContent, CardHeader, CardTitle } from '@/components/ui/card'
@@ -62,16 +63,23 @@ export default async function AdminPayoutsPage({
     )
   }
 
-  // Fetch payouts for this tenant
-  const { data: payouts } = await supabase
-    .from('payouts')
-    .select(
-      'payout_id, amount, currency, status, period_start, period_end, stripe_payout_id, paid_at, failure_reason, created_at, note, payout_method'
-    )
-    .eq('tenant_id', tenantId)
-    .order('created_at', { ascending: false })
-
-  const rows = payouts || []
+  // Fetch payouts for this tenant. Paged and count-verified (#548): the
+  // summary below sums this list, so a read truncated at the API row cap would
+  // understate the school's own "Total paid" without any error to notice.
+  // `created_at` is not unique, so `payout_id` breaks ties — an unstable sort
+  // makes `.range()` windows overlap or skip.
+  const rows = await fetchAllRows('payouts', (from, to) =>
+    supabase
+      .from('payouts')
+      .select(
+        'payout_id, amount, currency, status, period_start, period_end, stripe_payout_id, paid_at, failure_reason, created_at, note, payout_method',
+        { count: 'exact' }
+      )
+      .eq('tenant_id', tenantId)
+      .order('created_at', { ascending: false })
+      .order('payout_id', { ascending: false })
+      .range(from, to)
+  )
 
   // Summary calculations. Paid payouts are totalled per currency and rendered as
   // one figure each — a school selling in USD and EUR reads two figures, not

app/[locale]/dashboard/teacher/revenue/page.tsx

@@ -1,5 +1,6 @@
 import { createAdminClient } from '@/lib/supabase/admin'
 import { getCurrentTenantId } from '@/lib/supabase/tenant'
+import { fetchAllRows } from '@/lib/supabase/fetch-all-rows'
 import { getUserRole } from '@/lib/supabase/get-user-role'
 import { redirect } from 'next/navigation'
 import { getTranslations } from 'next-intl/server'
@@ -23,12 +24,23 @@ export default async function RevenuePage() {
   const tenantId = await getCurrentTenantId()
   const t = await getTranslations('dashboard.teacher.revenue')
 
-  // Parallelize all 4 independent queries
-  const [{ data: tenant }, { data: split }, { data: transactions }, { data: payouts }] = await Promise.all([
+  // Parallelize all 4 independent queries.
+  //
+  // The transaction read is paged and count-verified (#548) — every figure on
+  // this page is a sum over it, so truncation at the API row cap would quietly
+  // understate the school's revenue rather than fail. `created_at` is not
+  // unique; `transaction_id` breaks ties so the paging windows are stable.
+  // The payout read is already bounded by `.limit(10)`.
+  const [{ data: tenant }, { data: split }, transactions, { data: payouts }] = await Promise.all([
     supabase.from('tenants').select('name, stripe_account_id').eq('id', tenantId).single(),
     supabase.from('revenue_splits').select('platform_percentage, school_percentage').eq('tenant_id', tenantId).single(),
-    supabase.from('transactions').select('amount, status, payment_provider, created_at')
-      .eq('tenant_id', tenantId).eq('status', 'successful').order('created_at', { ascending: false }),
+    fetchAllRows('transactions', (from, to) =>
+      supabase.from('transactions').select('amount, status, payment_provider, created_at', { count: 'exact' })
+        .eq('tenant_id', tenantId).eq('status', 'successful')
+        .order('created_at', { ascending: false })
+        .order('transaction_id', { ascending: false })
+        .range(from, to)
+    ),
     supabase.from('payouts').select('*').eq('tenant_id', tenantId).order('created_at', { ascending: false }).limit(10),
   ])
 
@@ -114,6 +126,10 @@ export default async function RevenuePage() {
               <p className="mt-0.5 text-xs text-amber-700 dark:text-amber-400">
                 {t('stripeNotConnected.description')}
               </p>
+              {/* Stays a plain anchor: /api/stripe/connect is a route handler that
+                  mints a Stripe onboarding link and redirects. next/link would
+                  prefetch it and fire that side effect on hover. */}
+              {/* eslint-disable-next-line @next/next/no-html-link-for-pages */}
               <a
                 href="/api/stripe/connect"
                 className="mt-3 inline-flex items-center justify-center rounded-lg text-xs font-medium bg-amber-600 text-white hover:bg-amber-700 h-8 px-4 transition-colors"

app/actions/admin/revenue.ts

@@ -3,6 +3,8 @@
 import { createClient } from '@/lib/supabase/server'
 import { createAdminClient } from '@/lib/supabase/admin'
 import {getCurrentTenantId, getCurrentUserId } from '@/lib/supabase/tenant'
+import { fetchAllRows } from '@/lib/supabase/fetch-all-rows'
+import { fetchAllRowsIn } from '@/lib/supabase/fetch-all-rows-in'
 
 async function verifyAdminAccess() {
   const supabase = await createClient()
@@ -29,14 +31,26 @@ export async function getRevenueOverview() {
   const { tenantId } = await verifyAdminAccess()
   const adminClient = await createAdminClient()
 
-  // Get all successful transactions for this tenant
-  const { data: transactions } = await adminClient
-    .from('transactions')
-    .select('amount, currency, transaction_date, product_id, plan_id, stripe_payment_intent_id')
-    .eq('tenant_id', tenantId)
-    .eq('status', 'successful')
-
-  if (!transactions || transactions.length === 0) {
+  // Get all successful transactions for this tenant. Paged and count-verified
+  // (#548): everything this action returns — total revenue, platform fees,
+  // per-product breakdown, the monthly trend — is a sum over this list, so a
+  // read truncated at the API row cap would report a confidently wrong number
+  // instead of an error. Ordered by the primary key so the paging windows
+  // neither overlap nor skip.
+  const transactions = await fetchAllRows('transactions', (from, to) =>
+    adminClient
+      .from('transactions')
+      .select(
+        'amount, currency, transaction_date, product_id, plan_id, stripe_payment_intent_id',
+        { count: 'exact' }
+      )
+      .eq('tenant_id', tenantId)
+      .eq('status', 'successful')
+      .order('transaction_id')
+      .range(from, to)
+  )
+
+  if (transactions.length === 0) {
     return {
       totalRevenue: 0,
       platformFees: 0,
@@ -79,16 +93,20 @@ export async function getRevenueOverview() {
     revenueByProduct[key] = (revenueByProduct[key] || 0) + Number(tx.amount)
   }
 
-  // Get product names
+  // Get product names. The id list is as long as the transaction set is
+  // varied, so it is chunked as well as paged (#548) — an over-long `.in()`
+  // fails on the request side, before any of the response paging matters.
   const productIds = [...new Set(transactions.map(t => t.product_id).filter(Boolean))]
-  const { data: products } = productIds.length > 0
-    ? await adminClient
-        .from('products')
-        .select('product_id, name')
-        .in('product_id', productIds)
-    : { data: [] }
-
-  const productMap = new Map((products || []).map(p => [p.product_id, p.name]))
+  const products = await fetchAllRowsIn('products', productIds, (chunk, from, to) =>
+    adminClient
+      .from('products')
+      .select('product_id, name', { count: 'exact' })
+      .in('product_id', chunk)
+      .order('product_id')
+      .range(from, to)
+  )
+
+  const productMap = new Map(products.map(p => [p.product_id, p.name]))
 
   const revenueByCourse = Object.entries(revenueByProduct).map(([id, amount]) => ({
     id: Number(id),

app/actions/platform/billing-health.ts

@@ -3,6 +3,8 @@
 import { createAdminClient } from '@/lib/supabase/admin'
 import { isSuperAdmin } from '@/lib/supabase/get-user-role'
 import { getCurrentUserId } from '@/lib/supabase/tenant'
+import { fetchAllRows } from '@/lib/supabase/fetch-all-rows'
+import { fetchAllRowsIn } from '@/lib/supabase/fetch-all-rows-in'
 import {
   computeBillingHealth,
   mergeAtRiskTenants,
@@ -85,19 +87,42 @@ export async function getAtRiskTenants(): Promise<AtRiskTenant[]> {
   await verifySuperAdmin()
   const admin = createAdminClient()
 
-  const [
-    { data: pastDueTenants },
-    { data: pastDueSubscriptions },
-    { data: cutoffTenants },
-  ] = await Promise.all([
-    admin.from('tenants').select(TENANT_SELECT).eq('billing_status', 'past_due'),
-    admin.from('platform_subscriptions').select(SUBSCRIPTION_SELECT).eq('status', 'past_due'),
-    admin.from('tenants').select(TENANT_SELECT).not('access_cutoff_at', 'is', null),
+  // All five reads are paged and count-verified (#548). This dashboard exists
+  // to make sure no at-risk school goes unnoticed, so a read silently capped
+  // at the API row limit defeats its entire purpose: the missing school looks
+  // exactly like a healthy one. Each is ordered by its primary key — `tenants`
+  // and `platform_subscriptions` have no natural sort here, and an unordered
+  // `.range()` window is not a stable page.
+  const [pastDueTenants, pastDueSubscriptions, cutoffTenants] = await Promise.all([
+    fetchAllRows('tenants:past_due', (from, to) =>
+      admin
+        .from('tenants')
+        .select(TENANT_SELECT, { count: 'exact' })
+        .eq('billing_status', 'past_due')
+        .order('id')
+        .range(from, to)
+    ),
+    fetchAllRows('platform_subscriptions:past_due', (from, to) =>
+      admin
+        .from('platform_subscriptions')
+        .select(SUBSCRIPTION_SELECT, { count: 'exact' })
+        .eq('status', 'past_due')
+        .order('subscription_id')
+        .range(from, to)
+    ),
+    fetchAllRows('tenants:access_cutoff', (from, to) =>
+      admin
+        .from('tenants')
+        .select(TENANT_SELECT, { count: 'exact' })
+        .not('access_cutoff_at', 'is', null)
+        .order('id')
+        .range(from, to)
+    ),
   ])
 
-  const pastDueTenantRows = (pastDueTenants ?? []).map(toTenantRow)
-  const cutoffTenantRows = (cutoffTenants ?? []).map(toTenantRow)
-  const pastDueSubscriptionRows = (pastDueSubscriptions ?? []).map(toSubscriptionInput)
+  const pastDueTenantRows = pastDueTenants.map(toTenantRow)
+  const cutoffTenantRows = cutoffTenants.map(toTenantRow)
+  const pastDueSubscriptionRows = pastDueSubscriptions.map(toSubscriptionInput)
 
   const knownTenantIds = new Set([
     ...pastDueTenantRows.map((t) => t.tenantId),
@@ -109,26 +134,31 @@ export async function getAtRiskTenants(): Promise<AtRiskTenant[]> {
   // Subscription-only past-due tenants appear in neither read above.
   const missingTenantIds = subscriptionPastDueTenantIds.filter((id) => !knownTenantIds.has(id))
 
-  const [{ data: extraTenants }, { data: knownSubscriptions }] = await Promise.all([
-    missingTenantIds.length
-      ? admin.from('tenants').select(TENANT_SELECT).in('id', missingTenantIds)
-      : Promise.resolve({ data: [] as never[] }),
-    knownTenantIds.size
-      ? admin
-          .from('platform_subscriptions')
-          .select(SUBSCRIPTION_SELECT)
-          .in('tenant_id', [...knownTenantIds])
-      : Promise.resolve({ data: [] as never[] }),
+  // Both follow-ups look up by id list, which grows with the reads above — so
+  // they are chunked as well as paged: past a few hundred ids the `.in()` URL
+  // is the thing that breaks, before any row cap is reached.
+  const [extraTenants, knownSubscriptions] = await Promise.all([
+    fetchAllRowsIn('tenants:subscription_only', missingTenantIds, (chunk, from, to) =>
+      admin.from('tenants').select(TENANT_SELECT, { count: 'exact' }).in('id', chunk).order('id').range(from, to)
+    ),
+    fetchAllRowsIn('platform_subscriptions:known', [...knownTenantIds], (chunk, from, to) =>
+      admin
+        .from('platform_subscriptions')
+        .select(SUBSCRIPTION_SELECT, { count: 'exact' })
+        .in('tenant_id', chunk)
+        .order('subscription_id')
+        .range(from, to)
+    ),
   ])
 
   return computeBillingHealth(
     mergeAtRiskTenants({
       pastDueTenants: pastDueTenantRows,
       cutoffTenants: cutoffTenantRows,
       subscriptionPastDueTenantIds,
-      extraTenants: (extraTenants ?? []).map(toTenantRow),
+      extraTenants: extraTenants.map(toTenantRow),
     }),
-    [...pastDueSubscriptionRows, ...(knownSubscriptions ?? []).map(toSubscriptionInput)],
+    [...pastDueSubscriptionRows, ...knownSubscriptions.map(toSubscriptionInput)],
     new Date(),
   )
 }

app/api/cron/solana-pull/route.ts

@@ -20,6 +20,7 @@ import { timingSafeEqual } from 'crypto'
 import { getSubscriptionState } from '@/lib/payments/solana-subscriptions'
 import { pullSplitForSubscription } from '@/lib/payments/solana-subscription-pull'
 import { decidePullAction } from '@/lib/payments/solana-pull-decision'
+import { fetchAllRows } from '@/lib/supabase/fetch-all-rows'
 
 export const runtime = 'nodejs'
 
@@ -43,6 +44,24 @@ interface SolanaSubMeta {
   mint: string
 }
 
+/** One row of the crank's work queue, as selected below. */
+interface SolanaSubRow {
+  subscription_id: string
+  tenant_id: string
+  plan_id: number | null
+  provider_metadata: unknown
+  subscription_status: string | null
+  cancel_at_period_end: boolean | null
+  cancel_at: string | null
+  /**
+   * The embedded `plans(price)`. Left `unknown`: the untyped service-role
+   * client types every embed as an array, while a to-one embed arrives as an
+   * object at runtime — the read site below narrows it with the same cast it
+   * always used.
+   */
+  plans: unknown
+}
+
 export async function GET(req: NextRequest) {
   const cronSecret = process.env.CRON_SECRET
   const provided = req.headers.get('authorization')?.replace('Bearer ', '')
@@ -63,17 +82,32 @@ export async function GET(req: NextRequest) {
   // Active native-subscription rows, with their on-chain coordinates + amount.
   // We also pull the cancel fields so the pull/cancel decision (which must never
   // charge a canceled sub — issue #460) has the full DB state.
-  const { data: subs, error } = await admin
-    .from('subscriptions')
-    .select('subscription_id, tenant_id, plan_id, provider_metadata, subscription_status, cancel_at_period_end, cancel_at, plans(price)')
-    .eq('payment_provider', 'solana_subs')
-    .eq('subscription_status', 'active')
-
-  if (error) {
-    console.error('[solana-pull] query error', error)
+  //
+  // Paged and count-verified (#548). This read IS the crank's work queue, and
+  // the crank is the only thing that ever charges a native Solana
+  // subscription: a row silently dropped at the API cap is a period that never
+  // gets billed, for a subscriber whose access we keep extending. Ordered by
+  // primary key so the pages are stable. `fetchAllRows` throws on an
+  // incomplete read, so a short queue can no longer look like a finished one.
+  let subs: SolanaSubRow[]
+  try {
+    subs = await fetchAllRows<SolanaSubRow>('subscriptions', (from, to) =>
+      admin
+        .from('subscriptions')
+        .select(
+          'subscription_id, tenant_id, plan_id, provider_metadata, subscription_status, cancel_at_period_end, cancel_at, plans(price)',
+          { count: 'exact' },
+        )
+        .eq('payment_provider', 'solana_subs')
+        .eq('subscription_status', 'active')
+        .order('subscription_id')
+        .range(from, to),
+    )
+  } catch (err) {
+    console.error('[solana-pull] query error', err)
     return NextResponse.json({ error: 'Query failed' }, { status: 500 })
   }
-  if (!subs?.length) return NextResponse.json({ pulled: 0 })
+  if (!subs.length) return NextResponse.json({ pulled: 0 })
 
   const nowSec = Math.floor(Date.now() / 1000)
   let pulled = 0