diff --git a/app/api/extensions/arcim-migration/complete-invoice-lines/cron/__tests__/route.test.ts b/app/api/extensions/arcim-migration/complete-invoice-lines/cron/__tests__/route.test.ts index 156b8c30..0813a694 100644 --- a/app/api/extensions/arcim-migration/complete-invoice-lines/cron/__tests__/route.test.ts +++ b/app/api/extensions/arcim-migration/complete-invoice-lines/cron/__tests__/route.test.ts @@ -50,7 +50,8 @@ const mockCount = vi.mocked(countRowlessInvoices) const EMPTY = { candidates: 0, providerInvoices: 0, matched: 0, unmatched: 0, completed: 0, headersUpdated: 0, - totalMismatch: 0, noLinesAtProvider: 0, rowsMismatch: 0, notHydrated: 0, vatUnresolved: 0, failed: 0, remaining: 0, + totalMismatch: 0, noLinesAtProvider: 0, rowsMismatch: 0, notHydrated: 0, vatUnresolved: 0, failed: 0, + historyAppended: 0, remaining: 0, hydration: { needed: 0, hydrated: 0, failed: 0, skippedForBudget: 0 }, dryRun: false, } diff --git a/app/api/extensions/arcim-migration/complete-invoice-lines/cron/route.ts b/app/api/extensions/arcim-migration/complete-invoice-lines/cron/route.ts index ff14a027..b17bbd96 100644 --- a/app/api/extensions/arcim-migration/complete-invoice-lines/cron/route.ts +++ b/app/api/extensions/arcim-migration/complete-invoice-lines/cron/route.ts @@ -159,6 +159,7 @@ export const GET = withCronContext('cron.arcim_migration_complete_invoice_lines' totalMismatch: 0, rowsMismatch: 0, failed: 0, + historyAppended: 0, } const completing = await ctx.forEach('register', registers, async ({ consent, candidates }, itemCtx) => { @@ -188,6 +189,7 @@ export const GET = withCronContext('cron.arcim_migration_complete_invoice_lines' totals.totalMismatch += result.totalMismatch totals.rowsMismatch += result.rowsMismatch totals.failed += result.failed + totals.historyAppended += result.historyAppended itemCtx.log.info('migrated invoice rows completed for company', { companyId: consent.company_id, provider: consent.provider, @@ -198,6 +200,8 @@ export const GET = withCronContext('cron.arcim_migration_complete_invoice_lines' notHydrated: result.notHydrated, totalMismatch: result.totalMismatch, rowsMismatch: result.rowsMismatch, + failed: result.failed, + historyAppended: result.historyAppended, hydration: result.hydration, }) }) diff --git a/extensions/general/arcim-migration/__tests__/migration-registration-links.test.ts b/extensions/general/arcim-migration/__tests__/migration-registration-links.test.ts index e13d1ce7..7e9dbd49 100644 --- a/extensions/general/arcim-migration/__tests__/migration-registration-links.test.ts +++ b/extensions/general/arcim-migration/__tests__/migration-registration-links.test.ts @@ -114,6 +114,7 @@ function baseOptions(overrides: Record = {}) { companyId: 'company-1', userId: 'user-1', supabase: supabase as unknown as SupabaseClient, + createHistoryClient: async () => ({ from: vi.fn() }) as unknown as Pick, importCompanyInfo: false, importCustomers: false, importSuppliers: false, diff --git a/extensions/general/arcim-migration/__tests__/migration-step-errors.test.ts b/extensions/general/arcim-migration/__tests__/migration-step-errors.test.ts index 003418b2..5d2a4bff 100644 --- a/extensions/general/arcim-migration/__tests__/migration-step-errors.test.ts +++ b/extensions/general/arcim-migration/__tests__/migration-step-errors.test.ts @@ -84,6 +84,7 @@ function baseOptions(overrides: Record = {}) { companyId: 'company-1', userId: 'user-1', supabase: {} as unknown as SupabaseClient, + createHistoryClient: async () => ({ from: vi.fn() }) as unknown as Pick, importCompanyInfo: false, importCustomers: false, importSuppliers: false, diff --git a/extensions/general/arcim-migration/index.ts b/extensions/general/arcim-migration/index.ts index 8e748401..15fe021d 100644 --- a/extensions/general/arcim-migration/index.ts +++ b/extensions/general/arcim-migration/index.ts @@ -1479,6 +1479,11 @@ export const arcimMigrationExtension: Extension = { companyId, userId: user.id, supabase, + // The behandlingshistorik rows the sales-invoice step writes need + // the service role (processing_history has no INSERT policy); + // built only when that step has rows to write. + createHistoryClient: async () => + (await import('@/lib/supabase/server')).createServiceClient(), importCompanyInfo, importCustomers, importSuppliers, diff --git a/extensions/general/arcim-migration/lib/__tests__/complete-invoice-lines.test.ts b/extensions/general/arcim-migration/lib/__tests__/complete-invoice-lines.test.ts index 41b8379a..3c9b599c 100644 --- a/extensions/general/arcim-migration/lib/__tests__/complete-invoice-lines.test.ts +++ b/extensions/general/arcim-migration/lib/__tests__/complete-invoice-lines.test.ts @@ -10,9 +10,17 @@ import type { SalesInvoiceDto } from '@/lib/providers/dto' * when the provider's total agrees with the stored one, and leave anything it * could not reach for the next run rather than guessing. Every write goes * through the complete_invoice_rows RPC, one call per invoice, which is what - * keeps two writers from doubling an invoice's rows. + * keeps two writers from doubling an invoice's rows, and every write that + * landed leaves one InvoiceRowsCompleted row in processing_history (#2312). + * The history append runs for real against the fake client, so the payload + * below is what the PII guard and the row shape actually accept. */ +vi.mock('@/lib/supabase/server', () => ({ + createClient: vi.fn(), + createServiceClient: vi.fn(), +})) + vi.mock('@/lib/providers/resolve-consent', () => ({ resolveConsent: vi.fn().mockResolvedValue({ consent: { provider: 'fortnox' }, @@ -86,6 +94,7 @@ function storedRow(overrides: Record = {}) { subtotal: 1250, vat_amount: 0, vat_rate: 25, + vat_treatment: 'standard_25', currency: 'SEK', exchange_rate: null, invoice_items: [], @@ -175,6 +184,18 @@ function headerUpdates(calls: Call[]): Record[] { .filter((h): h is Record => h !== null) } +/** The processing_history rows the run appended, in order. */ +function historyRows(calls: Call[]): Record[] { + return calls + .filter((c) => c.table === 'processing_history' && c.method === 'insert') + .map((c) => c.args[0] as Record) +} + +/** Every call that touched a table other than the trail. */ +function tableWrites(calls: Call[]): Call[] { + return calls.filter((c) => c.method !== 'rpc' && c.table !== 'processing_history') +} + describe('completeMigratedInvoiceLines', () => { beforeEach(() => { vi.clearAllMocks() @@ -193,6 +214,7 @@ describe('completeMigratedInvoiceLines', () => { expect(result).toMatchObject({ candidates: 1, providerInvoices: 1, matched: 1, unmatched: 0, completed: 1, headersUpdated: 1, remaining: 0, totalMismatch: 0, notHydrated: 0, failed: 0, + historyAppended: 1, }) // Only the matched subset is hydrated, so the budget is never spent on // invoices already complete on our side. @@ -228,8 +250,30 @@ describe('completeMigratedInvoiceLines', () => { // Never the total, status or payments. expect(Object.keys(sent[0].p_header!)).not.toContain('total') expect(Object.keys(sent[0].p_header!)).not.toContain('status') - // Nothing is written outside the RPC. - expect(calls.filter((c) => c.method !== 'rpc')).toHaveLength(0) + // Nothing is written outside the RPC, except the trail. + expect(tableWrites(calls)).toHaveLength(0) + + // One InvoiceRowsCompleted on the invoice: the writer, the provider, + // the consent, and the header split before and after (the pre-#1745 + // 25 %-beside-0-kr shape replaced by what the detail form established). + const trail = historyRows(calls) + expect(trail).toHaveLength(1) + expect(trail[0]).toMatchObject({ + company_id: 'co-1', + aggregate_type: 'Invoice', + aggregate_id: 'inv-1', + event_type: 'InvoiceRowsCompleted', + actor: { type: 'cron', id: 'complete-invoice-lines' }, + payload: { + source: 'complete-invoice-lines', + provider: 'fortnox', + consent_id: 'c-1', + rows: 1, + header_updated: true, + header_before: { subtotal: 1250, vat_amount: 0, vat_rate: 25, vat_treatment: 'standard_25' }, + header_after: { subtotal: 1000, vat_amount: 250, vat_rate: 25, vat_treatment: 'standard_25' }, + }, + }) }) it('writes the rows but leaves a header whose split is consistent (momsfri)', async () => { @@ -241,10 +285,14 @@ describe('completeMigratedInvoiceLines', () => { const result = await completeMigratedInvoiceLines({ supabase, companyId: 'co-1', consentId: 'c-1' }) - expect(result).toMatchObject({ completed: 1, headersUpdated: 0 }) + expect(result).toMatchObject({ completed: 1, headersUpdated: 0, historyAppended: 1 }) expect(insertedRows(calls)).toHaveLength(1) expect(writes(calls)[0].p_header).toBeNull() expect(headerUpdates(calls)).toHaveLength(0) + // The event says the header was left alone, and carries no split. + expect(historyRows(calls)[0]).toMatchObject({ + payload: { rows: 1, header_updated: false, header_before: null, header_after: null }, + }) }) it('fills a header whose rate is null (the post-#1745 "source did not say")', async () => { @@ -390,9 +438,11 @@ describe('completeMigratedInvoiceLines', () => { const result = await completeMigratedInvoiceLines({ supabase, companyId: 'co-1', consentId: 'c-1' }) - expect(result).toMatchObject({ completed: 0, headersUpdated: 0, failed: 0, remaining: 1 }) + expect(result).toMatchObject({ completed: 0, headersUpdated: 0, failed: 0, remaining: 1, historyAppended: 0 }) expect(writes(calls)).toHaveLength(1) expect(writes(calls)[0].p_header).not.toBeNull() + // Nothing changed, so nothing is recorded. + expect(historyRows(calls)).toHaveLength(0) }) it('counts a refused or failed call as failed and goes on with the rest', async () => { @@ -419,8 +469,46 @@ describe('completeMigratedInvoiceLines', () => { const result = await completeMigratedInvoiceLines({ supabase, companyId: 'co-1', consentId: 'c-1' }) - expect(result).toMatchObject({ completed: 1, failed: 2, headersUpdated: 1, remaining: 2 }) + expect(result).toMatchObject({ completed: 1, failed: 2, headersUpdated: 1, remaining: 2, historyAppended: 1 }) expect(writes(calls).map((w) => w.p_invoice_id)).toEqual(['inv-1', 'inv-2', 'inv-3']) + // The trail names only the invoice whose write landed. + expect(historyRows(calls).map((r) => r.aggregate_id)).toEqual(['inv-1']) + }) + + it('threads one correlation id through every event of a run, attributed to the caller\'s actor', async () => { + mFetchAll.mockResolvedValue([storedRow(), storedRow({ id: 'inv-2', invoice_number: '1002' })]) + const dtos = [providerInvoice(), providerInvoice({ id: '1002', invoiceNumber: '1002' })] + mList.mockResolvedValue(dtos) + mHydrate.mockResolvedValue(hydratedAll(dtos)) + const { supabase, calls } = makeSupabase(() => ok) + + const result = await completeMigratedInvoiceLines({ + supabase, companyId: 'co-1', consentId: 'c-1', actor: { type: 'user', id: 'user-9' }, + }) + + expect(result).toMatchObject({ completed: 2, historyAppended: 2 }) + const trail = historyRows(calls) + expect(trail.map((r) => r.aggregate_id)).toEqual(['inv-1', 'inv-2']) + expect(trail[0].correlation_id).toMatch(/^[0-9a-f-]{36}$/) + expect(trail[1].correlation_id).toBe(trail[0].correlation_id) + expect(trail.every((r) => (r.actor as { id: string }).id === 'user-9')).toBe(true) + }) + + it('counts a write whose trail append failed as completed, not appended', async () => { + // The rows are committed by then; a missing change-log row is a logged + // gap, never a failed invoice the next run would find full. + mFetchAll.mockResolvedValue([storedRow()]) + const dto = providerInvoice() + mList.mockResolvedValue([dto]) + mHydrate.mockResolvedValue(hydratedAll([dto])) + const { supabase, calls } = makeSupabase((table) => + table === 'processing_history' ? { data: null, error: { message: 'connection reset' } } : ok, + ) + + const result = await completeMigratedInvoiceLines({ supabase, companyId: 'co-1', consentId: 'c-1' }) + + expect(result).toMatchObject({ completed: 1, failed: 0, historyAppended: 0 }) + expect(historyRows(calls)).toHaveLength(1) }) it('dry run: reports the plan and writes nothing', async () => { @@ -432,7 +520,7 @@ describe('completeMigratedInvoiceLines', () => { const result = await completeMigratedInvoiceLines({ supabase, companyId: 'co-1', consentId: 'c-1', dryRun: true }) - expect(result).toMatchObject({ dryRun: true, completed: 1, headersUpdated: 1, remaining: 0 }) + expect(result).toMatchObject({ dryRun: true, completed: 1, headersUpdated: 1, remaining: 0, historyAppended: 0 }) expect(calls).toHaveLength(0) }) diff --git a/extensions/general/arcim-migration/lib/complete-invoice-lines.ts b/extensions/general/arcim-migration/lib/complete-invoice-lines.ts index 1459a42e..c21775f1 100644 --- a/extensions/general/arcim-migration/lib/complete-invoice-lines.ts +++ b/extensions/general/arcim-migration/lib/complete-invoice-lines.ts @@ -35,11 +35,22 @@ * locks the invoice, inserts the rows only when the invoice still has none * and applies the header split in the same transaction, so an invoice's rows * are written at most once whichever writer gets there first, and "rows - * landed, header did not" is not a reachable state. + * landed, header did not" is not a reachable state. The call goes through + * completeInvoiceRows (lib/invoices/complete-invoice-rows.ts), which also + * writes the behandlingshistorik event for every invoice the RPC filled: one + * InvoiceRowsCompleted per invoice, with the header split before and after + * (BFL 5 kap 11 §, BFNAR 2013:2 p. 9.16). Every invoice a run completes + * shares one correlation id, so a run is one thread in the trail. */ import type { SupabaseClient } from '@supabase/supabase-js' +import type { ProcessingHistoryActor } from '@/types' import { ISO_DATE_RE } from '@/lib/invariants' +import { + completeInvoiceRows, + type CompleteInvoiceRowsTrail, + type InvoiceHeaderVatSplit, +} from '@/lib/invoices/complete-invoice-rows' import { createLogger } from '@/lib/logger' import { equalOre, roundOre } from '@/lib/money' import type { SalesInvoiceDto } from '@/lib/providers/dto' @@ -55,6 +66,9 @@ import { mapSalesInvoice } from './entity-mapper' const log = createLogger('extensions/arcim-migration/complete-invoice-lines') +/** The writer named in the trail; also the default actor id (the hourly cron). */ +export const COMPLETE_INVOICE_LINES_SOURCE = 'complete-invoice-lines' + /** * How many invoices in this company the pass would try to complete: the * predicate `loadCandidates` uses (non-draft sales invoices with no rows), @@ -87,6 +101,8 @@ export interface CompleteInvoiceLinesOptions { dryRun?: boolean /** Wall-clock ceiling for the provider detail fetches, in ms. */ budgetMs?: number + /** Who the behandlingshistorik events are attributed to; the hourly cron when unset. */ + actor?: ProcessingHistoryActor } export interface CompleteInvoiceLinesResult { @@ -118,6 +134,8 @@ export interface CompleteInvoiceLinesResult { vatUnresolved: number /** Invoices whose write failed at the database. */ failed: number + /** Of `completed`, those whose InvoiceRowsCompleted event landed in processing_history. */ + historyAppended: number /** Candidates still without rows after this run: `candidates - completed`. */ remaining: number hydration: HydrationReport @@ -134,6 +152,7 @@ interface CandidateRow { subtotal: number | null vat_amount: number | null vat_rate: number | null + vat_treatment: string | null currency: string | null exchange_rate: number | null invoice_items: { id: string }[] | null @@ -208,7 +227,7 @@ async function loadCandidates(supabase: SupabaseClient, companyId: string): Prom supabase .from('invoices') .select( - 'id, user_id, customer_id, invoice_number, invoice_date, total, subtotal, vat_amount, vat_rate, currency, exchange_rate, invoice_items(id)', + 'id, user_id, customer_id, invoice_number, invoice_date, total, subtotal, vat_amount, vat_rate, vat_treatment, currency, exchange_rate, invoice_items(id)', ) .eq('company_id', companyId) .eq('document_type', 'invoice') @@ -222,39 +241,21 @@ async function loadCandidates(supabase: SupabaseClient, companyId: string): Prom return rows.filter((row) => (row.invoice_items?.length ?? 0) === 0) } -/** - * The header VAT split the detail form established, as the RPC's p_header: - * the six invoice columns it may rewrite, all present or none. - */ -interface HeaderFill { - subtotal: number - subtotal_sek: number | null - vat_amount: number - vat_amount_sek: number | null - vat_rate: number | null - vat_treatment: string -} - interface PlannedWrite { row: CandidateRow /** The invoice_items columns per row; the RPC stamps invoice_id itself. */ items: Record[] - header: HeaderFill | null -} - -/** What complete_invoice_rows returns (migration 20260906135730). */ -interface CompleteRowsOutcome { - ok: boolean - code?: string - wrote?: boolean - rows?: number - header_updated?: boolean + /** The header VAT split the detail form established, as the RPC's p_header. */ + header: InvoiceHeaderVatSplit | null } export async function completeMigratedInvoiceLines( options: CompleteInvoiceLinesOptions, ): Promise { - const { supabase, companyId, consentId, dryRun = false, budgetMs } = options + const { + supabase, companyId, consentId, dryRun = false, budgetMs, + actor = { type: 'cron', id: COMPLETE_INVOICE_LINES_SOURCE }, + } = options const result: CompleteInvoiceLinesResult = { candidates: 0, @@ -269,6 +270,7 @@ export async function completeMigratedInvoiceLines( notHydrated: 0, vatUnresolved: 0, failed: 0, + historyAppended: 0, remaining: 0, hydration: { needed: 0, hydrated: 0, failed: 0, skippedForBudget: 0 }, dryRun, @@ -355,7 +357,7 @@ export async function completeMigratedInvoiceLines( } } - let header: HeaderFill | null = null + let header: InvoiceHeaderVatSplit | null = null if (mapped.vatUnresolved) { result.vatUnresolved++ } else if (headerHoldsNoVatEvidence(row)) { @@ -381,36 +383,55 @@ export async function completeMigratedInvoiceLines( return result } + // One run, one correlation id: the invoices this run completed read as + // one thread in the behandlingshistorik. + const trail: CompleteInvoiceRowsTrail = { + source: COMPLETE_INVOICE_LINES_SOURCE, + provider, + consentId, + correlationId: crypto.randomUUID(), + actor, + } + for (const plan of planned) { // One call per invoice, so a bad row set rejects its own invoice and not // the hundred beside it. A concurrent run (the wizard and the cron, or // two crons overlapping) that filled this invoice since the candidates - // were loaded leaves this call with wrote = false: rows are appended, - // never replaced, and the RPC's invoice lock is what keeps a second - // writer from doubling them. - const { data, error } = await supabase.rpc('complete_invoice_rows', { - p_company_id: companyId, - p_invoice_id: plan.row.id, - p_rows: plan.items, - p_header: plan.header, + // were loaded leaves this call already_filled: rows are appended, never + // replaced, and the RPC's invoice lock is what keeps a second writer + // from doubling them. The trail event is written only for a write that + // landed, by completeInvoiceRows itself. + const outcome = await completeInvoiceRows(supabase, { + companyId, + invoiceId: plan.row.id, + rows: plan.items, + header: plan.header, + headerBefore: { + subtotal: plan.row.subtotal, + vat_amount: plan.row.vat_amount, + vat_rate: plan.row.vat_rate, + vat_treatment: plan.row.vat_treatment, + }, + trail, + // The cron's client is the service client already. + historyClient: supabase, }) - const outcome = (data ?? null) as CompleteRowsOutcome | null - if (error || !outcome?.ok) { + if (outcome.status === 'failed') { result.failed++ log.error('complete_invoice_rows failed', { - companyId, invoiceId: plan.row.id, invoiceNumber: plan.row.invoice_number, - reason: error?.message ?? outcome?.code ?? 'empty RPC response', + companyId, invoiceId: plan.row.id, invoiceNumber: plan.row.invoice_number, reason: outcome.reason, }) continue } - if (!outcome.wrote) { + if (outcome.status === 'already_filled') { log.info('invoice gained rows since the candidates were loaded; left as is', { companyId, invoiceId: plan.row.id, }) continue } result.completed++ - if (outcome.header_updated) result.headersUpdated++ + if (outcome.headerUpdated) result.headersUpdated++ + if (outcome.eventId) result.historyAppended++ } result.remaining = candidates.length - result.completed @@ -424,6 +445,7 @@ export async function completeMigratedInvoiceLines( totalMismatch: result.totalMismatch, rowsMismatch: result.rowsMismatch, failed: result.failed, + historyAppended: result.historyAppended, remaining: result.remaining, hydration: result.hydration, }) diff --git a/extensions/general/arcim-migration/lib/migration-orchestrator.ts b/extensions/general/arcim-migration/lib/migration-orchestrator.ts index 8d383ee7..d1704065 100644 --- a/extensions/general/arcim-migration/lib/migration-orchestrator.ts +++ b/extensions/general/arcim-migration/lib/migration-orchestrator.ts @@ -40,6 +40,7 @@ import { fetchAllRows } from '@/lib/supabase/fetch-all' import { suggestPartiesForCompany } from '@/lib/parties/suggest' import { createLogger } from '@/lib/logger' import { reconcileSupplierInvoiceVouchers } from '@/lib/invoices/bulk-reconcile-supplier-vouchers' +import { completeInvoiceRows, type CompleteInvoiceRowsTrail } from '@/lib/invoices/complete-invoice-rows' import { linkMigratedRegistrationVouchers, type MigratedInvoiceLinkInput, @@ -69,6 +70,14 @@ export interface MigrationOptions { companyId: string userId: string supabase: SupabaseClient + /** + * Service-role client for the behandlingshistorik rows the sales-invoice + * step writes (processing_history has no INSERT policy, and `supabase` is + * the user's session client). Resolved once, and only when there are rows + * to write, so a run that imports nothing invoice-shaped never builds it. + * See lib/invoices/complete-invoice-rows.ts. + */ + createHistoryClient: () => Promise> importCompanyInfo?: boolean importCustomers?: boolean importSuppliers?: boolean @@ -88,6 +97,8 @@ export interface MigrationOptions { const INSERT_CHUNK_SIZE = 500 /** Sales-invoice row writes in flight at once (one complete_invoice_rows call per invoice). */ const ITEM_RPC_CONCURRENCY = 8 +/** The writer named in the behandlingshistorik event for every invoice's rows. */ +export const MIGRATION_WIZARD_SOURCE = 'migration-wizard' const ENRICHMENT_CONCURRENCY = 10 function emitProgress(options: MigrationOptions, progress: MigrationProgress) { @@ -210,6 +221,9 @@ function logFxUnresolved(kind: string, invoiceNumber: string, fx: FxUnresolved): export async function executeMigration(options: MigrationOptions): Promise { const { consentId, companyId, userId, supabase } = options const results: MigrationResults = {} + // One id per run: every InvoiceRowsCompleted event this import writes + // shares it, so the import reads as one thread in the behandlingshistorik. + const runId = crypto.randomUUID() // What this run has proven about the grant, read by recordStepError: a 403 // once a call has already succeeded is one closed register, not a dead token. const runState: ProviderRunState = { grantProven: false } @@ -779,22 +793,36 @@ export async function executeMigration(options: MigrationOptions): Promise { - const { data, error: itemErr } = await supabase.rpc('complete_invoice_rows', { - p_company_id: companyId, - p_invoice_id: invoiceId, - p_rows: rows, - }) - const rpcOutcome = (data ?? null) as { ok?: boolean; code?: string } | null - if (itemErr || !rpcOutcome?.ok) { - console.error( - `[migration] Sales invoice items insert failed for ${invoiceId}:`, - itemErr?.message ?? rpcOutcome?.code ?? 'empty RPC response', - ) - } - })) + // the wizard's clock. completeInvoiceRows also writes the + // InvoiceRowsCompleted event for every invoice the RPC filled, the + // same event the pass writes, so the two writers reconcile per + // invoice (BFNAR 2013:2 p. 9.16). + if (rowsByInvoice.length > 0) { + const historyClient = await options.createHistoryClient() + const trail: CompleteInvoiceRowsTrail = { + source: MIGRATION_WIZARD_SOURCE, + provider, + consentId, + correlationId: runId, + actor: { type: 'user', id: userId }, + } + for (const group of chunk(rowsByInvoice, ITEM_RPC_CONCURRENCY)) { + await Promise.all(group.map(async ({ invoiceId, rows }) => { + const outcome = await completeInvoiceRows(supabase, { + companyId, + invoiceId, + rows, + trail, + historyClient, + }) + if (outcome.status === 'failed') { + console.error( + `[migration] Sales invoice items insert failed for ${invoiceId}:`, + outcome.reason, + ) + } + })) + } } } diff --git a/lib/invoices/__tests__/complete-invoice-rows.test.ts b/lib/invoices/__tests__/complete-invoice-rows.test.ts new file mode 100644 index 00000000..debbe1e8 --- /dev/null +++ b/lib/invoices/__tests__/complete-invoice-rows.test.ts @@ -0,0 +1,204 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest' +import { PROCESSING_EVENT_TYPES } from '@/lib/processing-history/append' +import { + completeInvoiceRows, + INVOICE_ROWS_COMPLETED_EVENT, + type CompleteInvoiceRowsTrail, +} from '../complete-invoice-rows' + +/** + * The one call site for complete_invoice_rows and the one emitter of the + * InvoiceRowsCompleted behandlingshistorik event (#2312). The append runs for + * real against a spy client: what lands in processing_history is the row + * shape and the PII guard of appendProcessingHistoryWithClient, not a mock's + * idea of it. The RPC is a spy. + */ + +vi.mock('@/lib/supabase/server', () => ({ + createClient: vi.fn(), + createServiceClient: vi.fn(), +})) + +const COMPANY = '11111111-1111-4111-8111-111111111111' +const INVOICE = '22222222-2222-4222-8222-222222222222' +const CONSENT = '33333333-3333-4333-8333-333333333333' +const RUN = '44444444-4444-4444-8444-444444444444' + +const ROWS = [ + { sort_order: 1, description: 'Konsulttid', quantity: 8, unit: 'h', unit_price: 100, line_total: 800, vat_rate: 25, vat_amount: 200, line_type: 'product' }, + { sort_order: 2, description: 'Resa', quantity: 1, unit: 'st', unit_price: 200, line_total: 200, vat_rate: 25, vat_amount: 50, line_type: 'product' }, +] + +const HEADER = { + subtotal: 1000, + subtotal_sek: 1000, + vat_amount: 250, + vat_amount_sek: 250, + vat_rate: 25, + vat_treatment: 'standard_25', +} + +/** The pre-#1745 shape: 25 % label beside 0 kr VAT and subtotal = total. */ +const BEFORE = { subtotal: 1250, vat_amount: 0, vat_rate: 25, vat_treatment: 'standard_25' } + +const trail: CompleteInvoiceRowsTrail = { + source: 'complete-invoice-lines', + provider: 'fortnox', + consentId: CONSENT, + correlationId: RUN, + actor: { type: 'cron', id: 'complete-invoice-lines' }, +} + +function rpcClient(reply: { data?: unknown; error?: { message: string } | null }) { + const rpc = vi.fn().mockResolvedValue({ data: reply.data ?? null, error: reply.error ?? null }) + return { client: { rpc } as never, rpc } +} + +function historyClient(error: { message: string } | null = null) { + const insert = vi.fn().mockResolvedValue({ error }) + const from = vi.fn().mockReturnValue({ insert }) + return { client: { from } as never, from, insert } +} + +const wrote = (rows: number, headerUpdated: boolean) => ({ + data: { ok: true, wrote: true, rows, header_updated: headerUpdated }, +}) + +describe('completeInvoiceRows', () => { + beforeEach(() => { + vi.clearAllMocks() + }) + + it('names an event type the catalog registers', () => { + // The union already refuses an unregistered literal at compile time; + // this pins the runtime list the pg test reads against the migration. + expect(PROCESSING_EVENT_TYPES).toContain(INVOICE_ROWS_COMPLETED_EVENT) + }) + + it('writes through the RPC and appends one InvoiceRowsCompleted on the invoice, with the split before and after', async () => { + const { client: supabase, rpc } = rpcClient(wrote(2, true)) + const history = historyClient() + + const result = await completeInvoiceRows(supabase, { + companyId: COMPANY, + invoiceId: INVOICE, + rows: ROWS, + header: HEADER, + headerBefore: BEFORE, + trail, + historyClient: history.client, + }) + + expect(rpc).toHaveBeenCalledTimes(1) + expect(rpc).toHaveBeenCalledWith('complete_invoice_rows', { + p_company_id: COMPANY, + p_invoice_id: INVOICE, + p_rows: ROWS, + p_header: HEADER, + }) + + expect(history.from).toHaveBeenCalledWith('processing_history') + expect(history.insert).toHaveBeenCalledTimes(1) + const row = history.insert.mock.calls[0][0] as Record + expect(row).toMatchObject({ + company_id: COMPANY, + correlation_id: RUN, + aggregate_type: 'Invoice', + aggregate_id: INVOICE, + event_type: 'InvoiceRowsCompleted', + actor: { type: 'cron', id: 'complete-invoice-lines' }, + payload_schema_version: 1, + }) + expect(row.payload).toEqual({ + source: 'complete-invoice-lines', + provider: 'fortnox', + consent_id: CONSENT, + rows: 2, + header_updated: true, + header_before: { subtotal: 1250, vat_amount: 0, vat_rate: 25, vat_treatment: 'standard_25' }, + header_after: { subtotal: 1000, vat_amount: 250, vat_rate: 25, vat_treatment: 'standard_25' }, + }) + // The SEK twins are derived, not evidence: the trail leaves them out. + expect(row.payload).not.toHaveProperty('header_after.subtotal_sek') + + expect(result).toEqual({ status: 'written', rows: 2, headerUpdated: true, eventId: row.event_id }) + expect(typeof result.status === 'string' && 'eventId' in result && result.eventId).toMatch(/^[0-9a-f-]{36}$/) + }) + + it('records no split when the header was left alone (the wizard path, or evidence already there)', async () => { + const { client: supabase, rpc } = rpcClient(wrote(2, false)) + const history = historyClient() + + const result = await completeInvoiceRows(supabase, { + companyId: COMPANY, + invoiceId: INVOICE, + rows: ROWS, + trail: { ...trail, source: 'migration-wizard', actor: { type: 'user', id: '55555555-5555-4555-8555-555555555555' } }, + historyClient: history.client, + }) + + // No header means an explicit null to the RPC, never a dropped argument. + expect(rpc.mock.calls[0][1]).toMatchObject({ p_header: null }) + const row = history.insert.mock.calls[0][0] as Record + expect(row.payload).toEqual({ + source: 'migration-wizard', + provider: 'fortnox', + consent_id: CONSENT, + rows: 2, + header_updated: false, + header_before: null, + header_after: null, + }) + expect(row.actor).toEqual({ type: 'user', id: '55555555-5555-4555-8555-555555555555' }) + expect(result).toMatchObject({ status: 'written', rows: 2, headerUpdated: false }) + }) + + it('records nothing when another writer filled the invoice first', async () => { + const { client: supabase } = rpcClient({ data: { ok: true, wrote: false, rows: 0, header_updated: false } }) + const history = historyClient() + + const result = await completeInvoiceRows(supabase, { + companyId: COMPANY, invoiceId: INVOICE, rows: ROWS, header: HEADER, headerBefore: BEFORE, trail, + historyClient: history.client, + }) + + expect(result).toEqual({ status: 'already_filled' }) + expect(history.insert).not.toHaveBeenCalled() + }) + + it('records nothing when the RPC errors or refuses', async () => { + const history = historyClient() + + const errored = rpcClient({ data: null, error: { message: 'check violation' } }) + await expect(completeInvoiceRows(errored.client, { + companyId: COMPANY, invoiceId: INVOICE, rows: ROWS, trail, historyClient: history.client, + })).resolves.toEqual({ status: 'failed', reason: 'check violation' }) + + const refused = rpcClient({ data: { ok: false, code: 'MISSING_REQUIRED', details: { column: 'vat_rate' } } }) + await expect(completeInvoiceRows(refused.client, { + companyId: COMPANY, invoiceId: INVOICE, rows: ROWS, trail, historyClient: history.client, + })).resolves.toEqual({ status: 'failed', reason: 'MISSING_REQUIRED' }) + + const empty = rpcClient({ data: null }) + await expect(completeInvoiceRows(empty.client, { + companyId: COMPANY, invoiceId: INVOICE, rows: ROWS, trail, historyClient: history.client, + })).resolves.toEqual({ status: 'failed', reason: 'empty RPC response' }) + + expect(history.insert).not.toHaveBeenCalled() + }) + + it('reports a write whose trail append failed as written, with no event id', async () => { + // The rows are committed by then. Failing the invoice would make the + // next run try again and find it full; the gap is logged instead. + const { client: supabase } = rpcClient(wrote(2, true)) + const history = historyClient({ message: 'insert or update on table "processing_history" violates foreign key constraint' }) + + const result = await completeInvoiceRows(supabase, { + companyId: COMPANY, invoiceId: INVOICE, rows: ROWS, header: HEADER, headerBefore: BEFORE, trail, + historyClient: history.client, + }) + + expect(history.insert).toHaveBeenCalledTimes(1) + expect(result).toEqual({ status: 'written', rows: 2, headerUpdated: true, eventId: null }) + }) +}) diff --git a/lib/invoices/complete-invoice-rows.ts b/lib/invoices/complete-invoice-rows.ts new file mode 100644 index 00000000..e0ea353b --- /dev/null +++ b/lib/invoices/complete-invoice-rows.ts @@ -0,0 +1,189 @@ +/** + * completeInvoiceRows: the one TypeScript call site for the + * complete_invoice_rows RPC (migration 20260906135730), and the one place the + * behandlingshistorik event for a completed migrated invoice is written. + * + * Two writers put rows under migrated sales invoices: the migration wizard + * (extensions/general/arcim-migration/lib/migration-orchestrator.ts, rows + * written milliseconds after the header) and the hourly row-completion pass + * (complete-invoice-lines.ts, the rows the wizard's hydration budget did not + * reach, plus the header VAT split when the stored one held no evidence). + * The RPC already gives them one write path; this wrapper gives them one + * trail. Every invoice whose rows the RPC wrote gets one InvoiceRowsCompleted + * event (BFL 5 kap 11 §, BFNAR 2013:2 p. 9.16: the behandlingshistorik has to + * say what was processed automatically, when, and by what) naming the writer, + * the provider the rows came from, the row count and, when the header split + * was rewritten, the split before and after. The pass writes no + * bokföringspost, so BFL 5 kap 5 § (rättelse) does not bind it; the migration + * that wrote the invoice header records no event of its own, so this event + * is what lets the two writers reconcile per invoice. + * + * Failure semantics: the event is written only after the RPC has answered + * wrote = true, so an invoice whose completion failed, or that another writer + * had already filled, gets no event. The append itself is best-effort, the + * convention every processing_history writer follows (the rows are + * committed; a missing change-log row is logged, not turned into a failed + * invoice that the next run would try again and find full). The caller sees + * a null eventId when that happened. + * + * PII boundary: the payload carries UUIDs, counts, amounts and enum strings + * only. Invoice numbers and provider document numbers are deliberately left + * out: a ten-digit number (2026090001) trips the personnummer guard in + * appendProcessingHistory, which would lose the event for exactly that + * invoice. The invoice id is the reference; its number is on the row. + */ + +import type { SupabaseClient } from '@supabase/supabase-js' +import type { ProcessingHistoryActor } from '@/types' +import { createLogger } from '@/lib/logger' +import { + appendProcessingHistoryWithClient, + type ProcessingHistoryEventType, +} from '@/lib/processing-history/append' + +const log = createLogger('invoices/complete-invoice-rows') + +/** Registered in processing_event_types by migration 20260906210100. */ +export const INVOICE_ROWS_COMPLETED_EVENT = 'InvoiceRowsCompleted' satisfies ProcessingHistoryEventType + +type RpcClient = Pick +type HistoryClient = Pick + +/** The six invoice columns the RPC may rewrite: all present or none. */ +export interface InvoiceHeaderVatSplit { + subtotal: number + subtotal_sek: number | null + vat_amount: number + vat_amount_sek: number | null + vat_rate: number | null + vat_treatment: string +} + +/** + * What the trail records of a header split. The SEK twins are left out: + * they are derived from these and the exchange rate the row already carries. + */ +export interface InvoiceHeaderVatSnapshot { + subtotal: number | null + vat_amount: number | null + vat_rate: number | null + vat_treatment: string | null +} + +export interface CompleteInvoiceRowsTrail { + /** Which writer: 'migration-wizard' or 'complete-invoice-lines'. */ + source: string + /** The provider the rows came from ('fortnox', 'briox', ...). */ + provider: string + /** The provider consent the rows were fetched under. */ + consentId: string + /** One id per run, shared by every invoice the run completed. */ + correlationId: string + actor: ProcessingHistoryActor +} + +export interface CompleteInvoiceRowsInput { + companyId: string + invoiceId: string + /** The invoice_items columns per row; the RPC stamps invoice_id itself. */ + rows: Record[] + /** The header split to apply in the same transaction, or null to leave the header alone. */ + header?: InvoiceHeaderVatSplit | null + /** The stored split before the write; recorded beside the new one when the header is rewritten. */ + headerBefore?: InvoiceHeaderVatSnapshot | null + trail: CompleteInvoiceRowsTrail + /** + * The client the event row is written with. processing_history has no + * INSERT policy, so this must be a service-role client: the cron's own + * client already is, the wizard (on the user's session client) passes one. + */ + historyClient: HistoryClient +} + +export type CompleteInvoiceRowsResult = + /** The rows (and the header, when one was given) landed; eventId is null when the trail append failed. */ + | { status: 'written'; rows: number; headerUpdated: boolean; eventId: string | null } + /** Another writer filled the invoice first; nothing was written and nothing is recorded. */ + | { status: 'already_filled' } + /** The RPC errored or refused (its code, or the Postgres message). */ + | { status: 'failed'; reason: string } + +/** What complete_invoice_rows returns (migration 20260906135730). */ +interface CompleteRowsRpcOutcome { + ok: boolean + code?: string + wrote?: boolean + rows?: number + header_updated?: boolean +} + +function snapshotOf(split: InvoiceHeaderVatSnapshot | InvoiceHeaderVatSplit | null | undefined): InvoiceHeaderVatSnapshot | null { + if (!split) return null + return { + subtotal: split.subtotal, + vat_amount: split.vat_amount, + vat_rate: split.vat_rate, + vat_treatment: split.vat_treatment, + } +} + +export async function completeInvoiceRows( + supabase: RpcClient, + input: CompleteInvoiceRowsInput, +): Promise { + const header = input.header ?? null + const { data, error } = await supabase.rpc('complete_invoice_rows', { + p_company_id: input.companyId, + p_invoice_id: input.invoiceId, + p_rows: input.rows, + p_header: header, + }) + const outcome = (data ?? null) as CompleteRowsRpcOutcome | null + if (error || !outcome?.ok) { + return { status: 'failed', reason: error?.message ?? outcome?.code ?? 'empty RPC response' } + } + if (!outcome.wrote) return { status: 'already_filled' } + + const rows = outcome.rows ?? input.rows.length + const headerUpdated = outcome.header_updated === true + const eventId = await appendCompletedEvent(input, rows, headerUpdated) + return { status: 'written', rows, headerUpdated, eventId } +} + +async function appendCompletedEvent( + input: CompleteInvoiceRowsInput, + rows: number, + headerUpdated: boolean, +): Promise { + const { trail } = input + try { + return await appendProcessingHistoryWithClient(input.historyClient, { + companyId: input.companyId, + correlationId: trail.correlationId, + aggregateType: 'Invoice', + aggregateId: input.invoiceId, + eventType: INVOICE_ROWS_COMPLETED_EVENT, + payload: { + source: trail.source, + provider: trail.provider, + consent_id: trail.consentId, + rows, + header_updated: headerUpdated, + header_before: headerUpdated ? snapshotOf(input.headerBefore) : null, + header_after: headerUpdated ? snapshotOf(input.header) : null, + }, + actor: trail.actor, + occurredAt: new Date(), + }) + } catch (err) { + // The rows are committed; the trail row is what is missing. Logged so + // the gap is visible, never a reason to report the invoice as failed. + log.error('InvoiceRowsCompleted append failed; rows written without their behandlingshistorik row', { + companyId: input.companyId, + invoiceId: input.invoiceId, + source: trail.source, + error: err instanceof Error ? err.message : String(err), + }) + return null + } +} diff --git a/lib/processing-history/append.ts b/lib/processing-history/append.ts index 96b15cab..90517b41 100644 --- a/lib/processing-history/append.ts +++ b/lib/processing-history/append.ts @@ -60,6 +60,7 @@ export const PROCESSING_EVENT_TYPES = [ 'InvoiceDuplicatePaymentDismissed', 'InvoiceJournalEntrySkipped', 'InvoicePaymentRowBackfilled', + 'InvoiceRowsCompleted', 'OAuthClientRevoked', 'PendingOperationApproved', 'PendingOperationRejected', diff --git a/supabase/migrations/20260906210100_invoice_rows_completed_event.sql b/supabase/migrations/20260906210100_invoice_rows_completed_event.sql new file mode 100644 index 00000000..c833db10 --- /dev/null +++ b/supabase/migrations/20260906210100_invoice_rows_completed_event.sql @@ -0,0 +1,54 @@ +-- InvoiceRowsCompleted: the behandlingshistorik event for a migrated sales +-- invoice whose rows were written by complete_invoice_rows (migration +-- 20260906135730), from either of its writers: the migration wizard or the +-- hourly row-completion pass (#2291, #2312). +-- +-- The pass inserts invoice_items and rewrites the header VAT split on +-- invoices whose stored split held no evidence. It writes no bokföringspost, +-- so BFL 5 kap 5 § (rättelse) does not bind it, but it is automated +-- processing of räkenskapsinformation, which BFL 5 kap 11 § and BFNAR 2013:2 +-- p. 9.16 want in the behandlingshistorik: what was processed, when, and by +-- what. Until now the only trail was a log line per invoice in Vercel, and +-- the migration that wrote the invoices in the first place records no event +-- either. +-- +-- One event per completed invoice, emitted by lib/invoices/complete-invoice- +-- rows.ts (the one TypeScript call site for the RPC, so a writer cannot reach +-- the RPC without the trail). Payload: the writer, the provider, the consent, +-- the row count, and the header split before and after when it was +-- rewritten. UUIDs, counts, amounts and enum strings only. +-- +-- The aggregate is the invoice, which the aggregate_type CHECK did not admit +-- (no invoice-level event existed before). 'Invoice' is added the way +-- 20260423140500 added the AI streams; every value already in the constraint +-- stays. The catalog row goes in as for every other event type: +-- processing_history.event_type has an FK to it and the append is +-- best-effort, so an unregistered type would be lost silently +-- (lib/processing-history/append.ts, tests/pg/processing-event-types.pg.test.ts). +-- +-- pg-test: tests/pg/invoice-rows-completed-event.pg.test.ts + +ALTER TABLE public.processing_history + DROP CONSTRAINT IF EXISTS processing_history_aggregate_type_check; + +ALTER TABLE public.processing_history + ADD CONSTRAINT processing_history_aggregate_type_check + CHECK (aggregate_type IN ( + 'Document', + 'BankTransaction', + 'MatchProposal', + 'Verifikation', + 'CounterpartyTemplate', + 'Period', + 'Migration', + 'System', + 'AIProposal', + 'AIRequest', + 'Invoice' + )); + +INSERT INTO public.processing_event_types (event_type) +VALUES ('InvoiceRowsCompleted') +ON CONFLICT (event_type) DO NOTHING; + +NOTIFY pgrst, 'reload schema'; diff --git a/tests/pg/invoice-rows-completed-event.pg.test.ts b/tests/pg/invoice-rows-completed-event.pg.test.ts new file mode 100644 index 00000000..4f01c5f3 --- /dev/null +++ b/tests/pg/invoice-rows-completed-event.pg.test.ts @@ -0,0 +1,66 @@ +import { describe, it, expect } from 'vitest' +import { randomUUID } from 'node:crypto' +import { getPool } from './setup' +import { seedCompany } from './fixtures' + +// pg-real coverage for 20260906210100_invoice_rows_completed_event: the +// behandlingshistorik event lib/invoices/complete-invoice-rows.ts writes for +// every migrated invoice whose rows complete_invoice_rows filled (#2312). +// The type must be in the catalog (the append is best-effort, so an +// unregistered type is a silently lost record) and the invoice must be an +// admitted aggregate (the CHECK did not know 'Invoice' before). + +describe('migration 20260906210100: InvoiceRowsCompleted event', () => { + it('registers InvoiceRowsCompleted in the processing history catalog', async () => { + const { rows } = await getPool().query( + `SELECT event_type + FROM public.processing_event_types + WHERE event_type = 'InvoiceRowsCompleted'`, + ) + + expect(rows).toEqual([{ event_type: 'InvoiceRowsCompleted' }]) + }) + + it('accepts the event on the Invoice aggregate, shaped the way the emitter writes it', async () => { + const { companyId } = await seedCompany() + const invoiceId = randomUUID() + const runId = randomUUID() + const { rows } = await getPool().query<{ aggregate_type: string; event_type: string }>( + `INSERT INTO public.processing_history + (company_id, correlation_id, aggregate_type, aggregate_id, event_type, payload, actor, occurred_at) + VALUES ($1, $2, 'Invoice', $3, 'InvoiceRowsCompleted', + $4::jsonb, '{"type":"cron","id":"complete-invoice-lines"}', now()) + RETURNING aggregate_type, event_type`, + [ + companyId, + runId, + invoiceId, + JSON.stringify({ + source: 'complete-invoice-lines', + provider: 'fortnox', + consent_id: randomUUID(), + rows: 2, + header_updated: true, + header_before: { subtotal: 1250, vat_amount: 0, vat_rate: 25, vat_treatment: 'standard_25' }, + header_after: { subtotal: 1000, vat_amount: 250, vat_rate: 25, vat_treatment: 'standard_25' }, + }), + ], + ) + + expect(rows).toEqual([{ aggregate_type: 'Invoice', event_type: 'InvoiceRowsCompleted' }]) + }) + + it('still refuses an aggregate the constraint does not name: the CHECK was widened, not dropped', async () => { + const { companyId } = await seedCompany() + const id = randomUUID() + await expect( + getPool().query( + `INSERT INTO public.processing_history + (company_id, correlation_id, aggregate_type, aggregate_id, event_type, payload, actor, occurred_at) + VALUES ($1, $2, 'Kitten', $2, 'InvoiceRowsCompleted', '{}'::jsonb, + '{"type":"system","id":"invoice-rows-completed-test"}', now())`, + [companyId, id], + ), + ).rejects.toMatchObject({ code: '23514' }) + }) +}) diff --git a/types/index.ts b/types/index.ts index 14d49e9a..2cb6f0ef 100644 --- a/types/index.ts +++ b/types/index.ts @@ -293,6 +293,7 @@ export type ProcessingHistoryAggregateType = | 'Period' | 'Migration' | 'System' + | 'Invoice' // Bank connection status // 'pending_selection' = PSD2 consent granted, awaiting user to pick which