feat(compliance): InvoiceRowsCompleted behandlingshistorik event for migrated invoice rows (#2312) (#2357)
Every migrated sales invoice whose rows complete_invoice_rows writes, from the migration wizard or the hourly row-completion pass, now leaves one InvoiceRowsCompleted row in processing_history on a new Invoice aggregate: the writer, the provider, the consent, the row count, and the header VAT split before and after when the pass rewrote it (BFL 5 kap 11 §, BFNAR 2013:2 p. 9.16). One run shares one correlation id. lib/invoices/complete-invoice-rows.ts is the one TypeScript call site for the RPC and the one emitter: it appends only on wrote = true, records nothing for already_filled or failed, and keeps the append best-effort (logged, eventId null) like every other processing_history writer. The wizard runs on the user's session client, so MigrationOptions takes a lazy createHistoryClient for the service role. Invoice numbers stay out of the payload (the personnummer guard would drop ten-digit ones). Migration 20260906210100 widens the aggregate_type CHECK with Invoice and registers the event type; pg test covers the catalog row, the aggregate, and that the CHECK still refuses unknown aggregates. Claude-Session: https://claude.ai/code/session_01LvMaHcTnwAfxzgYD1fGYX1 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:
co-authored by
Jakob Wennberg
Claude Fable 5.1
parent
ebbe50c0f3
commit
6906bc4aa2
@@ -114,6 +114,7 @@ function baseOptions(overrides: Record<string, unknown> = {}) {
|
||||
companyId: 'company-1',
|
||||
userId: 'user-1',
|
||||
supabase: supabase as unknown as SupabaseClient,
|
||||
createHistoryClient: async () => ({ from: vi.fn() }) as unknown as Pick<SupabaseClient, 'from'>,
|
||||
importCompanyInfo: false,
|
||||
importCustomers: false,
|
||||
importSuppliers: false,
|
||||
|
||||
@@ -84,6 +84,7 @@ function baseOptions(overrides: Record<string, unknown> = {}) {
|
||||
companyId: 'company-1',
|
||||
userId: 'user-1',
|
||||
supabase: {} as unknown as SupabaseClient,
|
||||
createHistoryClient: async () => ({ from: vi.fn() }) as unknown as Pick<SupabaseClient, 'from'>,
|
||||
importCompanyInfo: false,
|
||||
importCustomers: false,
|
||||
importSuppliers: false,
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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<string, unknown> = {}) {
|
||||
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<string, unknown>[] {
|
||||
.filter((h): h is Record<string, unknown> => h !== null)
|
||||
}
|
||||
|
||||
/** The processing_history rows the run appended, in order. */
|
||||
function historyRows(calls: Call[]): Record<string, unknown>[] {
|
||||
return calls
|
||||
.filter((c) => c.table === 'processing_history' && c.method === 'insert')
|
||||
.map((c) => c.args[0] as Record<string, unknown>)
|
||||
}
|
||||
|
||||
/** 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)
|
||||
})
|
||||
|
||||
|
||||
@@ -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<string, unknown>[]
|
||||
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<CompleteInvoiceLinesResult> {
|
||||
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,
|
||||
})
|
||||
|
||||
@@ -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<Pick<SupabaseClient, 'from'>>
|
||||
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<MigrationResults> {
|
||||
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<Migra
|
||||
// an invoice's rows are written at most once whichever writer gets
|
||||
// there first, and a bad row set rejects its own invoice rather than
|
||||
// the whole chunk. Small concurrent groups keep the round trips off
|
||||
// the wizard's clock.
|
||||
for (const group of chunk(rowsByInvoice, ITEM_RPC_CONCURRENCY)) {
|
||||
await Promise.all(group.map(async ({ invoiceId, rows }) => {
|
||||
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,
|
||||
)
|
||||
}
|
||||
}))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user