feat(connect): bank sync through the connector operation, with a per-company canary (#2205)
* feat(connect): bank sync through the connector operation, with a per-company canary In connector mode the enable-banking sync no longer pages Enable Banking on the instance: it calls POST /api/connect/bank/sync on the hosted service with the session id it holds and the account, and receives booked, normalized rows plus the raw provider pages to archive. Everything downstream is shared with the direct path (stored external ids computed here from booking_date, amount and the account scope; ingest; archive; balance refresh), so a company that moves to the connector produces byte-identical keys. A 410 from the service maps onto the same SessionExpiredError the direct path throws. bankConnectorMode(companyId) gains CONNECT_BANK_CANARY_COMPANIES: listed companies use the connector even while the installation has its own Enable Banking credentials, which is how hosted Accounted moves its bank sync to Connect a few companies at a time before dropping its keys. The contract package gains the bank sync request/response schemas (2026-09-03). Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Signed-off-by: Jakob Wennberg <311770904+jakobwennberg-oss@users.noreply.github.com> * fix(connect): calendar-valid dates, body read inside the timeout, service origin as the default Review follow-ups on #2205. The contract validates date_from, date_to and booking_date with z.iso.date() (2026-02-30 and an empty booking date are refused; the installation derives its stored keys from booking_date). The connector sync reads the response body inside the timeout window so a service that stalls the body cannot hold the sync open. DEFAULT_CONNECT_BASE_URL now names the connector service (connect.accounted.se), which is where the sync operation exists; the hosted app's copy of the connector routes is legacy and hosted Accounted itself sets GNUBOK_CONNECT_URL explicitly. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Signed-off-by: Jakob Wennberg <311770904+jakobwennberg-oss@users.noreply.github.com> --------- Signed-off-by: Jakob Wennberg <311770904+jakobwennberg-oss@users.noreply.github.com> Co-authored-by: Jakob Wennberg <311770904+jakobwennberg-oss@users.noreply.github.com> Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Jakob Wennberg
co-authored by
Claude Fable 5.1
Jakob Wennberg
Claude Fable 5.1
Jakob Wennberg
parent
828628d882
commit
4c6feea64d
@@ -0,0 +1,114 @@
|
||||
import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'
|
||||
|
||||
const h = vi.hoisted(() => ({ upload: vi.fn(), connector: { value: null as null | { baseUrl: string; key: string } } }))
|
||||
vi.mock('@/lib/core/documents/document-service', () => ({ uploadDocument: (...a: unknown[]) => h.upload(...a) }))
|
||||
vi.mock('@/lib/connect/instance/upstreams', () => ({
|
||||
bankConnectorMode: () => h.connector.value,
|
||||
CONNECTOR_COMPANY_HEADER: 'X-Connector-Company',
|
||||
}))
|
||||
vi.mock('../lib/api-client', async () => {
|
||||
const actual = await vi.importActual<typeof import('../lib/api-client')>('../lib/api-client')
|
||||
return { ...actual, getAllTransactionsWithRaw: vi.fn(), getAccountBalance: vi.fn().mockResolvedValue(null) }
|
||||
})
|
||||
|
||||
import { syncAccountTransactions } from '../lib/sync'
|
||||
import { SessionExpiredError } from '../lib/api-client'
|
||||
import { buildStableExternalIds } from '@/lib/transactions/external-id'
|
||||
import type { StoredAccount } from '../types'
|
||||
|
||||
const fetchMock = vi.fn()
|
||||
vi.stubGlobal('fetch', fetchMock)
|
||||
|
||||
const account: StoredAccount = {
|
||||
uid: 'acc-1',
|
||||
iban: 'SE45 5000 0000 0583 9825 7466',
|
||||
name: 'Företagskonto',
|
||||
currency: 'SEK',
|
||||
enabled: true,
|
||||
balance_updated_at: new Date().toISOString(),
|
||||
}
|
||||
|
||||
/** A supabase stand-in that answers the connection lookup with the session id. */
|
||||
const supabase = {
|
||||
from: (table: string) => {
|
||||
if (table !== 'bank_connections') throw new Error(`unexpected table ${table}`)
|
||||
const chain = { select: () => chain, eq: () => chain, maybeSingle: async () => ({ data: { session_id: 'sess-1' }, error: null }) }
|
||||
return chain
|
||||
},
|
||||
} as never
|
||||
|
||||
const remote = {
|
||||
transactions: [
|
||||
{ booking_date: '2026-09-01', amount: -125.5, currency: 'SEK', description: 'Kortköp ICA', counterparty_name: 'ICA Maxi', counterparty_account: 'SE45 5000 0000 0583 9825 7466', reference: null, merchant_category_code: '5411', bank_transaction_code: 'PMNT/CCRD', proprietary_bank_transaction_code: null },
|
||||
{ booking_date: '2026-09-01', amount: -125.5, currency: 'SEK', description: 'Kortköp ICA', counterparty_name: 'ICA Maxi', counterparty_account: '123456789', reference: null, merchant_category_code: null, bank_transaction_code: null, proprietary_bank_transaction_code: null },
|
||||
],
|
||||
raw_pages: ['{"page":1}', '{"page":2}'],
|
||||
skipped_pending: 1,
|
||||
returned_min_booking_date: '2026-09-01',
|
||||
returned_max_booking_date: '2026-09-01',
|
||||
effective_date_from: null,
|
||||
pages: 2,
|
||||
}
|
||||
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks()
|
||||
h.connector.value = { baseUrl: 'https://connect.accounted.se/api/connect/bank', key: 'gnubok_ck_test' }
|
||||
h.upload.mockResolvedValue({ id: 'doc' })
|
||||
})
|
||||
afterEach(() => vi.unstubAllEnvs())
|
||||
|
||||
describe('syncAccountTransactions in connector mode', () => {
|
||||
it('calls the connector sync operation and ingests with the same stored keys the direct path mints', async () => {
|
||||
fetchMock.mockResolvedValueOnce(new Response(JSON.stringify(remote), { status: 200, headers: { 'content-type': 'application/json' } }))
|
||||
const ingest = vi.fn().mockResolvedValue({ imported: 2, duplicates: 0, errors: 0, reconciled: 0, auto_categorized: 0, auto_matched_invoices: 0, transaction_ids: [] })
|
||||
const result = await syncAccountTransactions(supabase, 'company-1', 'user-1', 'conn-1', account, '2026-08-01', '2026-09-03', ingest, { strategy: 'longest' })
|
||||
|
||||
const [url, init] = fetchMock.mock.calls[0] as [string, RequestInit]
|
||||
expect(url).toBe('https://connect.accounted.se/api/connect/bank/sync')
|
||||
expect((init.headers as Record<string, string>).Authorization).toBe('Bearer gnubok_ck_test')
|
||||
expect((init.headers as Record<string, string>)['X-Connector-Company']).toBe('company-1')
|
||||
expect(JSON.parse(init.body as string)).toEqual({ session_id: 'sess-1', account_uid: 'acc-1', account_currency: 'SEK', date_from: '2026-08-01', date_to: '2026-09-03', strategy: 'longest' })
|
||||
|
||||
const raw = ingest.mock.calls[0][3] as Array<{ external_id: string; date: string; amount: number; counterparty_iban: string | null; counterparty_account: string | null; import_source: string; bank_connection_id: string }>
|
||||
const expectedIds = buildStableExternalIds('eb', 'SE4550000000058398257466', [{ date: '2026-09-01', amount: -125.5 }, { date: '2026-09-01', amount: -125.5 }])
|
||||
expect(raw.map((r) => r.external_id)).toEqual(expectedIds)
|
||||
expect(raw[0]).toMatchObject({ date: '2026-09-01', amount: -125.5, counterparty_iban: 'SE4550000000058398257466', counterparty_account: null, import_source: 'enable_banking', bank_connection_id: 'conn-1' })
|
||||
expect(raw[1]).toMatchObject({ counterparty_iban: null, counterparty_account: '123456789' })
|
||||
expect(account.dedup_scope).toBe('SE4550000000058398257466')
|
||||
expect(h.upload).toHaveBeenCalledTimes(2)
|
||||
expect(result).toMatchObject({ imported: 2, returnedMinBookingDate: '2026-09-01', returnedMaxBookingDate: '2026-09-01' })
|
||||
})
|
||||
|
||||
it('maps the connector 410 onto SessionExpiredError so callers flip the connection to expired', async () => {
|
||||
fetchMock.mockResolvedValueOnce(new Response(JSON.stringify({ error: 'expired', code: 'CONNECTOR_BANK_SESSION_EXPIRED', retryable: false }), { status: 410 }))
|
||||
await expect(syncAccountTransactions(supabase, 'company-1', 'user-1', 'conn-1', account, '2026-08-01', '2026-09-03', vi.fn())).rejects.toBeInstanceOf(SessionExpiredError)
|
||||
})
|
||||
|
||||
it('surfaces other connector failures as errors and refuses an unexpected response shape', async () => {
|
||||
fetchMock.mockResolvedValueOnce(new Response(JSON.stringify({ error: 'busy', code: 'CONNECTOR_RATE_LIMITED' }), { status: 429 }))
|
||||
await expect(syncAccountTransactions(supabase, 'company-1', 'user-1', 'conn-1', account, '2026-08-01', '2026-09-03', vi.fn())).rejects.toThrow(/CONNECTOR_RATE_LIMITED/)
|
||||
fetchMock.mockResolvedValueOnce(new Response(JSON.stringify({ transactions: 'nope' }), { status: 200 }))
|
||||
await expect(syncAccountTransactions(supabase, 'company-1', 'user-1', 'conn-1', account, '2026-08-01', '2026-09-03', vi.fn())).rejects.toThrow(/unexpected shape/)
|
||||
})
|
||||
})
|
||||
|
||||
describe('connector timeout covers the body read', () => {
|
||||
it('aborts a response whose body stalls instead of hanging past the budget', async () => {
|
||||
vi.useFakeTimers()
|
||||
try {
|
||||
const stalled = { ok: true, status: 200, text: () => new Promise<string>(() => {}) } as unknown as Response
|
||||
fetchMock.mockImplementationOnce(() => Promise.resolve(stalled))
|
||||
const pending = syncAccountTransactions(supabase, 'company-1', 'user-1', 'conn-1', account, '2026-08-01', '2026-09-03', vi.fn())
|
||||
// Only the abort signal can end this: the body promise never settles.
|
||||
const raced = Promise.race([pending.then(() => 'settled', () => 'settled'), new Promise((r) => setTimeout(r, 200_000, 'timed-out'))])
|
||||
await vi.advanceTimersByTimeAsync(130_000)
|
||||
expect(fetchMock).toHaveBeenCalledTimes(1)
|
||||
const signal = (fetchMock.mock.calls[0] as [string, RequestInit])[1].signal as AbortSignal
|
||||
expect(signal.aborted).toBe(true)
|
||||
await vi.advanceTimersByTimeAsync(200_000)
|
||||
await raced
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
})
|
||||
@@ -1,5 +1,7 @@
|
||||
import type { SupabaseClient } from '@supabase/supabase-js'
|
||||
import { getAllTransactionsWithRaw, convertTransaction, getAccountBalance } from './api-client'
|
||||
import { getAllTransactionsWithRaw, convertTransaction, getAccountBalance, SessionExpiredError } from './api-client'
|
||||
import { bankSyncResponseSchema, connectorErrorSchema } from '@accounted/connect-contract'
|
||||
import { bankConnectorMode, CONNECTOR_COMPANY_HEADER } from '@/lib/connect/instance/upstreams'
|
||||
import { uploadDocument } from '@/lib/core/documents/document-service'
|
||||
import { ingestTransactions as defaultIngest } from '@/lib/transactions/ingest'
|
||||
import { buildStableExternalIds, FALLBACK_DESCRIPTION } from '@/lib/transactions/external-id'
|
||||
@@ -46,6 +48,101 @@ export interface SyncResult {
|
||||
returnedMaxBookingDate?: string
|
||||
}
|
||||
|
||||
/** The subset of a converted bank transaction that the ingest mapping below reads. */
|
||||
interface BookedTransactionFields {
|
||||
amount: number
|
||||
currency: string
|
||||
description: string
|
||||
counterparty_name?: string
|
||||
counterparty_account?: string
|
||||
reference?: string
|
||||
merchant_category_code?: string
|
||||
bank_transaction_code?: string
|
||||
proprietary_bank_transaction_code?: string
|
||||
}
|
||||
|
||||
const CONNECTOR_SYNC_TIMEOUT_MS = 120_000
|
||||
|
||||
/**
|
||||
* Booked transactions through the hosted connector. The session id is the
|
||||
* installation's own (it stays here); the service proves ownership from its
|
||||
* ledger, does the provider work, and answers with the contract's response
|
||||
* shape. A 410 means the consent is over and maps onto the same
|
||||
* SessionExpiredError the direct path throws, so callers flip the connection
|
||||
* to expired identically.
|
||||
*/
|
||||
async function fetchBookedViaConnector(
|
||||
connector: { baseUrl: string; key: string },
|
||||
args: {
|
||||
supabase: SupabaseClient
|
||||
companyId: string
|
||||
connectionId: string
|
||||
account: StoredAccount
|
||||
fromDate: string
|
||||
toDate: string
|
||||
strategy?: TransactionsFetchStrategy
|
||||
},
|
||||
): Promise<ReturnType<typeof bankSyncResponseSchema.parse>> {
|
||||
// The PSD2 session id rests on the connection row (never on the account
|
||||
// payload), and it stays on this installation: the service only receives
|
||||
// it per call and proves ownership from its own ledger.
|
||||
const { data: row, error } = await args.supabase
|
||||
.from('bank_connections')
|
||||
.select('session_id')
|
||||
.eq('id', args.connectionId)
|
||||
.maybeSingle()
|
||||
if (error) throw new Error(`Connector bank sync could not read the connection: ${error.message}`)
|
||||
const sessionId = (row as { session_id: string | null } | null)?.session_id
|
||||
if (!sessionId) throw new Error('Connector bank sync requires a connection with a session id')
|
||||
// The body is read INSIDE the timeout window: a service that sends headers
|
||||
// and then stalls the body must not hold the sync open past the budget.
|
||||
const controller = new AbortController()
|
||||
const timeout = setTimeout(() => controller.abort(), CONNECTOR_SYNC_TIMEOUT_MS)
|
||||
let response: Response
|
||||
let text: string
|
||||
try {
|
||||
response = await fetch(`${connector.baseUrl}/sync`, {
|
||||
method: 'POST',
|
||||
signal: controller.signal,
|
||||
redirect: 'error',
|
||||
headers: {
|
||||
Authorization: `Bearer ${connector.key}`,
|
||||
'Content-Type': 'application/json',
|
||||
Accept: 'application/json',
|
||||
[CONNECTOR_COMPANY_HEADER]: args.companyId,
|
||||
},
|
||||
body: JSON.stringify({
|
||||
session_id: sessionId,
|
||||
account_uid: args.account.uid,
|
||||
account_currency: args.account.currency,
|
||||
date_from: args.fromDate,
|
||||
date_to: args.toDate,
|
||||
...(args.strategy ? { strategy: args.strategy } : {}),
|
||||
}),
|
||||
})
|
||||
text = await response.text()
|
||||
} finally {
|
||||
clearTimeout(timeout)
|
||||
}
|
||||
let json: unknown = null
|
||||
try {
|
||||
json = text ? JSON.parse(text) : null
|
||||
} catch {
|
||||
json = null
|
||||
}
|
||||
if (!response.ok) {
|
||||
const envelope = connectorErrorSchema.safeParse(json)
|
||||
const code = envelope.success ? envelope.data.code : `HTTP_${response.status}`
|
||||
if (response.status === 410 || code === 'CONNECTOR_BANK_SESSION_EXPIRED') {
|
||||
throw new SessionExpiredError(response.status, text)
|
||||
}
|
||||
throw new Error(`Connector bank sync failed (${response.status} ${code})`)
|
||||
}
|
||||
const parsed = bankSyncResponseSchema.safeParse(json)
|
||||
if (!parsed.success) throw new Error('Connector bank sync answered with an unexpected shape')
|
||||
return parsed.data
|
||||
}
|
||||
|
||||
/**
|
||||
* Sync transactions for a single bank account via Enable Banking PSD2.
|
||||
*
|
||||
@@ -77,28 +174,87 @@ export async function syncAccountTransactions(
|
||||
strategy: syncOptions?.strategy,
|
||||
})
|
||||
|
||||
const { transactions, rawPages } = await getAllTransactionsWithRaw(
|
||||
account.uid,
|
||||
fromDate,
|
||||
toDate,
|
||||
syncOptions?.strategy,
|
||||
)
|
||||
// Two ways to obtain booked transactions. Direct: this installation's own
|
||||
// Enable Banking credentials, exactly as before. Connector: the hosted
|
||||
// service does the paging, the booked-only filter and the normalization
|
||||
// (POST /api/connect/bank/sync) and returns rows plus the raw pages. Both
|
||||
// paths converge on bookedEntries + rawPages, and everything below
|
||||
// (external ids, ingest, archive, balance) is shared, so a company moved to
|
||||
// the connector produces byte-identical stored keys.
|
||||
const connector = bankConnectorMode(companyId)
|
||||
let rawPages: string[]
|
||||
let bookedEntries: Array<{ tx: BookedTransactionFields; bookingDate: string }>
|
||||
let totalFetched: number
|
||||
if (connector) {
|
||||
const remote = await fetchBookedViaConnector(connector, {
|
||||
supabase,
|
||||
companyId,
|
||||
connectionId,
|
||||
account,
|
||||
fromDate,
|
||||
toDate,
|
||||
strategy: syncOptions?.strategy,
|
||||
})
|
||||
rawPages = remote.raw_pages
|
||||
totalFetched = remote.transactions.length + remote.skipped_pending
|
||||
bookedEntries = remote.transactions.map((tx) => ({
|
||||
tx: {
|
||||
amount: tx.amount,
|
||||
currency: tx.currency,
|
||||
description: tx.description,
|
||||
counterparty_name: tx.counterparty_name ?? undefined,
|
||||
counterparty_account: tx.counterparty_account ?? undefined,
|
||||
reference: tx.reference ?? undefined,
|
||||
merchant_category_code: tx.merchant_category_code ?? undefined,
|
||||
bank_transaction_code: tx.bank_transaction_code ?? undefined,
|
||||
proprietary_bank_transaction_code: tx.proprietary_bank_transaction_code ?? undefined,
|
||||
},
|
||||
bookingDate: tx.booking_date,
|
||||
}))
|
||||
} else {
|
||||
const fetched = await getAllTransactionsWithRaw(
|
||||
account.uid,
|
||||
fromDate,
|
||||
toDate,
|
||||
syncOptions?.strategy,
|
||||
)
|
||||
rawPages = fetched.rawPages
|
||||
totalFetched = fetched.transactions.length
|
||||
const bankTransactions = fetched.transactions.map(tx => convertTransaction(tx, account.currency))
|
||||
// Only ingest BOOKED transactions: those the ASPSP returned with a real
|
||||
// booking_date. Pending entries are intentionally skipped: a pending row is
|
||||
// unstable across syncs (a later "synka nu" returns the same transaction
|
||||
// either still pending or finally booked, often with a *different* effective
|
||||
// date). Because BOTH the dedup external_id and the content-dedup key are
|
||||
// date-derived, that drift mints a brand-new id and re-imports a transaction
|
||||
// that already exists. Gating the import set on a stable booking_date
|
||||
// removes the drift at the source, and leaves booked rows' ids byte-identical.
|
||||
//
|
||||
// booking_date is read from the RAW transaction, index-aligned with
|
||||
// bankTransactions: convertTransaction's booking_date already falls back to
|
||||
// value_date/today, so it cannot tell booked from pending.
|
||||
bookedEntries = bankTransactions.flatMap((tx, i) => {
|
||||
const bookingDate = fetched.transactions[i]?.booking_date
|
||||
return typeof bookingDate === 'string' && bookingDate.trim() !== ''
|
||||
? [{ tx, bookingDate: bookingDate.trim() }]
|
||||
: []
|
||||
})
|
||||
}
|
||||
|
||||
// Log the actual date range returned so we can compare against the requested
|
||||
// window. Helps diagnose when an ASPSP truncates history below what we asked for.
|
||||
let minBookingDate: string | undefined
|
||||
let maxBookingDate: string | undefined
|
||||
for (const tx of transactions) {
|
||||
const d = tx.booking_date || tx.value_date
|
||||
if (!d) continue
|
||||
if (!minBookingDate || d < minBookingDate) minBookingDate = d
|
||||
if (!maxBookingDate || d > maxBookingDate) maxBookingDate = d
|
||||
for (const { bookingDate } of bookedEntries) {
|
||||
if (!minBookingDate || bookingDate < minBookingDate) minBookingDate = bookingDate
|
||||
if (!maxBookingDate || bookingDate > maxBookingDate) maxBookingDate = bookingDate
|
||||
}
|
||||
|
||||
console.log('[enable-banking] Fetched transactions from API', {
|
||||
console.log('[enable-banking] Fetched transactions', {
|
||||
connectionId,
|
||||
accountUid: account.uid,
|
||||
transactionCount: transactions.length,
|
||||
via: connector ? 'connector' : 'direct',
|
||||
transactionCount: totalFetched,
|
||||
rawPageCount: rawPages.length,
|
||||
requestedFromDate: fromDate,
|
||||
requestedToDate: toDate,
|
||||
@@ -107,37 +263,13 @@ export async function syncAccountTransactions(
|
||||
strategy: syncOptions?.strategy,
|
||||
})
|
||||
|
||||
const bankTransactions = transactions.map(tx => convertTransaction(tx, account.currency))
|
||||
|
||||
// Only ingest BOOKED transactions: those the ASPSP returned with a real
|
||||
// booking_date. Pending entries are intentionally skipped: a pending row is
|
||||
// unstable across syncs (a later "synka nu" returns the same transaction
|
||||
// either still pending or finally booked, often with a *different* effective
|
||||
// date). Because BOTH the dedup external_id and the content-dedup key are
|
||||
// date-derived, that drift mints a brand-new id and re-imports a transaction
|
||||
// that already exists. Observed in production as the same amount+description
|
||||
// landing twice with different dates: the bank's value_date in one sync, its
|
||||
// booking_date in another. Gating the import set on a stable booking_date
|
||||
// removes the drift at the source, and leaves booked rows' ids byte-identical
|
||||
// (so the existing rows are NOT re-orphaned).
|
||||
//
|
||||
// booking_date is read from the RAW transaction (transactions[i]), index-
|
||||
// aligned with bankTransactions: convertTransaction's booking_date already
|
||||
// falls back to value_date/today, so it cannot tell booked from pending.
|
||||
const bookedEntries = bankTransactions.flatMap((tx, i) => {
|
||||
const bookingDate = transactions[i]?.booking_date
|
||||
return typeof bookingDate === 'string' && bookingDate.trim() !== ''
|
||||
? [{ tx, bookingDate: bookingDate.trim() }]
|
||||
: []
|
||||
})
|
||||
|
||||
const skippedPending = bankTransactions.length - bookedEntries.length
|
||||
const skippedPending = totalFetched - bookedEntries.length
|
||||
if (skippedPending > 0) {
|
||||
console.log('[enable-banking] Skipped pending transactions (no booking_date)', {
|
||||
connectionId,
|
||||
accountUid: account.uid,
|
||||
skippedPending,
|
||||
total: bankTransactions.length,
|
||||
total: totalFetched,
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user