fix(invoices): write migrated invoice rows through one locking RPC so two writers cannot double them (#2313) (#2340)
* fix(invoices): write migrated invoice rows through one locking RPC so two writers cannot double them The row-completion pass (#2291) and the migration wizard both wrote invoice_items for migrated sales invoices with check-then-insert across separate statements and nothing serializing them per invoice; the pass also wrote the header VAT split in a third statement, so "rows landed, header did not" was reachable and never revisited. Adds complete_invoice_rows (SECURITY DEFINER, FOR UPDATE on the invoice scoped to the company, inserts only when the invoice still has no rows, optional header split in the same transaction, returns wrote) and routes both writers through it: the pass one call per invoice (wrote = false is skipped, not completed), the wizard one call per invoice in small concurrent groups. Unknown row keys and partial headers are refused rather than dropped. Grants: revoked from PUBLIC and anon, kept for authenticated (membership gate in the body) and service_role. pg test proves the invariant (first call writes, second returns wrote = false with rows and header unchanged), the rollback of rows on a failing header, every refusal, the grants and two-connection serialization. Closes #2313 Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019SaJfqNi4VmsG8FMKq99G6 * fix(invoices): complete_invoice_rows requires the row's tax facts instead of defaulting vat_rate to 25 Review finding on #2340: COALESCE(r.vat_rate, 25) let a row without a rate land with a fabricated 25 % (ML 17 kap 24 § p.9). Both writers always send vat_rate, line_total, vat_amount and description, so the defaults were never needed and only hid a bug. The RPC now refuses a row missing any of the four (absent or JSON null) with MISSING_REQUIRED naming the column; sort_order, quantity, unit and line_type keep their table defaults since none states a tax fact. The rate's value is deliberately not restricted to the Swedish set: 0 (omvänd skattskyldighet, export) and foreign rates (OSS) are legitimate on a migrated row, and the pg test pins both as accepted. Migration edited in place: unshipped, preview branches only. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.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
9cb1d105e3
commit
39d409d257
@@ -8,7 +8,9 @@ import type { SalesInvoiceDto } from '@/lib/providers/dto'
|
||||
* migration's detail hydration is budget-bounded). It must start from OUR
|
||||
* row-less invoices, join strictly, hydrate only that subset, write rows only
|
||||
* when the provider's total agrees with the stored one, and leave anything it
|
||||
* could not reach for the next run rather than guessing.
|
||||
* 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.
|
||||
*/
|
||||
|
||||
vi.mock('@/lib/providers/resolve-consent', () => ({
|
||||
@@ -102,11 +104,33 @@ function hydratedAll(invoices: SalesInvoiceDto[], unhydratedIds: string[] = [])
|
||||
|
||||
interface Call { table: string; method: string; args: unknown[] }
|
||||
|
||||
/** The complete_invoice_rows payload, as the pass sends it. */
|
||||
interface RpcArgs {
|
||||
p_company_id: string
|
||||
p_invoice_id: string
|
||||
p_rows: Record<string, unknown>[]
|
||||
p_header: Record<string, unknown> | null
|
||||
}
|
||||
|
||||
/** What the real RPC answers a caller that wrote: N rows, header iff one was sent. */
|
||||
function rpcWrote(args: RpcArgs) {
|
||||
return {
|
||||
data: { ok: true, wrote: true, rows: args.p_rows.length, header_updated: args.p_header !== null },
|
||||
error: null,
|
||||
}
|
||||
}
|
||||
|
||||
const rpcAlreadyFilled = { data: { ok: true, wrote: false, rows: 0, header_updated: false }, error: null }
|
||||
|
||||
/**
|
||||
* Thenable query-builder stand-in. Records every call; resolves with what
|
||||
* `respond` returns for the table and the methods used on the chain.
|
||||
* Query-builder and RPC stand-in. Records every call; `respond` answers the
|
||||
* table chains, `respondRpc` answers `.rpc()` (defaults to a successful
|
||||
* write shaped like the real function's reply).
|
||||
*/
|
||||
function makeSupabase(respond: (table: string, methods: string[], calls: Call[]) => unknown) {
|
||||
function makeSupabase(
|
||||
respond: (table: string, methods: string[], calls: Call[]) => unknown,
|
||||
respondRpc: (fn: string, args: RpcArgs) => unknown = (_fn, args) => rpcWrote(args),
|
||||
) {
|
||||
const calls: Call[] = []
|
||||
const from = vi.fn((table: string) => {
|
||||
const chain: Call[] = []
|
||||
@@ -125,21 +149,30 @@ function makeSupabase(respond: (table: string, methods: string[], calls: Call[])
|
||||
.then(resolve, reject)
|
||||
return builder
|
||||
})
|
||||
return { supabase: { from } as unknown as SupabaseClient, calls }
|
||||
const rpc = vi.fn((fn: string, args: RpcArgs) => {
|
||||
calls.push({ table: `rpc:${fn}`, method: 'rpc', args: [fn, args] })
|
||||
return Promise.resolve(respondRpc(fn, args))
|
||||
})
|
||||
return { supabase: { from, rpc } as unknown as SupabaseClient, calls }
|
||||
}
|
||||
|
||||
const ok = { data: [], error: null }
|
||||
|
||||
function insertedRows(calls: Call[]): Record<string, unknown>[] {
|
||||
function writes(calls: Call[]): RpcArgs[] {
|
||||
return calls
|
||||
.filter((c) => c.table === 'invoice_items' && c.method === 'insert')
|
||||
.flatMap((c) => c.args[0] as Record<string, unknown>[])
|
||||
.filter((c) => c.method === 'rpc' && c.args[0] === 'complete_invoice_rows')
|
||||
.map((c) => c.args[1] as RpcArgs)
|
||||
}
|
||||
|
||||
/** The rows sent, each tagged with the invoice the call named. */
|
||||
function insertedRows(calls: Call[]): Record<string, unknown>[] {
|
||||
return writes(calls).flatMap((w) => w.p_rows.map((row) => ({ ...row, invoice_id: w.p_invoice_id })))
|
||||
}
|
||||
|
||||
function headerUpdates(calls: Call[]): Record<string, unknown>[] {
|
||||
return calls
|
||||
.filter((c) => c.table === 'invoices' && c.method === 'update')
|
||||
.map((c) => c.args[0] as Record<string, unknown>)
|
||||
return writes(calls)
|
||||
.map((w) => w.p_header)
|
||||
.filter((h): h is Record<string, unknown> => h !== null)
|
||||
}
|
||||
|
||||
describe('completeMigratedInvoiceLines', () => {
|
||||
@@ -148,7 +181,7 @@ describe('completeMigratedInvoiceLines', () => {
|
||||
mResolve.mockResolvedValue({ consent: { provider: 'fortnox' }, accessToken: 'tok', providerCompanyId: undefined })
|
||||
})
|
||||
|
||||
it('writes the rows and fills a header that held no VAT evidence', async () => {
|
||||
it('writes the rows and fills a header that held no VAT evidence, through one RPC call per invoice', async () => {
|
||||
mFetchAll.mockResolvedValue([storedRow()])
|
||||
const dto = providerInvoice()
|
||||
mList.mockResolvedValue([dto])
|
||||
@@ -166,10 +199,14 @@ describe('completeMigratedInvoiceLines', () => {
|
||||
expect(mHydrate).toHaveBeenCalledTimes(1)
|
||||
expect(mHydrate.mock.calls[0][3]).toEqual([dto])
|
||||
|
||||
const rows = insertedRows(calls)
|
||||
expect(rows).toHaveLength(1)
|
||||
expect(rows[0]).toMatchObject({
|
||||
invoice_id: 'inv-1',
|
||||
// One call, scoped to the company as well as the invoice: the RPC locks
|
||||
// the invoice and stamps invoice_id itself, so the rows carry none.
|
||||
const sent = writes(calls)
|
||||
expect(sent).toHaveLength(1)
|
||||
expect(sent[0]).toMatchObject({ p_company_id: 'co-1', p_invoice_id: 'inv-1' })
|
||||
expect(sent[0].p_rows).toHaveLength(1)
|
||||
expect(sent[0].p_rows[0]).not.toHaveProperty('invoice_id')
|
||||
expect(sent[0].p_rows[0]).toEqual({
|
||||
sort_order: 1,
|
||||
description: 'Konsulttid',
|
||||
quantity: 10,
|
||||
@@ -178,11 +215,9 @@ describe('completeMigratedInvoiceLines', () => {
|
||||
line_total: 1000,
|
||||
vat_rate: 25,
|
||||
vat_amount: 250,
|
||||
line_type: 'product',
|
||||
})
|
||||
|
||||
const headers = headerUpdates(calls)
|
||||
expect(headers).toHaveLength(1)
|
||||
expect(headers[0]).toEqual({
|
||||
expect(sent[0].p_header).toEqual({
|
||||
subtotal: 1000,
|
||||
subtotal_sek: 1000,
|
||||
vat_amount: 250,
|
||||
@@ -190,14 +225,11 @@ describe('completeMigratedInvoiceLines', () => {
|
||||
vat_rate: 25,
|
||||
vat_treatment: 'standard_25',
|
||||
})
|
||||
// Scoped to the company as well as the id: defense in depth on a
|
||||
// service-role client.
|
||||
const update = calls.find((c) => c.table === 'invoices' && c.method === 'update')!
|
||||
const scope = calls.filter((c) => c.table === 'invoices' && c.method === 'eq' && calls.indexOf(c) > calls.indexOf(update))
|
||||
expect(scope.map((c) => c.args)).toEqual([['id', 'inv-1'], ['company_id', 'co-1']])
|
||||
// Never the total, status or payments.
|
||||
expect(Object.keys(headers[0])).not.toContain('total')
|
||||
expect(Object.keys(headers[0])).not.toContain('status')
|
||||
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)
|
||||
})
|
||||
|
||||
it('writes the rows but leaves a header whose split is consistent (momsfri)', async () => {
|
||||
@@ -211,6 +243,7 @@ describe('completeMigratedInvoiceLines', () => {
|
||||
|
||||
expect(result).toMatchObject({ completed: 1, headersUpdated: 0 })
|
||||
expect(insertedRows(calls)).toHaveLength(1)
|
||||
expect(writes(calls)[0].p_header).toBeNull()
|
||||
expect(headerUpdates(calls)).toHaveLength(0)
|
||||
})
|
||||
|
||||
@@ -256,8 +289,7 @@ describe('completeMigratedInvoiceLines', () => {
|
||||
const result = await completeMigratedInvoiceLines({ supabase, companyId: 'co-1', consentId: 'c-1' })
|
||||
|
||||
expect(result).toMatchObject({ matched: 1, completed: 0, rowsMismatch: 1, remaining: 1 })
|
||||
expect(insertedRows(calls)).toHaveLength(0)
|
||||
expect(headerUpdates(calls)).toHaveLength(0)
|
||||
expect(writes(calls)).toHaveLength(0)
|
||||
})
|
||||
|
||||
it('tolerates öresavrundning between the rows and the header', async () => {
|
||||
@@ -288,8 +320,7 @@ describe('completeMigratedInvoiceLines', () => {
|
||||
const result = await completeMigratedInvoiceLines({ supabase, companyId: 'co-1', consentId: 'c-1' })
|
||||
|
||||
expect(result).toMatchObject({ matched: 1, completed: 0, totalMismatch: 1, remaining: 1 })
|
||||
expect(insertedRows(calls)).toHaveLength(0)
|
||||
expect(headerUpdates(calls)).toHaveLength(0)
|
||||
expect(writes(calls)).toHaveLength(0)
|
||||
})
|
||||
|
||||
it('reverses the rows of a kreditfaktura the way the migration does', async () => {
|
||||
@@ -318,7 +349,7 @@ describe('completeMigratedInvoiceLines', () => {
|
||||
const result = await completeMigratedInvoiceLines({ supabase, companyId: 'co-1', consentId: 'c-1' })
|
||||
|
||||
expect(result).toMatchObject({ candidates: 2, matched: 2, completed: 1, notHydrated: 1, remaining: 1 })
|
||||
expect(insertedRows(calls).map((r) => r.invoice_id)).toEqual(['inv-1'])
|
||||
expect(writes(calls).map((w) => w.p_invoice_id)).toEqual(['inv-1'])
|
||||
})
|
||||
|
||||
it('does not join an ambiguous key, and does not hydrate when nothing joined', async () => {
|
||||
@@ -332,7 +363,7 @@ describe('completeMigratedInvoiceLines', () => {
|
||||
|
||||
expect(result).toMatchObject({ candidates: 2, matched: 0, unmatched: 2, completed: 0, remaining: 2 })
|
||||
expect(mHydrate).not.toHaveBeenCalled()
|
||||
expect(insertedRows(calls)).toHaveLength(0)
|
||||
expect(writes(calls)).toHaveLength(0)
|
||||
})
|
||||
|
||||
it('costs one query and no provider call when the company has nothing to complete', async () => {
|
||||
@@ -347,39 +378,49 @@ describe('completeMigratedInvoiceLines', () => {
|
||||
expect(calls).toHaveLength(0)
|
||||
})
|
||||
|
||||
it('skips an invoice that gained rows since the candidates were loaded', async () => {
|
||||
it('counts an invoice that gained rows since the candidates were loaded as skipped, not completed', async () => {
|
||||
// The RPC answers wrote = false: another writer (the wizard, an
|
||||
// overlapping cron) got there first. The header in the payload must not
|
||||
// be counted either; header_updated comes from the RPC, not the plan.
|
||||
mFetchAll.mockResolvedValue([storedRow()])
|
||||
const dto = providerInvoice()
|
||||
mList.mockResolvedValue([dto])
|
||||
mHydrate.mockResolvedValue(hydratedAll([dto]))
|
||||
const { supabase, calls } = makeSupabase((table, methods) =>
|
||||
table === 'invoice_items' && methods.includes('in') ? { data: [{ invoice_id: 'inv-1' }], error: null } : ok,
|
||||
)
|
||||
const { supabase, calls } = makeSupabase(() => ok, () => rpcAlreadyFilled)
|
||||
|
||||
const result = await completeMigratedInvoiceLines({ supabase, companyId: 'co-1', consentId: 'c-1' })
|
||||
|
||||
expect(result).toMatchObject({ completed: 0, failed: 0, remaining: 1 })
|
||||
expect(insertedRows(calls)).toHaveLength(0)
|
||||
expect(headerUpdates(calls)).toHaveLength(0)
|
||||
expect(result).toMatchObject({ completed: 0, headersUpdated: 0, failed: 0, remaining: 1 })
|
||||
expect(writes(calls)).toHaveLength(1)
|
||||
expect(writes(calls)[0].p_header).not.toBeNull()
|
||||
})
|
||||
|
||||
it('retries per invoice when the batch insert fails, and counts the offender', async () => {
|
||||
mFetchAll.mockResolvedValue([storedRow(), storedRow({ id: 'inv-2', invoice_number: '1002' })])
|
||||
const first = providerInvoice()
|
||||
const second = providerInvoice({ id: '1002', invoiceNumber: '1002' })
|
||||
mList.mockResolvedValue([first, second])
|
||||
mHydrate.mockResolvedValue(hydratedAll([first, second]))
|
||||
const { supabase, calls } = makeSupabase((table, methods, chain) => {
|
||||
if (table !== 'invoice_items' || !methods.includes('insert')) return ok
|
||||
const rows = chain[0].args[0] as { invoice_id: string }[]
|
||||
if (rows.length > 1) return { data: null, error: { message: 'batch rejected' } }
|
||||
return rows[0].invoice_id === 'inv-2' ? { data: null, error: { message: 'check violation' } } : ok
|
||||
it('counts a refused or failed call as failed and goes on with the rest', async () => {
|
||||
mFetchAll.mockResolvedValue([
|
||||
storedRow(),
|
||||
storedRow({ id: 'inv-2', invoice_number: '1002' }),
|
||||
storedRow({ id: 'inv-3', invoice_number: '1003' }),
|
||||
])
|
||||
const dtos = [
|
||||
providerInvoice(),
|
||||
providerInvoice({ id: '1002', invoiceNumber: '1002' }),
|
||||
providerInvoice({ id: '1003', invoiceNumber: '1003' }),
|
||||
]
|
||||
mList.mockResolvedValue(dtos)
|
||||
mHydrate.mockResolvedValue(hydratedAll(dtos))
|
||||
const { supabase, calls } = makeSupabase(() => ok, (_fn, args) => {
|
||||
// A Postgres error (a check violation inside the RPC) and a refusal
|
||||
// the function answers with ok = false: both leave the invoice for a
|
||||
// human to look at, neither stops the run.
|
||||
if (args.p_invoice_id === 'inv-2') return { data: null, error: { message: 'check violation' } }
|
||||
if (args.p_invoice_id === 'inv-3') return { data: { ok: false, code: 'INVOICE_NOT_FOUND' }, error: null }
|
||||
return rpcWrote(args)
|
||||
})
|
||||
|
||||
const result = await completeMigratedInvoiceLines({ supabase, companyId: 'co-1', consentId: 'c-1' })
|
||||
|
||||
expect(result).toMatchObject({ completed: 1, failed: 1, headersUpdated: 1, remaining: 1 })
|
||||
expect(headerUpdates(calls)).toHaveLength(1)
|
||||
expect(result).toMatchObject({ completed: 1, failed: 2, headersUpdated: 1, remaining: 2 })
|
||||
expect(writes(calls).map((w) => w.p_invoice_id)).toEqual(['inv-1', 'inv-2', 'inv-3'])
|
||||
})
|
||||
|
||||
it('dry run: reports the plan and writes nothing', async () => {
|
||||
|
||||
@@ -29,13 +29,19 @@
|
||||
* momsdeklaration and every report read the ledger, not these columns, so
|
||||
* filling them changes what the invoice page shows and nothing that was
|
||||
* filed (see the 2026-08-22 verification in DECISIONS.md).
|
||||
*
|
||||
* Both are written by one call to the complete_invoice_rows RPC per invoice
|
||||
* (migration 20260906135730), the write path the migration wizard shares: it
|
||||
* 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.
|
||||
*/
|
||||
|
||||
import type { SupabaseClient } from '@supabase/supabase-js'
|
||||
import { ISO_DATE_RE } from '@/lib/invariants'
|
||||
import { createLogger } from '@/lib/logger'
|
||||
import { equalOre, roundOre } from '@/lib/money'
|
||||
import { chunk } from '@/lib/utils'
|
||||
import type { SalesInvoiceDto } from '@/lib/providers/dto'
|
||||
import type { ProviderName } from '@/lib/providers/types'
|
||||
import { resolveConsent } from '@/lib/providers/resolve-consent'
|
||||
@@ -133,9 +139,6 @@ interface CandidateRow {
|
||||
invoice_items: { id: string }[] | null
|
||||
}
|
||||
|
||||
/** Invoices per statement. Small enough that a chunk's rows stay one request. */
|
||||
const WRITE_CHUNK_SIZE = 100
|
||||
|
||||
/**
|
||||
* How far the rows may disagree with the header before the invoice is left
|
||||
* alone. Öresavrundning puts up to 0.50 kr between a Fortnox `Total` and the
|
||||
@@ -219,32 +222,35 @@ async function loadCandidates(supabase: SupabaseClient, companyId: string): Prom
|
||||
return rows.filter((row) => (row.invoice_items?.length ?? 0) === 0)
|
||||
}
|
||||
|
||||
/** Ids among `ids` that gained rows since the candidates were loaded. */
|
||||
async function alreadyFilled(supabase: SupabaseClient, ids: string[]): Promise<Set<string>> {
|
||||
const { data, error } = await supabase
|
||||
.from('invoice_items')
|
||||
.select('invoice_id')
|
||||
.in('invoice_id', ids)
|
||||
if (error) throw new Error(`invoice_items lookup failed: ${error.message}`)
|
||||
return new Set(((data ?? []) as { invoice_id: string }[]).map((r) => r.invoice_id))
|
||||
}
|
||||
|
||||
/** The header VAT split the detail form established, ready to write. */
|
||||
/**
|
||||
* 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
|
||||
subtotalSek: number | null
|
||||
vatAmount: number
|
||||
vatAmountSek: number | null
|
||||
vatRate: number | null
|
||||
vatTreatment: string
|
||||
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
|
||||
}
|
||||
|
||||
export async function completeMigratedInvoiceLines(
|
||||
options: CompleteInvoiceLinesOptions,
|
||||
): Promise<CompleteInvoiceLinesResult> {
|
||||
@@ -357,19 +363,15 @@ export async function completeMigratedInvoiceLines(
|
||||
const vatAmount = mapped.invoice.vat_amount as number
|
||||
header = {
|
||||
subtotal,
|
||||
subtotalSek: toRowSek(subtotal, row),
|
||||
vatAmount,
|
||||
vatAmountSek: toRowSek(vatAmount, row),
|
||||
vatRate: mapped.invoice.vat_rate as number | null,
|
||||
vatTreatment: mapped.invoice.vat_treatment as string,
|
||||
subtotal_sek: toRowSek(subtotal, row),
|
||||
vat_amount: vatAmount,
|
||||
vat_amount_sek: toRowSek(vatAmount, row),
|
||||
vat_rate: mapped.invoice.vat_rate as number | null,
|
||||
vat_treatment: mapped.invoice.vat_treatment as string,
|
||||
}
|
||||
}
|
||||
|
||||
planned.push({
|
||||
row,
|
||||
items: mapped.items.map((item) => ({ ...item, invoice_id: row.id })),
|
||||
header,
|
||||
})
|
||||
planned.push({ row, items: mapped.items, header })
|
||||
}
|
||||
|
||||
if (dryRun) {
|
||||
@@ -379,44 +381,36 @@ export async function completeMigratedInvoiceLines(
|
||||
return result
|
||||
}
|
||||
|
||||
for (const batch of chunk(planned, WRITE_CHUNK_SIZE)) {
|
||||
// A concurrent run (the wizard and the cron, or two crons overlapping)
|
||||
// may have filled some of these since the candidates were loaded. Rows
|
||||
// are appended, never replaced, so a second write would double them.
|
||||
const filled = await alreadyFilled(supabase, batch.map((p) => p.row.id))
|
||||
const todo = batch.filter((p) => !filled.has(p.row.id))
|
||||
|
||||
const written = await insertRows(supabase, todo)
|
||||
for (const plan of todo) {
|
||||
if (!written.has(plan.row.id)) {
|
||||
result.failed++
|
||||
continue
|
||||
}
|
||||
result.completed++
|
||||
if (!plan.header) continue
|
||||
// Written as a literal so the schema guard checks these columns.
|
||||
const { error } = await supabase
|
||||
.from('invoices')
|
||||
.update({
|
||||
subtotal: plan.header.subtotal,
|
||||
subtotal_sek: plan.header.subtotalSek,
|
||||
vat_amount: plan.header.vatAmount,
|
||||
vat_amount_sek: plan.header.vatAmountSek,
|
||||
vat_rate: plan.header.vatRate,
|
||||
vat_treatment: plan.header.vatTreatment,
|
||||
})
|
||||
.eq('id', plan.row.id)
|
||||
.eq('company_id', companyId)
|
||||
if (error) {
|
||||
// The rows landed; only the header split is still the old shape. The
|
||||
// next run will not revisit this invoice (it now has rows), so say so.
|
||||
log.error('header VAT update failed after the rows were written', {
|
||||
companyId, invoiceId: plan.row.id, reason: error.message,
|
||||
})
|
||||
continue
|
||||
}
|
||||
result.headersUpdated++
|
||||
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,
|
||||
})
|
||||
const outcome = (data ?? null) as CompleteRowsOutcome | null
|
||||
if (error || !outcome?.ok) {
|
||||
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',
|
||||
})
|
||||
continue
|
||||
}
|
||||
if (!outcome.wrote) {
|
||||
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++
|
||||
}
|
||||
|
||||
result.remaining = candidates.length - result.completed
|
||||
@@ -435,33 +429,3 @@ export async function completeMigratedInvoiceLines(
|
||||
})
|
||||
return result
|
||||
}
|
||||
|
||||
/**
|
||||
* Insert every plan's rows: one statement for the batch, and on failure one
|
||||
* statement per invoice so a single bad row rejects its own invoice, not the
|
||||
* hundred beside it. An invoice's rows never split across statements: it
|
||||
* either has all of them or none.
|
||||
*/
|
||||
async function insertRows(supabase: SupabaseClient, plans: PlannedWrite[]): Promise<Set<string>> {
|
||||
const written = new Set<string>()
|
||||
if (plans.length === 0) return written
|
||||
|
||||
const bulk = await supabase.from('invoice_items').insert(plans.flatMap((p) => p.items))
|
||||
if (!bulk.error) {
|
||||
for (const plan of plans) written.add(plan.row.id)
|
||||
return written
|
||||
}
|
||||
|
||||
log.warn('bulk invoice_items insert failed; retrying per invoice', { reason: bulk.error.message })
|
||||
for (const plan of plans) {
|
||||
const { error } = await supabase.from('invoice_items').insert(plan.items)
|
||||
if (error) {
|
||||
log.error('invoice_items insert failed', {
|
||||
invoiceId: plan.row.id, invoiceNumber: plan.row.invoice_number, reason: error.message,
|
||||
})
|
||||
continue
|
||||
}
|
||||
written.add(plan.row.id)
|
||||
}
|
||||
return written
|
||||
}
|
||||
|
||||
@@ -86,6 +86,8 @@ export interface MigrationOptions {
|
||||
* PostgREST's practical size limit while minimising round-trips.
|
||||
*/
|
||||
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
|
||||
const ENRICHMENT_CONCURRENCY = 10
|
||||
|
||||
function emitProgress(options: MigrationOptions, progress: MigrationProgress) {
|
||||
@@ -736,13 +738,13 @@ export async function executeMigration(options: MigrationOptions): Promise<Migra
|
||||
errorSample ??= outcome.firstError
|
||||
}
|
||||
|
||||
const allItems: Record<string, unknown>[] = []
|
||||
const rowsByInvoice: { invoiceId: string; rows: Record<string, unknown>[] }[] = []
|
||||
for (let i = 0; i < mappedBatch.length; i++) {
|
||||
const insertedRow = outcome.returned[i]
|
||||
if (!insertedRow) continue
|
||||
const invoiceId = insertedRow.id
|
||||
for (const item of mappedBatch[i].items) {
|
||||
allItems.push({ ...item, invoice_id: invoiceId })
|
||||
if (mappedBatch[i].items.length > 0) {
|
||||
rowsByInvoice.push({ invoiceId: String(invoiceId), rows: mappedBatch[i].items })
|
||||
}
|
||||
registrationLinkInputs.push({
|
||||
invoiceId: String(invoiceId),
|
||||
@@ -772,13 +774,27 @@ export async function executeMigration(options: MigrationOptions): Promise<Migra
|
||||
imported++
|
||||
}
|
||||
|
||||
if (allItems.length > 0) {
|
||||
for (const itemBatch of chunk(allItems, INSERT_CHUNK_SIZE)) {
|
||||
const { error: itemErr } = await supabase.from('invoice_items').insert(itemBatch)
|
||||
if (itemErr) {
|
||||
console.error(`[migration] Sales invoice items insert failed (${itemBatch.length}):`, itemErr.message)
|
||||
// One complete_invoice_rows call per invoice (migration
|
||||
// 20260906135730): the write path the row-completion pass uses, so
|
||||
// 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',
|
||||
)
|
||||
}
|
||||
}
|
||||
}))
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user