feat(connect): Enable Banking proxy for self-hosted instances, with a secret-free ownership ledger and a global rate budget (#1751)

* feat(entitlements): partition the self-host bypass so connector capabilities fall through to grants; capability_grants.source accepts 'connector'

Sovereign plan WS3 PR3: ships dark, nothing changes for hosted.

- lib/entitlements/keys.ts: CONNECTOR_CAPABILITIES = bank_sync,
  skatteverket, org_lookup, migration (services Accounted operates that a
  self-hosted instance cannot provide itself) + isConnectorCapability().
  Separate from PAID_CAPABILITIES and outside the trial-seed trigger on
  purpose: a hosted company can never hold a connector grant.
- lib/entitlements/has-capability.ts: isPaywallBypassed() -> isBypassedFor(key).
  Hosted: byte-identical (dev / DISABLE_PAYWALL bypass, FORCE_PAYWALL wins,
  else the grant lookup). Self-host: local capabilities always on
  (FORCE_PAYWALL included, as the existing test demands); connector
  capabilities behave like hosted, i.e. dev bypass, FORCE_PAYWALL, else the
  grant lookup where the connector sync will write source='connector' rows.
  getCompanyEntitlements on a self-host: local paid keys + active connector
  keys, state 'paid' with an active connector grant else 'none' (never the
  hosted trial copy).
- Migration 20260820122000: capability_grants.source CHECK gains
  'connector', found through pg_constraint (the CHECK was declared inline
  and auto-named; Postgres stores IN as = ANY, matched accordingly).
  pg-real test: connector accepted, unknown source rejected, upsert on the
  (scope, key, source) identity, trial seed writes no connector rows.
- Tests: self-hosted connector matrix (local all-on without DB, connector
  gated by grant/expiry, dev bypass all-on, FORCE_PAYWALL gates connector
  keys only, bulk resolution, entitlements shape); two pre-existing tests
  that asserted the old "self-host holds connector keys" contract updated
  to the new one.

Verified: full unit suite green, pg-real suite for lib/entitlements green
against a local supabase/postgres with every migration applied, lint
ratchet, guards. Deferred to the instance-wiring PR: adding the connector
extensions to the self-host Docker preset (dead-end upsells until a key can
be issued).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* refactor(entitlements): fold the self-host branch into the existing grants query

One .or(scopeFilter), not two: the duplicated helper pushed the
no-phantom-columns unresolvable-expression count to 380/379. Behaviour is
unchanged; the self-host matrix tests still pass.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* feat(connect): hosted connector-key registry + validate RPC + entitlements endpoint; instance sync writes connector grants hourly

Sovereign plan WS3 PR4 ("key infra enabling manual sales"), stacked on the
entitlement partition (#1747). Nothing is purchasable yet; this is the
plumbing both ends need before the first manually issued key.

Hosted side:
- Migration 20260820123000: connector_keys (SHA-256 key_hash, prefix,
  org_number, pinned instance_url, scopes, status, Stripe ids,
  current_period_end, per-minute rate limit, active_company_count,
  last_seen/synced) and connector_usage_events (per-request metering,
  separate from metered_events whose company_id references hosted
  companies). RLS on, NO policies: service role only. RPC
  validate_and_increment_connector_key copies the api_keys pattern (FOR
  UPDATE, minute window, suspended reported not counted, revoked = no row)
  and is REVOKEd from PUBLIC/anon/authenticated, GRANTed to service_role.
  pg-real test covers validate/count, unknown+revoked, suspended, rate
  limit, execute privileges per role, RLS invisibility, usage cascade.
- lib/connect/contract.ts (shared wire types), lib/connect/hosted/keys.ts
  (generate/hash/validate -> 401/403/429 mapping),
  with-connector-auth.ts (Bearer or X-Connector-Key, one usage row per
  request, 500 envelope on handler throw), /api/connect/entitlements GET +
  POST (records active_company_count, pins instance_url on first report,
  never moves a pinned one), scripts/issue-connector-key.ts (dry run unless
  --confirm, prints the key once + the .env lines).

Instance side:
- lib/connect/instance/config.ts (GNUBOK_CONNECTOR_KEY, GNUBOK_CONNECT_URL
  default https://app.gnubok.se), sync.ts: reports the active company count
  and writes source='connector' grants for every company x covered scope,
  expires_at = min(now+72h, period_end+3d); 401/403 or a non-active status
  deletes them (freeze-and-retain); network/5xx/429 leave them alone.
  /api/connector/sync/cron (hourly) runs it; not_configured without a key.
- Crontab generator gains EXTRA_JOBS (variant-only jobs not in vercel.json,
  with reasons) + drift tests; docker/crontab.self-hosted regenerated with
  the hourly sync. Docs (SELF-HOSTING connector section, env templates),
  DECISIONS.

Tests: 52 new unit tests (keys, auth wrapper, route, config, sync outcomes
and grant arithmetic, cron route, crontab EXTRA_JOBS) + 7 pg-real tests
run locally against supabase/postgres with every migration applied.
no-phantom-columns ceiling +1 with a reason (the bulk grant upsert).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* feat(connect): Enable Banking proxy for self-hosted instances, with a secret-free ownership ledger and a global rate budget

Sovereign plan WS3 PR5a, stacked on the connector-key infra (#1748). A
self-hosted instance with a `bank_sync`-scoped connector key can now connect
a bank through Arcim's PSD2 credentials; the bank session id and all
transaction data stay in the instance's own database (founder decision:
tokens on the instance, proxy stateless).

- Migration 20260820124000: `connector_connections` (secret-free ledger:
  sha256 of the EB session id + account uids, service-role only),
  `connector_upstream_counters` + RPC `connector_reserve_upstream` (global
  budget under EB Annex 1 §5's 300/min, shared with hosted), and
  `connector_keys.limits` jsonb; validate RPC v2 returns limits. All RPCs
  REVOKEd from PUBLIC/anon/authenticated, GRANTed service_role. pg-real
  covers all of it.
- EB JWT minting moved to lib/connect/upstreams/enable-banking-jwt.ts (core
  must not import @/extensions/); the extension re-exports it, tests
  unchanged.
- lib/connect/hosted/{state,ledger,upstream-budget}.ts: HMAC-signed
  connector state (15-min TTL) so the consent redirect can use OUR
  registered EB callback and bounce back to the instance, no per-instance
  redirect URI at EB; the callback route gains that connector branch.
- app/api/connect/bank/[...path]: path allowlist (aspsps, auth, sessions,
  accounts/{uid}/{balances,transactions}), never open passthrough. POST
  /auth enforces the per-company connection quota + rewrites redirect/state;
  reads/deletes verify ledger ownership; every upstream call takes the
  global budget (429 + Retry-After when exhausted).
- issue-connector-key.ts: scopes default bank_sync,skatteverket (TIC out of
  v1), --bank/skv-connections-per-company + --sync-min-interval.
- Docs (SELF-HOSTING: bank connector live), DECISIONS.

Verified: 52 connect unit tests + 13 pg-real (run locally against
supabase/postgres with all migrations) + EB extension suite (225, jwt
relocation intact); full unit suite 15 979 green; tsc, guards, lint clean.
Not in this PR: SKV broker (PR5b) and instance wiring (PR6).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* chore(connect): update ledger pg test to re-versioned migration 20260831200000

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UzNkSsR18pLFitJdYn8QEb

* fix(connect): redact opaque path segments before usage metering; correct stale RPC-source comment

GET/DELETE /sessions/{id} and /accounts/{uid}/... carry the raw EB
session id / account uid in the pathname; metering persisted it in
cleartext next to the ledger that stores only sha256(handle). Opaque
segments (UUID, long hex, long base64url) now become ':id' before the
connector_usage_events insert. Migration comment now cites the real
prior RPC source (20260831190000).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UzNkSsR18pLFitJdYn8QEb

* fix(connect): percent-encoded path segments count as opaque in metering redaction

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UzNkSsR18pLFitJdYn8QEb

* fix(connect): PR #1751 review batch: https-only EB URL, body-covering timeout, quota reservation, delete-after-success, doc fix

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UzNkSsR18pLFitJdYn8QEb

* fix(connect): bind the /sessions code exchange to its verified pending state; ceiling +1

Verified state signature, key/service match, and an existing pending
row now precede the EB exchange; a concurrently consumed state closes
the just-minted upstream session and 409s. no-phantom-columns ceiling
391 for countHeldConnections' computed .or() timestamp filter.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UzNkSsR18pLFitJdYn8QEb

---------

Co-authored-by: Jakob Wennberg <311770904+jakobwennberg-oss@users.noreply.github.com>
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
Co-authored-by: Emil <emilmattsson14@gmail.com>

lib/connect/hosted/__tests__/keys.test.ts

@@ -15,6 +15,7 @@ const ROW = {
   status: 'active',
   current_period_end: '2027-01-01T00:00:00.000Z',
   rate_limited: false,
+  limits: { bank_connections_per_company: 2, skv_connections_per_company: 1, sync_min_interval_s: 3600 },
 }
 
 describe('connector key primitives', () => {
@@ -56,10 +57,17 @@ describe('validateConnectorKey', () => {
         scopes: ['bank_sync', 'skatteverket'],
         status: 'active',
         currentPeriodEnd: '2027-01-01T00:00:00.000Z',
+        limits: { bank_connections_per_company: 2, skv_connections_per_company: 1, sync_min_interval_s: 3600 },
       },
     })
   })
 
+  it('fills default limits when the RPC returns null limits', async () => {
+    const { key } = generateConnectorKey()
+    const result = await validateConnectorKey(key, supabaseWithRpc({ data: [{ ...ROW, limits: null }] }).supabase)
+    expect(result.ok && result.key.limits).toEqual({ bank_connections_per_company: 1, skv_connections_per_company: 1, sync_min_interval_s: 0 })
+  })
+
   it('maps no row (unknown/revoked) to 401, but an RPC error to 503', async () => {
     const { key } = generateConnectorKey()
     expect(await validateConnectorKey(key, supabaseWithRpc({ data: [] }).supabase)).toMatchObject({ ok: false, status: 401 })

lib/connect/hosted/__tests__/state.test.ts

@@ -0,0 +1,53 @@
+import { describe, it, expect, afterEach, vi } from 'vitest'
+import { isConnectorState, signConnectorState, verifyConnectorState } from '../state'
+
+afterEach(() => vi.unstubAllEnvs())
+
+const BASE = { kid: 'k1', svc: 'bank' as const, ret: 'https://bokforing.example.se/cb', st: 'inst-state', cref: 'company-1' }
+
+describe('connector state', () => {
+  it('round-trips a signed payload', () => {
+    vi.stubEnv('CONNECTOR_STATE_SECRET', 'secret')
+    const now = 1_000_000
+    const token = signConnectorState(BASE, now)
+    expect(isConnectorState(token)).toBe(true)
+    const v = verifyConnectorState(token, now)
+    expect(v).toEqual({ ok: true, payload: { ...BASE, iat: now } })
+  })
+
+  it('rejects a tampered payload', () => {
+    vi.stubEnv('CONNECTOR_STATE_SECRET', 'secret')
+    const token = signConnectorState(BASE, 1000)
+    const [ver, body, sig] = token.split('.')
+    const tamperedBody = Buffer.from(JSON.stringify({ ...BASE, cref: 'other', iat: 1000 })).toString('base64url')
+    expect(verifyConnectorState(`${ver}.${tamperedBody}.${sig}`, 1000)).toEqual({ ok: false, reason: 'bad_signature' })
+  })
+
+  it('rejects a signature made with a different secret', () => {
+    vi.stubEnv('CONNECTOR_STATE_SECRET', 'secret-a')
+    const token = signConnectorState(BASE, 1000)
+    vi.stubEnv('CONNECTOR_STATE_SECRET', 'secret-b')
+    expect(verifyConnectorState(token, 1000)).toEqual({ ok: false, reason: 'bad_signature' })
+  })
+
+  it('expires after the TTL and rejects a future iat', () => {
+    vi.stubEnv('CONNECTOR_STATE_SECRET', 'secret')
+    const token = signConnectorState(BASE, 1000)
+    expect(verifyConnectorState(token, 1000 + 16 * 60 * 1000)).toEqual({ ok: false, reason: 'expired' })
+    const future = signConnectorState(BASE, 10_000_000)
+    expect(verifyConnectorState(future, 1000)).toEqual({ ok: false, reason: 'expired' })
+  })
+
+  it('flags malformed tokens', () => {
+    vi.stubEnv('CONNECTOR_STATE_SECRET', 'secret')
+    expect(verifyConnectorState('nope')).toEqual({ ok: false, reason: 'malformed' })
+    expect(isConnectorState('random-uuid-state')).toBe(false)
+  })
+
+  it('derives a secret from the service-role key when none is set (still verifiable)', () => {
+    vi.stubEnv('CONNECTOR_STATE_SECRET', '')
+    vi.stubEnv('SUPABASE_SERVICE_ROLE_KEY', 'svc-key')
+    const token = signConnectorState(BASE, 2000)
+    expect(verifyConnectorState(token, 2000).ok).toBe(true)
+  })
+})

lib/connect/hosted/__tests__/upstream-budget.test.ts

@@ -0,0 +1,37 @@
+import { describe, it, expect, vi, afterEach } from 'vitest'
+import type { SupabaseClient } from '@supabase/supabase-js'
+import { budgetFor, reserveUpstream } from '../upstream-budget'
+
+afterEach(() => vi.unstubAllEnvs())
+
+function supa(result: { data?: unknown; error?: unknown }) {
+  const rpc = vi.fn().mockResolvedValue({ data: result.data ?? null, error: result.error ?? null })
+  return { supabase: { rpc } as unknown as SupabaseClient, rpc }
+}
+
+describe('budgetFor', () => {
+  it('defaults sit under the EB per-minute quota and are env-overridable', () => {
+    expect(budgetFor('bank').minuteMax).toBe(90)
+    vi.stubEnv('CONNECT_BANK_RPM_BUDGET', '50')
+    expect(budgetFor('bank').minuteMax).toBe(50)
+  })
+})
+
+describe('reserveUpstream', () => {
+  it('passes the service and the resolved budget to the RPC and returns ok', async () => {
+    const { supabase, rpc } = supa({ data: { ok: true } })
+    expect(await reserveUpstream(supabase, 'bank')).toEqual({ ok: true })
+    expect(rpc).toHaveBeenCalledWith('connector_reserve_upstream', { p_service: 'bank', p_minute_max: 90, p_hour_max: 3000 })
+  })
+
+  it('maps a budget rejection to a Retry-After result', async () => {
+    const { supabase } = supa({ data: { ok: false, scope: 'hour', retry_after_sec: 3600 } })
+    expect(await reserveUpstream(supabase, 'bank')).toEqual({ ok: false, scope: 'hour', retryAfterSec: 3600 })
+  })
+
+  // Fail-open: a broken counter table must not block every connector call.
+  it('fails open on a DB error', async () => {
+    const { supabase } = supa({ error: { message: 'boom' } })
+    expect(await reserveUpstream(supabase, 'skatteverket')).toEqual({ ok: true })
+  })
+})

lib/connect/hosted/__tests__/with-connector-auth.test.ts

@@ -12,7 +12,7 @@ vi.mock('@/lib/auth/api-keys', () => ({
   createServiceClientNoCookies: () => ({ from }),
 }))
 
-import { extractConnectorKey, withConnectorAuth } from '../with-connector-auth'
+import { extractConnectorKey, withConnectorAuth, redactEndpoint } from '../with-connector-auth'
 
 const VALID = {
   ok: true,
@@ -53,6 +53,23 @@ describe('extractConnectorKey', () => {
   })
 })
 
+describe('redactEndpoint', () => {
+  it('replaces opaque handle segments with :id so raw EB session ids never rest in metering', () => {
+    expect(redactEndpoint('/api/connect/bank/sessions/8f14e45f-ceea-467f-a8d5-91be6ce7cbc4')).toBe(
+      '/api/connect/bank/sessions/:id',
+    )
+    expect(redactEndpoint('/api/connect/bank/accounts/9a1b2c3d-0000-4111-8222-333344445555/transactions')).toBe(
+      '/api/connect/bank/accounts/:id/transactions',
+    )
+    expect(redactEndpoint('/api/connect/bank/sessions/deadbeefdeadbeefdeadbeef')).toBe('/api/connect/bank/sessions/:id')
+    expect(redactEndpoint('/api/connect/bank/sessions/8f14e45f%2Dceea-467f-a8d5-91be6ce7cbc4')).toBe(
+      '/api/connect/bank/sessions/:id',
+    )
+    expect(redactEndpoint('/api/connect/entitlements')).toBe('/api/connect/entitlements')
+    expect(redactEndpoint('/api/connect/bank/aspsps')).toBe('/api/connect/bank/aspsps')
+  })
+})
+
 describe('withConnectorAuth', () => {
   const handler = vi.fn(async (_req: Request, _ctx: { key: { id: string } }) => NextResponse.json({ data: 'ok' }))
   const wrapped = withConnectorAuth('connect.entitlements', handler)

lib/connect/hosted/keys.ts

@@ -25,6 +25,18 @@ export function isConnectorKeyFormat(key: string): boolean {
   return key.startsWith(CONNECTOR_KEY_PREFIX) && key.length > CONNECTOR_KEY_PREFIX.length + 16
 }
 
+export interface ConnectorKeyLimits {
+  bank_connections_per_company: number
+  skv_connections_per_company: number
+  sync_min_interval_s: number
+}
+
+export const DEFAULT_CONNECTOR_LIMITS: ConnectorKeyLimits = {
+  bank_connections_per_company: 1,
+  skv_connections_per_company: 1,
+  sync_min_interval_s: 0,
+}
+
 export interface ValidatedConnectorKey {
   id: string
   orgNumber: string
@@ -32,6 +44,7 @@ export interface ValidatedConnectorKey {
   scopes: string[]
   status: ConnectorKeyStatus
   currentPeriodEnd: string | null
+  limits: ConnectorKeyLimits
 }
 
 export type ConnectorKeyValidation =
@@ -80,6 +93,7 @@ export async function validateConnectorKey(
     status: string
     current_period_end: string | null
     rate_limited: boolean
+    limits: Partial<ConnectorKeyLimits> | null
   }
   if (row.status !== 'active') {
     return { ok: false, status: 403, code: 'CONNECTOR_KEY_SUSPENDED', error: 'Connector key is suspended' }
@@ -96,6 +110,7 @@ export async function validateConnectorKey(
       scopes: row.scopes ?? [],
       status: 'active',
       currentPeriodEnd: row.current_period_end,
+      limits: { ...DEFAULT_CONNECTOR_LIMITS, ...(row.limits ?? {}) },
     },
   }
 }

lib/connect/hosted/ledger.ts

@@ -0,0 +1,200 @@
+import crypto from 'node:crypto'
+import type { SupabaseClient } from '@supabase/supabase-js'
+
+/**
+ * The connector connection ledger: proof, without secrets, that a connection
+ * belongs to a given connector key. Every bank/SKV connection is born through
+ * the proxy (the consent redirect is ours), so the proxy records it at
+ * creation and checks ownership on every later use. The upstream handle (EB
+ * session id, SKV access token) is hashed; the value never rests here.
+ */
+
+export type ConnectorService = 'bank' | 'skatteverket'
+
+export function hashHandle(handle: string): string {
+  return crypto.createHash('sha256').update(handle).digest('hex')
+}
+
+export interface LedgerRow {
+  id: string
+  connector_key_id: string
+  service: ConnectorService
+  company_ref: string
+  provider: string | null
+  account_uids: string[]
+  status: 'pending' | 'active' | 'revoked'
+}
+
+/** Active connections for one company under one key and service. Enforces the per-company limit. */
+export async function countActiveConnections(
+  supabase: SupabaseClient,
+  keyId: string,
+  service: ConnectorService,
+  companyRef: string,
+): Promise<number> {
+  const { count, error } = await supabase
+    .from('connector_connections')
+    .select('id', { count: 'exact', head: true })
+    .eq('connector_key_id', keyId)
+    .eq('service', service)
+    .eq('company_ref', companyRef)
+    .eq('status', 'active')
+  if (error) throw new Error(`ledger count failed: ${error.message}`)
+  return count ?? 0
+}
+
+/**
+ * Pending rows count toward quota only while their consent window is open:
+ * the signed connector state expires after 15 minutes, so an abandoned
+ * consent stops reserving capacity once its state can no longer activate it.
+ */
+export const PENDING_QUOTA_WINDOW_MS = 15 * 60 * 1000
+
+/**
+ * Rows currently holding or reserving quota for (key, service, company):
+ * active connections plus fresh pending reservations. Used by the /auth
+ * quota check both before insert (fast reject) and after insert (the
+ * reservation re-count that closes the concurrent-auth race).
+ */
+export async function countHeldConnections(
+  supabase: SupabaseClient,
+  keyId: string,
+  service: ConnectorService,
+  companyRef: string,
+  now: Date = new Date(),
+): Promise<number> {
+  const freshPendingSince = new Date(now.getTime() - PENDING_QUOTA_WINDOW_MS).toISOString()
+  const { count, error } = await supabase
+    .from('connector_connections')
+    .select('id', { count: 'exact', head: true })
+    .eq('connector_key_id', keyId)
+    .eq('service', service)
+    .eq('company_ref', companyRef)
+    .or(`status.eq.active,and(status.eq.pending,created_at.gte.${freshPendingSince})`)
+  if (error) throw new Error(`ledger count failed: ${error.message}`)
+  return count ?? 0
+}
+
+/** Roll back a just-created pending reservation (lost the quota re-count). */
+export async function deletePendingConnectionById(supabase: SupabaseClient, id: string): Promise<void> {
+  await supabase.from('connector_connections').delete().eq('id', id).eq('status', 'pending')
+}
+
+export async function createPendingConnection(
+  supabase: SupabaseClient,
+  params: { keyId: string; service: ConnectorService; companyRef: string; provider: string | null; pendingState: string },
+): Promise<string> {
+  const { data, error } = await supabase
+    .from('connector_connections')
+    .insert({
+      connector_key_id: params.keyId,
+      service: params.service,
+      company_ref: params.companyRef,
+      provider: params.provider,
+      pending_state: params.pendingState,
+      status: 'pending',
+    })
+    .select('id')
+    .single()
+  if (error || !data) throw new Error(`ledger insert failed: ${error?.message}`)
+  return (data as { id: string }).id
+}
+
+/**
+ * The pending row a signed state belongs to, under the presenting key.
+ * Precondition for the code exchange at POST /sessions: exchanging a code
+ * against a state with no pending row would mint an upstream session the
+ * ledger never records (and cross-flow substitution could smuggle a code
+ * into a foreign state). The state's own TTL bounds freshness.
+ */
+export async function findPendingByState(
+  supabase: SupabaseClient,
+  params: { keyId: string; pendingState: string },
+): Promise<LedgerRow | null> {
+  const { data, error } = await supabase
+    .from('connector_connections')
+    .select('id, connector_key_id, service, company_ref, provider, account_uids, status')
+    .eq('connector_key_id', params.keyId)
+    .eq('pending_state', params.pendingState)
+    .eq('status', 'pending')
+    .maybeSingle()
+  if (error) throw new Error(`ledger pending lookup failed: ${error.message}`)
+  return (data as LedgerRow | null) ?? null
+}
+
+/** Activate a pending connection (found by its signed pending_state) with the live handle + accounts. */
+export async function activateByPendingState(
+  supabase: SupabaseClient,
+  params: { keyId: string; pendingState: string; handle: string; accountUids?: string[] },
+): Promise<LedgerRow | null> {
+  const { data, error } = await supabase
+    .from('connector_connections')
+    .update({
+      status: 'active',
+      handle_hash: hashHandle(params.handle),
+      account_uids: params.accountUids ?? [],
+      pending_state: null,
+      activated_at: new Date().toISOString(),
+      last_used_at: new Date().toISOString(),
+    })
+    .eq('connector_key_id', params.keyId)
+    .eq('pending_state', params.pendingState)
+    .eq('status', 'pending')
+    .select('id, connector_key_id, service, company_ref, provider, account_uids, status')
+    .maybeSingle()
+  if (error) throw new Error(`ledger activate failed: ${error.message}`)
+  return (data as LedgerRow | null) ?? null
+}
+
+/** The active ledger row that owns a given handle under a key. Ownership check for reads/writes. */
+export async function findByHandle(
+  supabase: SupabaseClient,
+  params: { keyId: string; service: ConnectorService; handle: string },
+): Promise<LedgerRow | null> {
+  const { data, error } = await supabase
+    .from('connector_connections')
+    .select('id, connector_key_id, service, company_ref, provider, account_uids, status')
+    .eq('connector_key_id', params.keyId)
+    .eq('service', params.service)
+    .eq('handle_hash', hashHandle(params.handle))
+    .eq('status', 'active')
+    .maybeSingle()
+  if (error) throw new Error(`ledger lookup failed: ${error.message}`)
+  return (data as LedgerRow | null) ?? null
+}
+
+/** The active ledger row that owns a bank account uid under a key. */
+export async function findByAccountUid(
+  supabase: SupabaseClient,
+  params: { keyId: string; accountUid: string },
+): Promise<LedgerRow | null> {
+  const { data, error } = await supabase
+    .from('connector_connections')
+    .select('id, connector_key_id, service, company_ref, provider, account_uids, status')
+    .eq('connector_key_id', params.keyId)
+    .eq('service', 'bank')
+    .eq('status', 'active')
+    .contains('account_uids', [params.accountUid])
+    .limit(1)
+    .maybeSingle()
+  if (error) throw new Error(`ledger account lookup failed: ${error.message}`)
+  return (data as LedgerRow | null) ?? null
+}
+
+export async function touchConnection(supabase: SupabaseClient, id: string): Promise<void> {
+  await supabase.from('connector_connections').update({ last_used_at: new Date().toISOString() }).eq('id', id)
+}
+
+/** Revoke by handle (DELETE /sessions). Idempotent. */
+export async function revokeByHandle(
+  supabase: SupabaseClient,
+  params: { keyId: string; service: ConnectorService; handle: string },
+): Promise<void> {
+  await supabase
+    .from('connector_connections')
+    .update({ status: 'revoked', revoked_at: new Date().toISOString() })
+    .eq('connector_key_id', params.keyId)
+    .eq('service', params.service)
+    .eq('handle_hash', hashHandle(params.handle))
+    .eq('status', 'active')
+}

lib/connect/hosted/state.ts

@@ -0,0 +1,88 @@
+import crypto from 'node:crypto'
+
+/**
+ * HMAC-signed connector state for the bank/SKV consent round-trip.
+ *
+ * The instance never registers a redirect URI with Enable Banking or
+ * Skatteverket: the consent redirect goes to OUR hosted callback (already
+ * registered), which then bounces the browser back to the instance. To do
+ * that safely the proxy replaces the upstream `state` with a token that
+ * carries where to return, the original instance state, and which key owns
+ * the flow, all signed so a tampered token is rejected. No storage: the token
+ * is self-contained and short-lived.
+ *
+ * Format: `ck1.<base64url(json)>.<base64url(hmac-sha256)>`.
+ */
+
+const VERSION = 'ck1'
+const DEFAULT_TTL_MS = 15 * 60 * 1000
+
+export interface ConnectorStatePayload {
+  /** connector_key id that owns this flow. */
+  kid: string
+  /** service: 'bank' | 'skv'. */
+  svc: 'bank' | 'skv'
+  /** Absolute return URL on the instance (the instance's own callback). */
+  ret: string
+  /** The instance's original state value, echoed back untouched. */
+  st: string
+  /** Instance company ref, for the ledger. */
+  cref: string
+  /** issued-at, ms. */
+  iat: number
+}
+
+function getSecret(): string {
+  const explicit = process.env.CONNECTOR_STATE_SECRET?.trim()
+  if (explicit) return explicit
+  // Fall back to a value derived from the service-role key so a deployment
+  // that forgot to set the dedicated secret still signs consistently. Never
+  // the raw key: a one-way derivation so the signing secret can't be reversed
+  // into the Supabase credential.
+  const svc = process.env.SUPABASE_SERVICE_ROLE_KEY
+  if (!svc) throw new Error('CONNECTOR_STATE_SECRET (or SUPABASE_SERVICE_ROLE_KEY) is required to sign connector state')
+  return crypto.createHash('sha256').update(`connector-state:${svc}`).digest('hex')
+}
+
+function b64urlEncode(buf: Buffer): string {
+  return buf.toString('base64').replace(/\+/g, '-').replace(/\//g, '_').replace(/=+$/, '')
+}
+function b64urlDecode(s: string): Buffer {
+  return Buffer.from(s.replace(/-/g, '+').replace(/_/g, '/'), 'base64')
+}
+
+export function signConnectorState(payload: Omit<ConnectorStatePayload, 'iat'>, now = Date.now()): string {
+  const body = b64urlEncode(Buffer.from(JSON.stringify({ ...payload, iat: now })))
+  const sig = crypto.createHmac('sha256', getSecret()).update(`${VERSION}.${body}`).digest()
+  return `${VERSION}.${body}.${b64urlEncode(sig)}`
+}
+
+export type VerifyStateResult =
+  | { ok: true; payload: ConnectorStatePayload }
+  | { ok: false; reason: 'malformed' | 'bad_signature' | 'expired' }
+
+export function verifyConnectorState(token: string, now = Date.now(), ttlMs = DEFAULT_TTL_MS): VerifyStateResult {
+  const parts = token.split('.')
+  if (parts.length !== 3 || parts[0] !== VERSION) return { ok: false, reason: 'malformed' }
+  const [, body, sig] = parts
+  const expected = crypto.createHmac('sha256', getSecret()).update(`${VERSION}.${body}`).digest()
+  const given = b64urlDecode(sig)
+  if (given.length !== expected.length || !crypto.timingSafeEqual(given, expected)) {
+    return { ok: false, reason: 'bad_signature' }
+  }
+  let payload: ConnectorStatePayload
+  try {
+    payload = JSON.parse(b64urlDecode(body).toString('utf8')) as ConnectorStatePayload
+  } catch {
+    return { ok: false, reason: 'malformed' }
+  }
+  if (typeof payload.iat !== 'number' || now - payload.iat > ttlMs || payload.iat > now + 60_000) {
+    return { ok: false, reason: 'expired' }
+  }
+  return { ok: true, payload }
+}
+
+/** True when a raw upstream `state` value is one of our signed connector states. */
+export function isConnectorState(state: string | null | undefined): boolean {
+  return typeof state === 'string' && state.startsWith(`${VERSION}.`)
+}

lib/connect/hosted/upstream-budget.ts

@@ -0,0 +1,68 @@
+import type { SupabaseClient } from '@supabase/supabase-js'
+
+/**
+ * Global upstream rate budget for the connector proxy.
+ *
+ * Enable Banking's quotas (Annex 1 §5: 20 rps / 300 rpm / 10 000 per hour) are
+ * shared by ALL of Arcim's traffic, hosted included. So connector traffic gets
+ * a ceiling well under those, reserved atomically in the DB (RPC
+ * connector_reserve_upstream) so two proxy requests can't both slip past. A
+ * self-hoster that hits the ceiling gets a 429 with Retry-After; hosted bank
+ * sync is never starved because the connector ceiling is a fraction of the
+ * provider quota.
+ *
+ * Configurable per service via env; the defaults sit around 30% of the EB
+ * per-minute quota.
+ */
+
+export type UpstreamService = 'bank' | 'skatteverket'
+
+interface Budget {
+  minuteMax: number
+  hourMax: number
+}
+
+function intFromEnv(name: string, fallback: number): number {
+  const v = Number(process.env[name])
+  return Number.isFinite(v) && v > 0 ? Math.floor(v) : fallback
+}
+
+export function budgetFor(service: UpstreamService): Budget {
+  if (service === 'bank') {
+    return {
+      minuteMax: intFromEnv('CONNECT_BANK_RPM_BUDGET', 90), // ~30% of EB's 300/min
+      hourMax: intFromEnv('CONNECT_BANK_RPH_BUDGET', 3000), // ~30% of EB's 10 000/h
+    }
+  }
+  return {
+    minuteMax: intFromEnv('CONNECT_SKV_RPM_BUDGET', 120),
+    hourMax: intFromEnv('CONNECT_SKV_RPH_BUDGET', 4000),
+  }
+}
+
+export type BudgetResult = { ok: true } | { ok: false; scope: 'minute' | 'hour'; retryAfterSec: number }
+
+/**
+ * Reserve one upstream call. Returns ok:false with a Retry-After when the
+ * global budget for this service is exhausted. A DB error fails OPEN (ok:true):
+ * the budget is a protective cap, not an auth boundary, and blocking every
+ * connector call because the counter table hiccuped would be worse than a
+ * brief overshoot the provider itself also rate-limits.
+ */
+export async function reserveUpstream(
+  supabase: SupabaseClient,
+  service: UpstreamService,
+): Promise<BudgetResult> {
+  const { minuteMax, hourMax } = budgetFor(service)
+  const { data, error } = await supabase.rpc('connector_reserve_upstream', {
+    p_service: service,
+    p_minute_max: minuteMax,
+    p_hour_max: hourMax,
+  })
+  if (error) return { ok: true }
+  const row = (data ?? {}) as { ok?: boolean; scope?: 'minute' | 'hour'; retry_after_sec?: number }
+  if (row.ok === false) {
+    return { ok: false, scope: row.scope ?? 'minute', retryAfterSec: row.retry_after_sec ?? 60 }
+  }
+  return { ok: true }
+}

lib/connect/hosted/with-connector-auth.ts

@@ -45,6 +45,23 @@ export function extractConnectorKey(request: Request): string | null {
   return bearer
 }
 
+/**
+ * Opaque path segments (UUIDs, long hex, long base64url tokens) become ':id'
+ * before a path is persisted for metering. Literal route words (sessions,
+ * accounts, balances, aspsps, ...) survive, so the metric keys stay useful.
+ */
+const OPAQUE_SEGMENT = /^(?:[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}|[0-9a-f]{16,}|[A-Za-z0-9_-]{20,})$/i
+
+export function redactEndpoint(pathname: string): string {
+  return pathname
+    .split('/')
+    // A percent-encoded segment is opaque too: URL.pathname does not decode,
+    // so an encoded handle would otherwise slip the pattern while the route
+    // decodes and uses it.
+    .map((segment) => (OPAQUE_SEGMENT.test(segment) || segment.includes('%') ? ':id' : segment))
+    .join('/')
+}
+
 export function withConnectorAuth(
   operation: string,
   handler: ConnectorHandler,
@@ -84,10 +101,14 @@ export function withConnectorAuth(
     }
     response.headers.set('X-Request-Id', requestId)
 
-    // Metering: one row per request, never on the critical path.
+    // Metering: one row per request, never on the critical path. The path is
+    // REDACTED first: proxied paths carry the EB session id / account uid as
+    // segments, and the whole ledger design is that those handles never rest
+    // hosted-side (connector_connections stores sha256 only). Persisting the
+    // raw pathname would put the cleartext handle in connector_usage_events.
     const endpoint = (() => {
       try {
-        return new URL(request.url).pathname
+        return redactEndpoint(new URL(request.url).pathname)
       } catch {
         return null
       }

lib/connect/upstreams/enable-banking-jwt.ts

@@ -0,0 +1,117 @@
+/**
+ * JWT generation for Enable Banking API authentication
+ *
+ * Enable Banking requires JWT tokens signed with RS256 using your private key.
+ * The JWT is included in the Authorization header for all API calls.
+ */
+
+import * as crypto from 'crypto'
+
+// Prefer _PRODUCTION variants when available (Vercel production deploys)
+const APP_ID = process.env.ENABLE_BANKING_APP_ID_PRODUCTION || process.env.ENABLE_BANKING_APP_ID
+const PRIVATE_KEY_RAW = process.env.ENABLE_BANKING_PRIVATE_KEY_PRODUCTION || process.env.ENABLE_BANKING_PRIVATE_KEY
+
+interface JWTHeader {
+  typ: string
+  alg: string
+  kid: string
+}
+
+interface JWTPayload {
+  iss: string
+  aud: string
+  iat: number
+  exp: number
+}
+
+function base64UrlEncode(data: Buffer | string): string {
+  const str = typeof data === 'string' ? data : data.toString('base64')
+  return str.replace(/\+/g, '-').replace(/\//g, '_').replace(/=+$/, '')
+}
+
+function getPrivateKey(): string {
+  if (!PRIVATE_KEY_RAW) {
+    throw new Error('ENABLE_BANKING_PRIVATE_KEY environment variable is not set')
+  }
+
+  // Try decoding as base64-encoded PEM (sandbox format: base64 wrapping a PEM string)
+  const decoded = Buffer.from(PRIVATE_KEY_RAW, 'base64').toString('utf-8')
+  if (decoded.startsWith('-----BEGIN')) {
+    return decoded
+  }
+
+  // Otherwise treat as raw base64 DER key material: wrap in PEM headers
+  const lines = PRIVATE_KEY_RAW.match(/.{1,64}/g) || []
+  return `-----BEGIN PRIVATE KEY-----\n${lines.join('\n')}\n-----END PRIVATE KEY-----`
+}
+
+/**
+ * Generate a JWT token for Enable Banking API authentication
+ *
+ * @param expiresInSeconds - Token validity in seconds (default: 3600 = 1 hour)
+ * @returns Signed JWT token
+ */
+export function generateJWT(expiresInSeconds: number = 3600): string {
+  if (!APP_ID) {
+    throw new Error('ENABLE_BANKING_APP_ID environment variable is not set')
+  }
+
+  const now = Math.floor(Date.now() / 1000)
+
+  const header: JWTHeader = {
+    typ: 'JWT',
+    alg: 'RS256',
+    kid: APP_ID
+  }
+
+  const payload: JWTPayload = {
+    iss: 'enablebanking.com',
+    aud: 'api.enablebanking.com',
+    iat: now,
+    exp: now + expiresInSeconds
+  }
+
+  // Encode header and payload
+  const headerBase64 = base64UrlEncode(Buffer.from(JSON.stringify(header)))
+  const payloadBase64 = base64UrlEncode(Buffer.from(JSON.stringify(payload)))
+
+  // Create signature
+  const signatureInput = `${headerBase64}.${payloadBase64}`
+  const privateKey = getPrivateKey()
+
+  const sign = crypto.createSign('RSA-SHA256')
+  sign.update(signatureInput)
+  sign.end()
+
+  const signature = sign.sign(privateKey)
+  const signatureBase64 = base64UrlEncode(signature)
+
+  return `${headerBase64}.${payloadBase64}.${signatureBase64}`
+}
+
+// JWT token cache
+let cachedToken: string | null = null
+let cachedTokenExpiry: number = 0
+
+/**
+ * Get the Authorization header value for Enable Banking API.
+ * Caches JWT tokens and reuses them until 60s before expiry.
+ */
+export function getAuthorizationHeader(): string {
+  const now = Math.floor(Date.now() / 1000)
+  if (cachedToken && now < cachedTokenExpiry - 60) {
+    return `Bearer ${cachedToken}`
+  }
+
+  const expiresInSeconds = 3600
+  const token = generateJWT(expiresInSeconds)
+  cachedToken = token
+  cachedTokenExpiry = now + expiresInSeconds
+  return `Bearer ${token}`
+}
+
+/** @internal Reset token cache: for testing only */
+export function _resetTokenCache(): void {
+  cachedToken = null
+  cachedTokenExpiry = 0
+}