fix(sync): stop expired trials from starving automatic bank and skattekonto sync (#1376)

The bank sync and skattekonto sync crons fetched the 50 oldest
connection/token rows and only then checked entitlements per item, so
expired-trial rows permanently occupied every batch slot and entitled
companies were never synced automatically.

Fetch all candidate rows, resolve capability grants in bulk via the new
getCompanyIdsWithCapability() (company and firm grants cascade, expired
grants excluded, explicit per-company disable wins), and apply the
50-item run cap after filtering. Entitlement query failures now fail the
run instead of silently skipping every company.

Fixes #563

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

app/api/extensions/enable-banking/sync/cron/__tests__/route.test.ts

@@ -25,7 +25,7 @@ const mocks = vi.hoisted(() => ({
   createClient: vi.fn(),
   probeSessionHealth: vi.fn(),
   syncAccountTransactions: vi.fn(),
-  hasCapability: vi.fn(),
+  getCompanyIdsWithCapability: vi.fn(),
   runReconciliation: vi.fn(),
 }))
 
@@ -46,7 +46,7 @@ vi.mock('@/extensions/general/enable-banking/lib/sync', () => ({
 }))
 
 vi.mock('@/lib/entitlements/has-capability', () => ({
-  hasCapability: (...args: unknown[]) => mocks.hasCapability(...args),
+  getCompanyIdsWithCapability: (...args: unknown[]) => mocks.getCompanyIdsWithCapability(...args),
 }))
 
 vi.mock('@/lib/reconciliation/bank-reconciliation', () => ({
@@ -101,7 +101,7 @@ function makeClient(state: ClientState) {
       }
 
       const chain: Record<string, unknown> = {}
-      const passthrough = ['select', 'not', 'lt', 'gte', 'order', 'limit']
+      const passthrough = ['select', 'not', 'lt', 'gte', 'order', 'limit', 'range']
       for (const method of passthrough) chain[method] = vi.fn(() => chain)
       chain.eq = vi.fn((col: string, value: unknown) => {
         filters[col] = value
@@ -156,7 +156,9 @@ beforeEach(() => {
   process.env.SUPABASE_SERVICE_ROLE_KEY = 'service-key'
   state = { active: [], probeCandidates: [], updates: [] }
   mocks.createClient.mockImplementation(() => makeClient(state))
-  mocks.hasCapability.mockResolvedValue(true)
+  mocks.getCompanyIdsWithCapability.mockImplementation(
+    async (_supabase: unknown, companyIds: string[]) => new Set(companyIds),
+  )
   mocks.syncAccountTransactions.mockResolvedValue({ imported: 0, duplicates: 0, errors: 0 })
   mocks.probeSessionHealth.mockResolvedValue('unknown')
 })
@@ -256,7 +258,7 @@ describe('GET /api/extensions/enable-banking/sync/cron: session health probe', (
     // The silent skip that let a dead connection sit at 'active' for days.
     state.active = [connection()]
     state.probeCandidates = [connection()]
-    mocks.hasCapability.mockResolvedValue(false)
+    mocks.getCompanyIdsWithCapability.mockResolvedValue(new Set())
     mocks.probeSessionHealth.mockResolvedValue('dead')
 
     await GET(cronRequest())
@@ -266,6 +268,37 @@ describe('GET /api/extensions/enable-banking/sync/cron: session health probe', (
     expect(state.updates[0].payload).toMatchObject({ status: 'expired' })
   })
 
+  it('selects an entitled connection after fifty ineligible queue rows', async () => {
+    state.active = [
+      ...Array.from({ length: 50 }, (_, index) => connection({
+        id: `free-${index}`,
+        company_id: `00000000-0000-4000-8000-${String(index).padStart(12, '0')}`,
+      })),
+      connection({ id: 'paid-connection', company_id: '11111111-1111-4111-8111-111111111111' }),
+    ]
+    mocks.getCompanyIdsWithCapability.mockResolvedValue(
+      new Set(['11111111-1111-4111-8111-111111111111']),
+    )
+
+    const response = await GET(cronRequest())
+
+    expect(response.status).toBe(200)
+    expect(mocks.syncAccountTransactions).toHaveBeenCalledTimes(1)
+    expect(mocks.syncAccountTransactions.mock.calls[0][3]).toBe('paid-connection')
+    await expect(response.json()).resolves.toMatchObject({ processed: 1 })
+  })
+
+  it('applies the fifty-connection cap after entitlement filtering', async () => {
+    state.active = Array.from({ length: 51 }, (_, index) => connection({
+      id: `paid-${index}`,
+      company_id: `11111111-1111-4111-8111-${String(index).padStart(12, '0')}`,
+    }))
+
+    await GET(cronRequest())
+
+    expect(mocks.syncAccountTransactions).toHaveBeenCalledTimes(50)
+  })
+
   it('probes a connection whose accounts are all deselected', async () => {
     // This branch reports 'synced' without writing last_synced_at, so the row
     // looks fresh forever.

app/api/extensions/enable-banking/sync/cron/route.ts

@@ -20,15 +20,18 @@ import {
   generateConsentExpiryEmailSubject,
 } from '@/lib/email/consent-notification-templates'
 import { ensureInitialized } from '@/lib/init'
-import { hasCapability } from '@/lib/entitlements/has-capability'
+import { getCompanyIdsWithCapability } from '@/lib/entitlements/has-capability'
 import { CAPABILITY } from '@/lib/entitlements/keys'
 import { withCronContext } from '@/lib/api/with-cron-context'
 import { errorResponse, errorResponseFromCode } from '@/lib/errors/get-structured-error'
 import { getBranding } from '@/lib/branding/service'
+import { fetchAllRows } from '@/lib/supabase/fetch-all'
 import type { StoredAccount } from '@/extensions/general/enable-banking/types'
 
 ensureInitialized()
 
+const MAX_CONNECTIONS_PER_RUN = 50
+
 /**
  * GET /api/extensions/enable-banking/sync/cron
  * Automatic daily bank transaction sync
@@ -64,21 +67,42 @@ export const GET = withCronContext('cron.bank_sync', async (_request, ctx) => {
     ctx.log.info('cleaned up stale pending connections', { count: stalePending.length })
   }
 
-  const { data: connections, error: connError } = await supabase
-    .from('bank_connections')
-    .select('*')
-    .eq('status', 'active')
-    .order('last_synced_at', { ascending: true, nullsFirst: true })
-    .limit(50)
-
-  if (connError) {
-    ctx.log.error('failed to fetch bank connections', connError, {
-      message: connError.message,
-      code: connError.code,
-    })
-    return errorResponse(connError, ctx.log, { requestId: ctx.requestId })
+  let candidateConnections
+  let entitledCompanyIds
+  try {
+    candidateConnections = await fetchAllRows(
+      ({ from, to }) => supabase
+        .from('bank_connections')
+        .select('*')
+        .eq('status', 'active')
+        .order('last_synced_at', { ascending: true, nullsFirst: true })
+        .order('id', { ascending: true })
+        .range(from, to),
+      { dedupeBy: connection => connection.id },
+    )
+    entitledCompanyIds = await getCompanyIdsWithCapability(
+      supabase,
+      candidateConnections.map(connection => connection.company_id),
+      CAPABILITY.bank_sync,
+    )
+  } catch (error) {
+    ctx.log.error('failed to build entitled bank sync work list', error as Error)
+    return errorResponse(error, ctx.log, { requestId: ctx.requestId })
   }
 
+  // Apply the batch limit only after entitlement filtering. Otherwise old
+  // free-tier rows can permanently occupy the first 50 queue positions and
+  // prevent every paying connection behind them from syncing.
+  const connections = candidateConnections
+    .filter(connection => entitledCompanyIds.has(connection.company_id))
+    .slice(0, MAX_CONNECTIONS_PER_RUN)
+
+  ctx.log.info('bank sync work list built', {
+    candidates: candidateConnections.length,
+    entitledCompanies: entitledCompanyIds.size,
+    selected: connections.length,
+  })
+
   // No early return on an empty set: the health probe below still has work to
   // do (a company whose only connection is parked in 'pending_selection' has
   // nothing to sync but can absolutely have a dead session).
@@ -107,17 +131,12 @@ export const GET = withCronContext('cron.bank_sync', async (_request, ctx) => {
   const notifyKey = (c: { user_id: string; session_id: string | null }) =>
     `${c.user_id}:${c.session_id ?? 'none'}`
 
-  for (const connection of connections ?? []) {
+  for (const connection of connections) {
     if (Date.now() - startTime > TIME_BUDGET_MS) {
       ctx.log.info('time budget reached', { processedSoFar: results.length })
       break
     }
 
-    if (!(await hasCapability(supabase, connection.company_id, CAPABILITY.bank_sync))) {
-      ctx.log.info('skip: capability not entitled', { companyId: connection.company_id })
-      continue
-    }
-
     try {
       const daysLeft = getDaysUntilExpiry(connection.consent_expires)
       const isExpired = daysLeft !== null && daysLeft <= 0

app/api/extensions/skatteverket/skattekonto/sync/cron/__tests__/route.test.ts

@@ -0,0 +1,174 @@
+/* eslint-disable @typescript-eslint/no-explicit-any */
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+
+const mocks = vi.hoisted(() => ({
+  createClient: vi.fn(),
+  verifyCronSecret: vi.fn(),
+  getCompanyIdsWithCapability: vi.fn(),
+  createExtensionContext: vi.fn(),
+  syncSkattekonto: vi.fn(),
+  computeSkattekontoDrift: vi.fn(),
+  maybeAlertDrift: vi.fn(),
+}))
+
+vi.mock('@supabase/supabase-js', () => ({
+  createClient: (...args: unknown[]) => mocks.createClient(...args),
+}))
+
+vi.mock('@/lib/init', () => ({ ensureInitialized: vi.fn() }))
+
+vi.mock('@/lib/auth/cron', () => ({
+  verifyCronSecret: (...args: unknown[]) => mocks.verifyCronSecret(...args),
+}))
+
+vi.mock('@/lib/entitlements/has-capability', () => ({
+  getCompanyIdsWithCapability: (...args: unknown[]) => mocks.getCompanyIdsWithCapability(...args),
+}))
+
+vi.mock('@/lib/extensions/context-factory', () => ({
+  createExtensionContext: (...args: unknown[]) => mocks.createExtensionContext(...args),
+}))
+
+vi.mock('@/extensions/general/skatteverket/lib/skattekonto-sync', () => ({
+  SKATTEKONTO_LAST_SYNCED_AT_KEY: 'skattekonto_last_synced_at',
+  syncSkattekonto: (...args: unknown[]) => mocks.syncSkattekonto(...args),
+}))
+
+vi.mock('@/extensions/general/skatteverket/lib/skattekonto-drift', () => ({
+  computeSkattekontoDrift: (...args: unknown[]) => mocks.computeSkattekontoDrift(...args),
+  maybeAlertDrift: (...args: unknown[]) => mocks.maybeAlertDrift(...args),
+}))
+
+vi.mock('@/extensions/general/skatteverket/lib/api-client', () => {
+  class SkatteverketAuthError extends Error {
+    constructor(
+      message: string,
+      public readonly code: string,
+    ) {
+      super(message)
+    }
+  }
+  return { SkatteverketAuthError }
+})
+
+vi.mock('@/extensions/general/skatteverket/lib/skattekonto-client', () => {
+  class SkatteverketSkattekontoError extends Error {
+    felkod = 'TEST'
+  }
+  return { SkatteverketSkattekontoError }
+})
+
+vi.mock('@/extensions/general/skatteverket/lib/token-store', () => ({
+  RECONSENT_ERROR_CODES: [] as const,
+  markNeedsReconsent: vi.fn(),
+}))
+
+vi.mock('@/extensions/general/skatteverket/lib/system-auth/config', () => ({
+  getSystemAuthMode: vi.fn(() => 'off'),
+  isSystemAuthConfigured: vi.fn(() => false),
+}))
+
+vi.mock('@/extensions/general/skatteverket/lib/connection-store', () => ({
+  listVerifiedCompanies: vi.fn().mockResolvedValue([]),
+  markGrantRevoked: vi.fn(),
+}))
+
+vi.mock('@/extensions/general/skatteverket/lib/resolve-auth', () => ({
+  currentSkvEnvironment: vi.fn(() => 'test'),
+  hasVerifiedGrant: vi.fn().mockResolvedValue(false),
+}))
+
+vi.mock('@/lib/errors/get-error-message', () => ({
+  getErrorMessage: vi.fn(() => 'Något gick fel. Försök igen.'),
+}))
+
+import { GET } from '../route'
+
+function makeRequest(): Request {
+  return new Request('http://localhost/api/extensions/skatteverket/skattekonto/sync/cron')
+}
+
+function makeSupabaseStub(tokens: Record<string, unknown>[]) {
+  return {
+    from: vi.fn((table: string) => {
+      const resolved = table === 'skatteverket_tokens'
+        ? { data: tokens, error: null }
+        : { data: null, error: null }
+      const chain: any = {}
+      for (const method of ['select', 'eq', 'order', 'range']) {
+        chain[method] = vi.fn(() => chain)
+      }
+      chain.maybeSingle = vi.fn().mockResolvedValue(resolved)
+      chain.then = (resolve: (value: unknown) => void) => resolve(resolved)
+      return chain
+    }),
+  }
+}
+
+describe('GET /api/extensions/skatteverket/skattekonto/sync/cron', () => {
+  let errorSpy: ReturnType<typeof vi.spyOn>
+  let infoSpy: ReturnType<typeof vi.spyOn>
+  let logSpy: ReturnType<typeof vi.spyOn>
+
+  beforeEach(() => {
+    vi.clearAllMocks()
+    process.env.SKATTEVERKET_ENABLED = 'true'
+    process.env.NEXT_PUBLIC_SUPABASE_URL = 'https://test.supabase.co'
+    process.env.SUPABASE_SERVICE_ROLE_KEY = 'service-key'
+    mocks.verifyCronSecret.mockReturnValue(null)
+    mocks.createExtensionContext.mockImplementation(
+      (supabase: unknown, userId: string, companyId: string) => ({ supabase, userId, companyId }),
+    )
+    mocks.syncSkattekonto.mockResolvedValue({ booked: 0, upcoming: 0 })
+    mocks.computeSkattekontoDrift.mockResolvedValue(null)
+    errorSpy = vi.spyOn(console, 'error').mockImplementation(() => {})
+    infoSpy = vi.spyOn(console, 'info').mockImplementation(() => {})
+    logSpy = vi.spyOn(console, 'log').mockImplementation(() => {})
+  })
+
+  afterEach(() => {
+    errorSpy.mockRestore()
+    infoSpy.mockRestore()
+    logSpy.mockRestore()
+    vi.unstubAllEnvs()
+  })
+
+  it('returns 401 before creating a database client when cron auth fails', async () => {
+    mocks.verifyCronSecret.mockReturnValueOnce(
+      new Response(JSON.stringify({ error: 'Unauthorized' }), { status: 401 }),
+    )
+
+    const response = await GET(makeRequest())
+
+    expect(response.status).toBe(401)
+    expect(mocks.createClient).not.toHaveBeenCalled()
+  })
+
+  it('syncs an entitled company after fifty ineligible token rows', async () => {
+    const entitledCompanyId = '11111111-1111-4111-8111-111111111111'
+    const tokens = [
+      ...Array.from({ length: 50 }, (_, index) => ({
+        user_id: `00000000-0000-4000-8000-${String(index).padStart(12, '0')}`,
+        company_id: `22222222-2222-4222-8222-${String(index).padStart(12, '0')}`,
+        expires_at: `2026-01-${String((index % 28) + 1).padStart(2, '0')}T00:00:00Z`,
+        refresh_count: 0,
+      })),
+      {
+        user_id: '33333333-3333-4333-8333-333333333333',
+        company_id: entitledCompanyId,
+        expires_at: '2099-01-01T00:00:00Z',
+        refresh_count: 0,
+      },
+    ]
+    mocks.createClient.mockReturnValue(makeSupabaseStub(tokens))
+    mocks.getCompanyIdsWithCapability.mockResolvedValue(new Set([entitledCompanyId]))
+
+    const response = await GET(makeRequest())
+    const body = await response.json()
+
+    expect(response.status).toBe(200)
+    expect(body).toMatchObject({ processed: 1, synced: 1, errors: 0 })
+    expect(mocks.syncSkattekonto).toHaveBeenCalledTimes(1)
+    expect(mocks.syncSkattekonto.mock.calls[0][0]).toMatchObject({ companyId: entitledCompanyId })
+  })
+})

app/api/extensions/skatteverket/skattekonto/sync/cron/route.ts

@@ -2,7 +2,7 @@ import { createClient } from '@supabase/supabase-js'
 import { NextResponse } from 'next/server'
 import { ensureInitialized } from '@/lib/init'
 import { verifyCronSecret } from '@/lib/auth/cron'
-import { hasCapability } from '@/lib/entitlements/has-capability'
+import { getCompanyIdsWithCapability } from '@/lib/entitlements/has-capability'
 import { CAPABILITY } from '@/lib/entitlements/keys'
 import { createExtensionContext } from '@/lib/extensions/context-factory'
 import { syncSkattekonto, SKATTEKONTO_LAST_SYNCED_AT_KEY } from '@/extensions/general/skatteverket/lib/skattekonto-sync'
@@ -14,11 +14,14 @@ import { getSystemAuthMode, isSystemAuthConfigured } from '@/extensions/general/
 import { listVerifiedCompanies, markGrantRevoked } from '@/extensions/general/skatteverket/lib/connection-store'
 import { getErrorMessage } from '@/lib/errors/get-error-message'
 import { currentSkvEnvironment, hasVerifiedGrant } from '@/extensions/general/skatteverket/lib/resolve-auth'
+import { fetchAllRows } from '@/lib/supabase/fetch-all'
 
 ensureInitialized()
 
 export const maxDuration = 60
 
+const MAX_COMPANIES_PER_RUN = 50
+
 /**
  * GET /api/extensions/skatteverket/skattekonto/sync/cron
  *
@@ -69,17 +72,21 @@ export async function GET(request: Request) {
   // company_id (multi-tenant refactor). Rows flagged needs_reconsent are
   // excluded: SKV's per-flow refresh tokens live 65 minutes, so a connection
   // that failed with a terminal auth error can never heal on its own.
-  const { data: tokens, error: tokensError } = await supabase
-    .from('skatteverket_tokens')
-    .select('user_id, company_id, expires_at, refresh_count')
-    .eq('status', 'active')
-    .order('expires_at', { ascending: true })
-    .limit(50)
-
-  if (tokensError) {
+  let tokens
+  try {
+    tokens = await fetchAllRows(
+      ({ from, to }) => supabase
+        .from('skatteverket_tokens')
+        .select('user_id, company_id, expires_at, refresh_count')
+        .eq('status', 'active')
+        .order('expires_at', { ascending: true })
+        .order('user_id', { ascending: true })
+        .range(from, to),
+      { dedupeBy: token => token.user_id },
+    )
+  } catch (error) {
     console.error('[skattekonto-sync-cron] Failed to fetch tokens', {
-      message: tokensError.message,
-      code: tokensError.code,
+      message: error instanceof Error ? error.message : String(error),
     })
     return NextResponse.json({ error: 'Failed to fetch tokens' }, { status: 500 })
   }
@@ -91,7 +98,7 @@ export async function GET(request: Request) {
 
   type WorkItem = { companyId: string; userId: string; source: 'system' | 'user' }
   const tokenByCompany = new Map<string, string>()
-  for (const token of tokens ?? []) {
+  for (const token of tokens) {
     if (token.company_id) tokenByCompany.set(token.company_id as string, token.user_id as string)
   }
 
@@ -105,7 +112,7 @@ export async function GET(request: Request) {
     systemCompanyIds.add(company.company_id)
     work.push({ companyId: company.company_id, userId, source: 'system' })
   }
-  for (const token of tokens ?? []) {
+  for (const token of tokens) {
     const companyId = token.company_id as string | null
     if (!companyId) {
       console.warn('[skattekonto-sync-cron] token without company_id skipped', {
@@ -121,6 +128,32 @@ export async function GET(request: Request) {
     return NextResponse.json({ message: 'No connected companies', processed: 0 })
   }
 
+  let entitledCompanyIds
+  try {
+    entitledCompanyIds = await getCompanyIdsWithCapability(
+      supabase,
+      work.map(item => item.companyId),
+      CAPABILITY.skatteverket,
+    )
+  } catch (error) {
+    console.error('[skattekonto-sync-cron] Failed to resolve entitled companies', {
+      message: error instanceof Error ? error.message : String(error),
+    })
+    return NextResponse.json({ error: 'Failed to resolve entitlements' }, { status: 500 })
+  }
+
+  // Limit the eligible work list, not the raw token list. Expired trials and
+  // disabled modules must not occupy all 50 positions ahead of paying firms.
+  const entitledWork = work
+    .filter(item => entitledCompanyIds.has(item.companyId))
+    .slice(0, MAX_COMPANIES_PER_RUN)
+
+  console.info('[skattekonto-sync-cron] Work list built', {
+    candidates: work.length,
+    entitledCompanies: entitledCompanyIds.size,
+    selected: entitledWork.length,
+  })
+
   const startTime = Date.now()
   const TIME_BUDGET_MS = 50_000
   const SYNC_COOLDOWN_MS = 60 * 60 * 1000 // 1 hour
@@ -140,7 +173,7 @@ export async function GET(request: Request) {
   // failure, skip the remaining system-mode entries this run.
   let systemAuthFailed = false
 
-  for (const item of work) {
+  for (const item of entitledWork) {
     if (Date.now() - startTime > TIME_BUDGET_MS) {
       console.log(`[skattekonto-sync-cron] Time budget reached after ${results.length} companies`)
       break
@@ -153,11 +186,6 @@ export async function GET(request: Request) {
       continue
     }
 
-    if (!(await hasCapability(supabase, companyId, CAPABILITY.skatteverket))) {
-      console.info('[skattekonto-sync-cron] skip: capability not entitled', { companyId })
-      continue
-    }
-
     try {
       // Cooldown: skip if synced within the last hour.
       const { data: lastSyncRow } = await supabase