From 2a1ec5ec2fb81f28499d29d6bfc4d8b3ef01554f Mon Sep 17 00:00:00 2001 From: Jack Ek <2727556+kjekac@users.noreply.github.com> Date: Thu, 16 Jul 2026 16:00:01 +0200 Subject: [PATCH] fix: Make SIE imports atomic (#860) * fix: Make SIE imports atomic * fix(import): carry dimensions + harden the atomic SIE RPC Rebased onto current main. The RPC now: - carries the per-line dimensions jsonb through the payload + INSERT so imported SIE object-list codes are not dropped (dimensions PR5 #866); - uses the NULL-safe caller_is_company_member guard (drops the banned NOT IN (SELECT user_company_ids()) pattern ratcheted since #881); - verifies the fiscal period belongs to the company; - enforces per-voucher balance (sum debit = sum credit > 0) since SECURITY DEFINER + the direct draft->posted UPDATE bypass the trigger path; - ships REVOKE ALL FROM PUBLIC, anon / GRANT EXECUTE TO authenticated, service_role (house style). Migration renamed to a current timestamp. Added pg-real coverage for the dimensions round-trip, unbalanced rejection, and foreign-fiscal-period guard. Co-Authored-By: Claude Fable 5 --------- Co-authored-by: Jakob Wennberg Co-authored-by: Claude Fable 5 --- .../__tests__/sie-import-atomic.pg.test.ts | 191 +++++++++++ lib/import/__tests__/sie-import.test.ts | 126 ++++--- lib/import/sie-import.ts | 319 +++++------------- ...2150000_import_sie_journal_entries_rpc.sql | 225 ++++++++++++ 4 files changed, 579 insertions(+), 282 deletions(-) create mode 100644 lib/import/__tests__/sie-import-atomic.pg.test.ts create mode 100644 supabase/migrations/20260712150000_import_sie_journal_entries_rpc.sql diff --git a/lib/import/__tests__/sie-import-atomic.pg.test.ts b/lib/import/__tests__/sie-import-atomic.pg.test.ts new file mode 100644 index 00000000..b9e772b6 --- /dev/null +++ b/lib/import/__tests__/sie-import-atomic.pg.test.ts @@ -0,0 +1,191 @@ +import { describe, expect, it } from 'vitest' +import { getPool } from '@/tests/pg/setup' +import { seedCompany } from '@/tests/pg/fixtures' + +describe('import_sie_journal_entries RPC', () => { + it('rolls back the journal entry header when a line insert fails', async () => { + const { userId, companyId, fiscalPeriodId } = await seedCompany() + + const payload = [ + { + sourceId: 'A1', + series: 'A', + date: '2026-01-15', + description: 'Bad imported voucher', + sourceSeries: 'A', + sourceNumber: 1, + sourceType: 'import', + lines: [ + { + account_number: '1930', + debit_amount: 100, + credit_amount: 0, + currency: 'SEK', + line_description: 'Bank', + sort_order: 0, + }, + { + account_number: null, + debit_amount: 0, + credit_amount: 100, + currency: 'SEK', + line_description: 'Invalid line', + sort_order: 1, + }, + ], + }, + ] + + await expect( + getPool().query( + `SELECT public.import_sie_journal_entries($1::uuid, $2::uuid, $3::uuid, $4::jsonb)`, + [companyId, userId, fiscalPeriodId, JSON.stringify(payload)], + ), + ).rejects.toThrow(/null value in column "account_number"|violates not-null constraint/i) + + const headers = await getPool().query<{ count: string }>( + `SELECT count(*)::text AS count + FROM public.journal_entries + WHERE company_id = $1 + AND fiscal_period_id = $2 + AND description = 'Bad imported voucher'`, + [companyId, fiscalPeriodId], + ) + expect(headers.rows[0]!.count).toBe('0') + + const sequence = await getPool().query<{ last_number: number }>( + `SELECT last_number + FROM public.voucher_sequences + WHERE company_id = $1 + AND fiscal_period_id = $2 + AND voucher_series = 'A'`, + [companyId, fiscalPeriodId], + ) + expect(sequence.rowCount).toBe(0) + }) + + it('posts a balanced voucher and carries the dimensions jsonb through to the generated mirrors', async () => { + const { userId, companyId, fiscalPeriodId } = await seedCompany() + + const payload = [ + { + sourceId: 'A1', + series: 'A', + date: '2026-02-01', + description: 'Dimensioned import', + sourceSeries: 'A', + sourceNumber: 1, + sourceType: 'import', + lines: [ + { + account_number: '5010', + debit_amount: 100, + credit_amount: 0, + currency: 'SEK', + line_description: 'Lokalhyra', + sort_order: 0, + // SIE object-list codes: 1 = kostnadsställe, 6 = projekt. + dimensions: { '1': 'CC-10', '6': 'PROJ-X' }, + }, + { + account_number: '1930', + debit_amount: 0, + credit_amount: 100, + currency: 'SEK', + line_description: 'Bank', + sort_order: 1, + }, + ], + }, + ] + + const res = await getPool().query<{ import_sie_journal_entries: { inserted_entries: unknown[] } }>( + `SELECT public.import_sie_journal_entries($1::uuid, $2::uuid, $3::uuid, $4::jsonb)`, + [companyId, userId, fiscalPeriodId, JSON.stringify(payload)], + ) + expect(res.rows[0]!.import_sie_journal_entries.inserted_entries).toHaveLength(1) + + const posted = await getPool().query<{ count: string }>( + `SELECT count(*)::text AS count + FROM public.journal_entries + WHERE company_id = $1 AND status = 'posted' AND description = 'Dimensioned import'`, + [companyId], + ) + expect(posted.rows[0]!.count).toBe('1') + + const dimLine = await getPool().query<{ + dimensions: Record + cost_center: string | null + project: string | null + }>( + `SELECT l.dimensions, l.cost_center, l.project + FROM public.journal_entry_lines l + JOIN public.journal_entries je ON je.id = l.journal_entry_id + WHERE je.company_id = $1 AND l.account_number = '5010'`, + [companyId], + ) + expect(dimLine.rows[0]!.dimensions).toEqual({ '1': 'CC-10', '6': 'PROJ-X' }) + // GENERATED mirrors derive from the jsonb: both must be populated. + expect(dimLine.rows[0]!.cost_center).not.toBeNull() + expect(dimLine.rows[0]!.project).not.toBeNull() + }) + + it('rejects an unbalanced voucher and rolls the whole import back', async () => { + const { userId, companyId, fiscalPeriodId } = await seedCompany() + + const payload = [ + { + sourceId: 'A1', + series: 'A', + date: '2026-02-01', + description: 'Unbalanced import', + sourceType: 'import', + lines: [ + { account_number: '5010', debit_amount: 100, credit_amount: 0, currency: 'SEK', sort_order: 0 }, + { account_number: '1930', debit_amount: 0, credit_amount: 90, currency: 'SEK', sort_order: 1 }, + ], + }, + ] + + await expect( + getPool().query( + `SELECT public.import_sie_journal_entries($1::uuid, $2::uuid, $3::uuid, $4::jsonb)`, + [companyId, userId, fiscalPeriodId, JSON.stringify(payload)], + ), + ).rejects.toThrow(/unbalanced/i) + + const headers = await getPool().query<{ count: string }>( + `SELECT count(*)::text AS count FROM public.journal_entries + WHERE company_id = $1 AND description = 'Unbalanced import'`, + [companyId], + ) + expect(headers.rows[0]!.count).toBe('0') + }) + + it('rejects a fiscal period that belongs to another company', async () => { + const a = await seedCompany() + const b = await seedCompany() + + const payload = [ + { + sourceId: 'A1', + series: 'A', + date: '2026-02-01', + description: 'Foreign fiscal period', + sourceType: 'import', + lines: [ + { account_number: '5010', debit_amount: 100, credit_amount: 0, currency: 'SEK', sort_order: 0 }, + { account_number: '1930', debit_amount: 0, credit_amount: 100, currency: 'SEK', sort_order: 1 }, + ], + }, + ] + + // company A's id + user, but company B's fiscal period. + await expect( + getPool().query( + `SELECT public.import_sie_journal_entries($1::uuid, $2::uuid, $3::uuid, $4::jsonb)`, + [a.companyId, a.userId, b.fiscalPeriodId, JSON.stringify(payload)], + ), + ).rejects.toThrow(/does not belong to company/i) + }) +}) diff --git a/lib/import/__tests__/sie-import.test.ts b/lib/import/__tests__/sie-import.test.ts index b805dcc1..3218e592 100644 --- a/lib/import/__tests__/sie-import.test.ts +++ b/lib/import/__tests__/sie-import.test.ts @@ -803,15 +803,12 @@ describe('importVouchers: per-voucher series preservation', () => { // voucher_series per inserted record. Uses a hand-rolled mock rather than // createQueuedMockSupabase because we need to inspect arguments, not just // return queued data. - function buildCapturingSupabase() { + function buildCapturingSupabase(options: { failImportRpc?: boolean } = {}) { const journalEntryInserts: Array> = [] const journalEntryLineInserts: Array> = [] const rpcCalls: Array<{ name: string; args: Record }> = [] - // Each `next_voucher_number` RPC call auto-increments per series, matching - // the DB function's ON CONFLICT behavior. const nextNumberBySeries = new Map() - let syntheticEntryId = 1 const supabase = { @@ -833,52 +830,56 @@ describe('importVouchers: per-voucher series preservation', () => { } } - if (table === 'journal_entries') { - return { - insert: (rows: Array>) => { - journalEntryInserts.push(...rows) - return { - select: () => ({ - then: (resolve: (v: { data: { id: string }[]; error: null }) => void) => - resolve({ - data: rows.map(() => ({ id: `entry-${syntheticEntryId++}` })), - error: null, - }), - }), - } - }, - } - } - - if (table === 'journal_entry_lines') { - return { - insert: (rows: Array>) => { - journalEntryLineInserts.push(...rows) - return Promise.resolve({ error: null }) - }, - } - } - throw new Error(`Unexpected table: ${table}`) }), rpc: vi.fn(async (name: string, args: Record) => { rpcCalls.push({ name, args }) - if (name === 'next_voucher_number') { - const series = args.p_series as string - const current = nextNumberBySeries.get(series) ?? 0 - const next = current + 1 - nextNumberBySeries.set(series, next) - return { data: next, error: null } - } - if (name === 'reserve_voucher_range') { - const series = args.p_series as string - const highest = args.p_highest_used as number - nextNumberBySeries.set(series, highest) - return { data: null, error: null } - } - if (name === 'release_voucher_range') { - return { data: null, error: null } + if (name === 'import_sie_journal_entries') { + if (options.failImportRpc) { + return { + data: null, + error: { message: 'line insert failed' }, + } + } + + const entries = args.p_entries as Array<{ + sourceId: string + series: string + sourceType: string + lines: Array> + }> + const inserted_entries = entries.map((entry) => { + const current = nextNumberBySeries.get(entry.series) ?? 0 + const next = current + 1 + nextNumberBySeries.set(entry.series, next) + const id = `entry-${syntheticEntryId++}` + journalEntryInserts.push({ + ...entry, + id, + voucher_series: entry.series, + voucher_number: next, + source_type: entry.sourceType, + source_voucher_series: (entry as { sourceSeries?: string | null }).sourceSeries ?? null, + source_voucher_number: (entry as { sourceNumber?: number | null }).sourceNumber ?? null, + }) + journalEntryLineInserts.push(...entry.lines.map((line) => ({ ...line, journal_entry_id: id }))) + return { + id, + sourceId: entry.sourceId, + series: entry.series, + voucherNumber: next, + sourceType: entry.sourceType, + } + }) + return { + data: { + inserted_entries, + skipped_duplicates: [], + validation_errors: [], + }, + error: null, + } } throw new Error(`Unexpected RPC: ${name}`) }), @@ -941,9 +942,9 @@ describe('importVouchers: per-voucher series preservation', () => { const seriesInInserts = journalEntryInserts.map((r) => r.voucher_series) expect(seriesInInserts).toEqual(['B', 'B', 'C', 'V']) - // Each series reserves its own voucher-number range independently - const reserveCalls = rpcCalls.filter((c) => c.name === 'reserve_voucher_range') - expect(reserveCalls.map((c) => c.args.p_series)).toEqual(['B', 'C', 'V']) + const importCalls = rpcCalls.filter((c) => c.name === 'import_sie_journal_entries') + expect(importCalls).toHaveLength(1) + expect((importCalls[0].args.p_entries as Array<{ series: string }>).map((e) => e.series)).toEqual(['B', 'B', 'C', 'V']) }) it('falls back to defaultSeries when source voucher has empty series (SIE4I)', async () => { @@ -1059,6 +1060,35 @@ describe('importVouchers: per-voucher series preservation', () => { expect(journalEntryInserts.map((r) => r.source_voucher_number)).toEqual([1, 3]) }) + it('does not report imported IDs or counts when the atomic RPC fails', async () => { + const { supabase, journalEntryInserts, journalEntryLineInserts } = buildCapturingSupabase({ + failImportRpc: true, + }) + const parsed = makeParsedFile({ + vouchers: [ + makeVoucher('A', 1), + ], + }) + + const result = await importVouchers( + supabase, + 'company-1', + 'user-1', + 'period-1', + parsed, + baseMap, + 'A', + ) + + expect(result.created).toBe(0) + expect(result.ids).toEqual([]) + expect(result.importTypedIds).toEqual([]) + expect(result.voucherNumberMapping).toEqual([]) + expect(result.errors.join(' ')).toContain('line insert failed') + expect(journalEntryInserts).toEqual([]) + expect(journalEntryLineInserts).toEqual([]) + }) + it('stores NULL source series/number when the source voucher has no series (SIE4I subsystem import)', async () => { const { supabase, journalEntryInserts } = buildCapturingSupabase() const parsed = makeParsedFile({ diff --git a/lib/import/sie-import.ts b/lib/import/sie-import.ts index ef1e7bfa..9d010494 100644 --- a/lib/import/sie-import.ts +++ b/lib/import/sie-import.ts @@ -1247,249 +1247,100 @@ export async function importVouchers( results.seriesUsed = [...seriesGroups.keys()] - // Batch insert journal entries (in chunks of 100) with retry logic. - // Retries handle transient errors (Supabase rate limits, Cloudflare 500s). - const BATCH_SIZE = 100 - const MAX_RETRIES = 3 - const INTER_BATCH_DELAY_MS = 50 // Prevent rate limiting under sustained load - let retriedBatches = 0 - let failedBatches = 0 + const voucherBySourceId = new Map(preparedVouchers.map((voucher) => [voucher.sourceId, voucher])) + const rpcPayload = preparedVouchers.map((voucher) => ({ + sourceId: voucher.sourceId, + series: voucher.series, + date: voucher.date, + description: voucher.description, + sourceSeries: voucher.sourceSeries, + sourceNumber: voucher.sourceNumber, + sourceType: voucher.sourceType, + lines: voucher.lines.map((line, lineIndex) => ({ + account_number: line.account_number, + account_id: accountIdMap.get(line.account_number) || null, + debit_amount: line.debit_amount, + credit_amount: line.credit_amount, + currency: 'SEK', + line_description: line.line_description, + sort_order: lineIndex, + // dimensions jsonb is the source of truth; cost_center/project are + // GENERATED mirrors the DB derives from it. SIE object-list codes carry + // through so imported dimension data is not dropped (dimensions PR5 #866). + dimensions: normalizeLineDimensions({ dimensions: line.dimensions ?? null }), + })), + })) - // Process each series as an independent mini-import. Voucher numbers must - // be monotonically increasing within a series; grouping first guarantees - // that without needing to interleave series-specific counters in one loop. - let seriesIndex = 0 - for (const [series, groupVouchers] of seriesGroups) { - // Get starting voucher number for this series - const { data: startNumber } = await supabase.rpc('next_voucher_number', { - p_company_id: companyId, - p_fiscal_period_id: fiscalPeriodId, - p_series: series, + type ImportSieJournalEntriesRpcResult = { + inserted_entries?: Array<{ + id: string + sourceId: string + series: string + voucherNumber: number + sourceType: 'import' | 'opening_balance' + }> + skipped_duplicates?: Array<{ sourceId?: string; reason?: string }> + validation_errors?: Array<{ sourceId?: string; message?: string }> + } + + const { data: rpcResult, error: rpcError } = await supabase.rpc('import_sie_journal_entries', { + p_company_id: companyId, + p_user_id: userId, + p_fiscal_period_id: fiscalPeriodId, + p_entries: rpcPayload, + }) + + if (rpcError) { + results.errors.push(`SIE-verifikationer kunde inte importeras atomiskt: ${rpcError.message}`) + results.failedBatches = 1 + return results + } + + const structuredResult = (rpcResult ?? {}) as ImportSieJournalEntriesRpcResult + for (const validationError of structuredResult.validation_errors ?? []) { + results.errors.push( + validationError.sourceId + ? `${validationError.sourceId}: ${validationError.message ?? 'valideringsfel'}` + : validationError.message ?? 'Valideringsfel vid SIE-import', + ) + } + for (const skippedDuplicate of structuredResult.skipped_duplicates ?? []) { + results.errors.push( + skippedDuplicate.sourceId + ? `${skippedDuplicate.sourceId}: duplicerad verifikation hoppades över` + : 'Duplicerad verifikation hoppades över', + ) + } + + if (results.errors.length > 0) { + return results + } + + for (const inserted of structuredResult.inserted_entries ?? []) { + const voucher = voucherBySourceId.get(inserted.sourceId) + if (!voucher) continue + + results.voucherNumberMapping.push({ + sourceId: inserted.sourceId, + series: inserted.series, + targetNumber: inserted.voucherNumber, }) - const currentVoucherNumber = (startNumber as number) || 1 - - // Reserve the full voucher number range upfront to prevent concurrent - // operations from claiming numbers in our range during batch insertion. - const reservedHighest = currentVoucherNumber + groupVouchers.length - 1 - await supabase.rpc('reserve_voucher_range', { - p_company_id: companyId, - p_fiscal_period_id: fiscalPeriodId, - p_series: series, - p_highest_used: reservedHighest, - }) - - let highestInsertedVoucher = currentVoucherNumber - 1 // nothing inserted yet - - for (let batchStart = 0; batchStart < groupVouchers.length; batchStart += BATCH_SIZE) { - const batch = groupVouchers.slice(batchStart, batchStart + BATCH_SIZE) - const batchNumber = Math.floor(batchStart / BATCH_SIZE) + 1 - let batchWasRetried = false - - // Prepare journal entry headers - const entryInserts = batch.map((v, i) => ({ - user_id: userId, - company_id: companyId, - fiscal_period_id: fiscalPeriodId, - voucher_number: currentVoucherNumber + batchStart + i, - voucher_series: series, - entry_date: v.date, - description: v.description, - source_type: v.sourceType, - source_voucher_series: v.sourceSeries, - source_voucher_number: v.sourceNumber, - status: 'posted', - committed_at: new Date().toISOString(), - })) - - // Insert headers with retry - let entries: { id: string }[] | null = null - let lastEntryError: string | null = null - - for (let attempt = 0; attempt <= MAX_RETRIES; attempt++) { - if (attempt > 0) { - batchWasRetried = true - const backoffMs = Math.pow(2, attempt - 1) * 1000 // 1s, 2s, 4s - console.log(`[sie-import] Retrying batch ${batchNumber} (attempt ${attempt + 1}/${MAX_RETRIES + 1}) after ${backoffMs}ms`) - await new Promise(resolve => setTimeout(resolve, backoffMs)) - } - - const { data, error: entryError } = await supabase - .from('journal_entries') - .insert(entryInserts) - .select('id') - - if (!entryError && data) { - entries = data - lastEntryError = null - break - } - - lastEntryError = entryError?.message || 'Failed to insert entries' + results.ids.push(inserted.id) + if (inserted.sourceType === 'import') { + results.importTypedIds.push(inserted.id) } + results.created++ - if (!entries) { - failedBatches++ - results.errors.push( - `Batch ${batchNumber} misslyckades efter ${MAX_RETRIES + 1} försök: ${lastEntryError}` + for (const line of voucher.lines) { + const net = line.debit_amount - line.credit_amount + results.movementsByAccount.set( + line.account_number, + (results.movementsByAccount.get(line.account_number) || 0) + net ) - continue - } - - // Prepare all lines for this batch - const allLines: { - journal_entry_id: string - account_number: string - account_id: string | null - debit_amount: number - credit_amount: number - currency: string - line_description: string | null - sort_order: number - dimensions: Record - }[] = [] - - for (let i = 0; i < batch.length; i++) { - const entryId = entries[i]?.id - if (!entryId) continue - - const voucher = batch[i] - const assignedNumber = currentVoucherNumber + batchStart + i - voucher.lines.forEach((line, lineIndex) => { - // dimensions jsonb is the source of truth; cost_center/project are - // derived mirrors: the same dual-write every sanctioned writer uses - // (see lib/bookkeeping/dimension-resolver.ts). SIE object-list codes - // survive verbatim on lines (legacy free-text is a documented - // exception to the registry format rules). - const dims = normalizeLineDimensions({ dimensions: line.dimensions ?? null }) - allLines.push({ - journal_entry_id: entryId, - account_number: line.account_number, - account_id: accountIdMap.get(line.account_number) || null, - debit_amount: line.debit_amount, - credit_amount: line.credit_amount, - currency: 'SEK', - line_description: line.line_description, - sort_order: lineIndex, - dimensions: dims, - }) - }) - - results.voucherNumberMapping.push({ - sourceId: voucher.sourceId, - series: voucher.series, - targetNumber: assignedNumber, - }) - - results.ids.push(entryId) - // #VER vouchers re-tagged as opening_balance never need an underlag and - // aren't in NEEDS_DOC_SOURCE_TYPES, so keep them out of the exempt set. - if (voucher.sourceType === 'import') { - results.importTypedIds.push(entryId) - } - results.created++ - } - - // Insert all lines with retry - if (allLines.length > 0) { - let linesInserted = false - let lastLinesError: string | null = null - - for (let attempt = 0; attempt <= MAX_RETRIES; attempt++) { - if (attempt > 0) { - batchWasRetried = true - const backoffMs = Math.pow(2, attempt - 1) * 1000 - console.log(`[sie-import] Retrying batch ${batchNumber} lines (attempt ${attempt + 1}/${MAX_RETRIES + 1}) after ${backoffMs}ms`) - await new Promise(resolve => setTimeout(resolve, backoffMs)) - } - - const { error: linesError } = await supabase - .from('journal_entry_lines') - .insert(allLines) - - if (!linesError) { - linesInserted = true - break - } - - lastLinesError = linesError.message - } - - if (linesInserted) { - // Track highest voucher number only after both headers AND lines succeed, - // to avoid counting orphaned entries with no lines as "used". - const batchHighest = currentVoucherNumber + batchStart + batch.length - 1 - highestInsertedVoucher = Math.max(highestInsertedVoucher, batchHighest) - - // Track movements ONLY for successfully inserted vouchers. - // This ensures the migration adjustment correctly compensates for - // any batches that failed completely. - for (let i = 0; i < batch.length; i++) { - const voucher = batch[i] - for (const line of voucher.lines) { - const net = line.debit_amount - line.credit_amount - results.movementsByAccount.set( - line.account_number, - (results.movementsByAccount.get(line.account_number) || 0) + net - ) - } - } - } else { - failedBatches++ - results.errors.push( - `Batch ${batchNumber} rader misslyckades efter ${MAX_RETRIES + 1} försök: ${lastLinesError}` - ) - } - } else { - // No lines to insert: still count movements for vouchers with entries - for (let i = 0; i < batch.length; i++) { - const voucher = batch[i] - for (const line of voucher.lines) { - const net = line.debit_amount - line.credit_amount - results.movementsByAccount.set( - line.account_number, - (results.movementsByAccount.get(line.account_number) || 0) + net - ) - } - } - } - - // Count distinct batches that needed retries (not individual attempts) - if (batchWasRetried) { - retriedBatches++ - } - - // Small delay between batches to prevent Supabase/Cloudflare rate limiting - const isLastBatchInSeries = batchStart + BATCH_SIZE >= groupVouchers.length - const isLastSeries = seriesIndex === seriesGroups.size - 1 - if (!isLastBatchInSeries || !isLastSeries) { - await new Promise(resolve => setTimeout(resolve, INTER_BATCH_DELAY_MS)) } } - // Adjust voucher sequence after insertion for this series. - // Range was pre-reserved to `reservedHighest`. If some batches failed, - // release the unused portion to avoid burned numbers and gap-explanation friction. - if (highestInsertedVoucher < reservedHighest) { - const releaseTarget = highestInsertedVoucher >= currentVoucherNumber - ? highestInsertedVoucher // partial success: set to actual highest - : currentVoucherNumber - 1 // total failure: roll back fully - await supabase.rpc('release_voucher_range', { - p_company_id: companyId, - p_fiscal_period_id: fiscalPeriodId, - p_series: series, - p_actual_last: releaseTarget, - p_reserved_highest: reservedHighest, - }) - } - - seriesIndex++ - } - - // Propagate batch retry stats - results.retriedBatches = retriedBatches - results.failedBatches = failedBatches - return results } diff --git a/supabase/migrations/20260712150000_import_sie_journal_entries_rpc.sql b/supabase/migrations/20260712150000_import_sie_journal_entries_rpc.sql new file mode 100644 index 00000000..569575b4 --- /dev/null +++ b/supabase/migrations/20260712150000_import_sie_journal_entries_rpc.sql @@ -0,0 +1,225 @@ +-- Atomic SIE journal-entry import. +-- +-- The TypeScript importer still parses, maps, and validates the SIE file. +-- This RPC owns the actual journal_entries + journal_entry_lines commit so a +-- failed line insert cannot leave posted header rows behind (the whole import +-- is one transaction: any RAISE rolls the entire file back). +-- +-- SECURITY DEFINER bypasses RLS and the draft-to-posted balance path, so this +-- function enforces, per voucher, on its own: (1) NULL-safe company membership +-- via caller_is_company_member (house rule since #881, ratcheted by +-- tests/pg/null-safe-tenant-guards.pg.test.ts); (2) the fiscal period belongs +-- to the company; (3) sum(debit) = sum(credit) and > 0 (hard rule #3, since +-- the enforcement triggers only cover immutability, not balance). Dimensions +-- jsonb carries through so imported SIE object-list codes are not dropped +-- (cost_center/project are GENERATED mirrors the DB derives, dimensions PR5). +CREATE OR REPLACE FUNCTION public.import_sie_journal_entries( + p_company_id uuid, + p_user_id uuid, + p_fiscal_period_id uuid, + p_entries jsonb +) +RETURNS jsonb +LANGUAGE plpgsql +SECURITY DEFINER +SET search_path = public +AS $$ +DECLARE + v_entry jsonb; + v_line jsonb; + v_series text; + v_count integer; + v_new_last integer; + v_start integer; + v_assigned_number integer; + v_entry_id uuid; + v_deb numeric; + v_cred numeric; + v_inserted jsonb := '[]'::jsonb; + v_jwt_role text := coalesce(nullif(current_setting('request.jwt.claims', true), '')::jsonb ->> 'role', ''); +BEGIN + IF p_entries IS NULL OR jsonb_typeof(p_entries) <> 'array' THEN + RAISE EXCEPTION 'p_entries must be a JSON array'; + END IF; + + -- Tenant guard: anon/authenticated may only import into their own companies; + -- service_role / direct access (no JWT role) bypasses for migrations and + -- server-side maintenance paths that scope company access before calling. + -- NULL-safe predicate (a NULL company resolves to false) per #881. + IF v_jwt_role IN ('anon', 'authenticated') + AND NOT public.caller_is_company_member(p_company_id) THEN + RAISE EXCEPTION 'unauthorized: caller is not a member of company %', p_company_id + USING ERRCODE = '42501'; + END IF; + + IF v_jwt_role IN ('anon', 'authenticated') + AND auth.uid() IS DISTINCT FROM p_user_id THEN + RAISE EXCEPTION 'unauthorized: p_user_id must match auth.uid()' + USING ERRCODE = '42501'; + END IF; + + -- The fiscal period must belong to the target company: a caller could + -- otherwise post into another company's period id (defense in depth; the + -- header company_id/FK would still scope the rows, but fail closed here). + IF NOT EXISTS ( + SELECT 1 FROM public.fiscal_periods + WHERE id = p_fiscal_period_id AND company_id = p_company_id + ) THEN + RAISE EXCEPTION 'fiscal period % does not belong to company %', p_fiscal_period_id, p_company_id + USING ERRCODE = '42501'; + END IF; + + CREATE TEMP TABLE IF NOT EXISTS pg_temp.sie_import_series_numbers ( + series text PRIMARY KEY, + next_number integer NOT NULL + ) ON COMMIT DROP; + TRUNCATE pg_temp.sie_import_series_numbers; + + FOR v_series, v_count IN + SELECT COALESCE(NULLIF(e.value->>'series', ''), 'A') AS series, count(*)::integer AS count + FROM jsonb_array_elements(p_entries) WITH ORDINALITY AS e(value, ord) + GROUP BY COALESCE(NULLIF(e.value->>'series', ''), 'A') + ORDER BY min(e.ord) + LOOP + INSERT INTO public.voucher_sequences + (company_id, user_id, fiscal_period_id, voucher_series, last_number) + VALUES + (p_company_id, p_user_id, p_fiscal_period_id, v_series, v_count) + ON CONFLICT (company_id, fiscal_period_id, voucher_series) + DO UPDATE SET + last_number = public.voucher_sequences.last_number + EXCLUDED.last_number, + updated_at = now() + RETURNING last_number INTO v_new_last; + + v_start := v_new_last - v_count + 1; + + INSERT INTO pg_temp.sie_import_series_numbers(series, next_number) + VALUES (v_series, v_start); + END LOOP; + + FOR v_entry IN + SELECT e.value + FROM jsonb_array_elements(p_entries) WITH ORDINALITY AS e(value, ord) + ORDER BY e.ord + LOOP + v_series := COALESCE(NULLIF(v_entry->>'series', ''), 'A'); + + SELECT next_number + INTO v_assigned_number + FROM pg_temp.sie_import_series_numbers + WHERE series = v_series + FOR UPDATE; + + UPDATE pg_temp.sie_import_series_numbers + SET next_number = next_number + 1 + WHERE series = v_series; + + INSERT INTO public.journal_entries ( + user_id, + company_id, + fiscal_period_id, + voucher_number, + voucher_series, + entry_date, + description, + source_type, + source_voucher_series, + source_voucher_number, + status + ) + VALUES ( + p_user_id, + p_company_id, + p_fiscal_period_id, + v_assigned_number, + v_series, + (v_entry->>'date')::date, + v_entry->>'description', + COALESCE(NULLIF(v_entry->>'sourceType', ''), 'import'), + NULLIF(v_entry->>'sourceSeries', ''), + CASE + WHEN v_entry ? 'sourceNumber' AND v_entry->>'sourceNumber' IS NOT NULL + THEN (v_entry->>'sourceNumber')::integer + ELSE NULL + END, + 'draft' + ) + RETURNING id INTO v_entry_id; + + IF jsonb_typeof(v_entry->'lines') <> 'array' OR jsonb_array_length(v_entry->'lines') = 0 THEN + RAISE EXCEPTION 'SIE journal entry % has no lines', COALESCE(v_entry->>'sourceId', ''); + END IF; + + FOR v_line IN + SELECT l.value + FROM jsonb_array_elements(v_entry->'lines') WITH ORDINALITY AS l(value, ord) + ORDER BY l.ord + LOOP + INSERT INTO public.journal_entry_lines ( + journal_entry_id, + account_number, + account_id, + debit_amount, + credit_amount, + currency, + line_description, + sort_order, + dimensions + ) + VALUES ( + v_entry_id, + v_line->>'account_number', + CASE + WHEN v_line ? 'account_id' AND v_line->>'account_id' IS NOT NULL + THEN (v_line->>'account_id')::uuid + ELSE NULL + END, + COALESCE((v_line->>'debit_amount')::numeric, 0), + COALESCE((v_line->>'credit_amount')::numeric, 0), + COALESCE(NULLIF(v_line->>'currency', ''), 'SEK'), + NULLIF(v_line->>'line_description', ''), + COALESCE((v_line->>'sort_order')::integer, 0), + COALESCE(v_line->'dimensions', '{}'::jsonb) + ); + END LOOP; + + -- Per-voucher balance enforcement (hard rule #3). SECURITY DEFINER + the + -- direct draft->posted UPDATE below bypass the trigger path, so assert + -- balance here; a RAISE rolls the whole atomic import back. + SELECT COALESCE(sum(debit_amount), 0), COALESCE(sum(credit_amount), 0) + INTO v_deb, v_cred + FROM public.journal_entry_lines + WHERE journal_entry_id = v_entry_id; + + IF round(v_deb, 2) <> round(v_cred, 2) OR round(v_deb, 2) <= 0 THEN + RAISE EXCEPTION 'SIE journal entry % is unbalanced (debit %, credit %)', + COALESCE(v_entry->>'sourceId', ''), v_deb, v_cred; + END IF; + + UPDATE public.journal_entries + SET status = 'posted', + committed_at = now() + WHERE id = v_entry_id + AND company_id = p_company_id; + + v_inserted := v_inserted || jsonb_build_array(jsonb_build_object( + 'id', v_entry_id, + 'sourceId', v_entry->>'sourceId', + 'series', v_series, + 'voucherNumber', v_assigned_number, + 'sourceType', COALESCE(NULLIF(v_entry->>'sourceType', ''), 'import') + )); + END LOOP; + + RETURN jsonb_build_object( + 'inserted_entries', v_inserted, + 'skipped_duplicates', '[]'::jsonb, + 'validation_errors', '[]'::jsonb + ); +END; +$$; + +REVOKE ALL ON FUNCTION public.import_sie_journal_entries(uuid, uuid, uuid, jsonb) FROM PUBLIC, anon; +GRANT EXECUTE ON FUNCTION public.import_sie_journal_entries(uuid, uuid, uuid, jsonb) TO authenticated, service_role; + +NOTIFY pgrst, 'reload schema';