fix(accounting): Belegvorbereitung im Hintergrund ausführen und Dubletten verhindern
- „Belege vorbereiten“ startet den Lauf im Hintergrund (202) und fragt den Status ab; ein laufender Vorgang wird nicht erneut gestartet und beim Wiederöffnen der Seite weiterverfolgt - Eingangsbeleg und Dateiverknüpfung werden in einer Transaktion angelegt; ist die Datei bereits verknüpft, wird der neue Beleg verworfen - Fehler einzelner Dateien brechen den Lauf nicht mehr ab Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -93,6 +93,15 @@ export type IncomingInvoicePreparationResult = {
|
||||
error?: string
|
||||
}
|
||||
|
||||
export type IncomingInvoicePreparationStatus = {
|
||||
running: boolean
|
||||
startedAt: string | null
|
||||
finishedAt: string | null
|
||||
result: IncomingInvoicePreparationResult | null
|
||||
}
|
||||
|
||||
class FileAlreadyPreparedError extends Error {}
|
||||
|
||||
export function prepareIncomingInvoices(server: FastifyInstance) {
|
||||
const processInvoices = async (tenantId:number, fileIds?: string[]) => {
|
||||
const result: IncomingInvoicePreparationResult = { filesFound: 0, created: [], skipped: [] }
|
||||
@@ -162,112 +171,152 @@ export function prepareIncomingInvoices(server: FastifyInstance) {
|
||||
// 3️⃣ Strukturierte E-Rechnung bevorzugen, GPT als PDF-Fallback verwenden
|
||||
// -------------------------------------------------------------
|
||||
for (const file of filesRes) {
|
||||
console.log(`Processing file ${file.id} for tenant ${tenantId}`)
|
||||
|
||||
let fileData: Buffer
|
||||
|
||||
try {
|
||||
fileData = await loadFileBuffer(file.path!)
|
||||
await processFile(tenantId, file, result)
|
||||
} catch (error) {
|
||||
server.log.error(error, `Datei ${file.id} konnte nicht aus S3 geladen werden.`)
|
||||
result.skipped.push({ fileId: file.id, reason: "Datei konnte nicht geladen werden" })
|
||||
continue
|
||||
}
|
||||
|
||||
const electronicInvoice = await detectElectronicInvoice(fileData, file)
|
||||
|
||||
let itemInfo: any = null
|
||||
|
||||
if (electronicInvoice) {
|
||||
server.log.info({
|
||||
fileId: file.id,
|
||||
container: electronicInvoice.container,
|
||||
syntax: electronicInvoice.syntax,
|
||||
attachmentName: electronicInvoice.attachmentName,
|
||||
}, "Strukturierte E-Rechnung erkannt.")
|
||||
|
||||
try {
|
||||
const parsedInvoice = parseElectronicInvoice(electronicInvoice.xml, electronicInvoice.syntax)
|
||||
const vendor = await findElectronicInvoiceVendor(server, tenantId, parsedInvoice)
|
||||
itemInfo = mapElectronicInvoiceToIncomingInvoice(
|
||||
parsedInvoice,
|
||||
tenantId,
|
||||
vendor?.id || null,
|
||||
electronicInvoice.container,
|
||||
electronicInvoice.attachmentName,
|
||||
)
|
||||
} catch (error) {
|
||||
server.log.error(error, `E-Rechnung aus Datei ${file.id} konnte nicht verarbeitet werden.`)
|
||||
|
||||
if (electronicInvoice.container === "xml") {
|
||||
result.skipped.push({ fileId: file.id, reason: "E-Rechnung konnte nicht verarbeitet werden" })
|
||||
continue
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (!itemInfo) {
|
||||
const data = await getInvoiceDataFromGPT(server, file, tenantId, fileData)
|
||||
|
||||
if (!data) {
|
||||
server.log.warn(`GPT returned no data for file ${file.id}`)
|
||||
result.skipped.push({ fileId: file.id, reason: "Keine Belegdaten erkannt" })
|
||||
if (error instanceof FileAlreadyPreparedError) {
|
||||
result.skipped.push({ fileId: file.id, reason: "Datei wurde bereits vorbereitet" })
|
||||
continue
|
||||
}
|
||||
|
||||
itemInfo = mapGptDataToIncomingInvoice(data, tenantId)
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------
|
||||
// 4️⃣ IncomingInvoice erstellen
|
||||
// ---------------------------------------------------------
|
||||
const inserted = await server.db
|
||||
.insert(incominginvoices)
|
||||
.values(itemInfo)
|
||||
.returning()
|
||||
|
||||
const newInvoice = inserted?.[0]
|
||||
|
||||
if (!newInvoice) {
|
||||
server.log.error(`Failed to insert incoming invoice for file ${file.id}`)
|
||||
server.log.error(error, `Datei ${file.id} konnte nicht vorbereitet werden.`)
|
||||
result.skipped.push({ fileId: file.id, reason: "Eingangsbeleg konnte nicht angelegt werden" })
|
||||
continue
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------
|
||||
// 5️⃣ Datei mit incominginvoice-ID verbinden
|
||||
// ---------------------------------------------------------
|
||||
await server.db
|
||||
.update(files)
|
||||
.set({ incominginvoice: newInvoice.id })
|
||||
.where(eq(files.id, file.id))
|
||||
|
||||
result.created.push({ fileId: file.id, invoiceId: newInvoice.id, source: newInvoice.preparationSource ?? null })
|
||||
console.log(`IncomingInvoice ${newInvoice.id} created for file ${file.id}`)
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
|
||||
return result
|
||||
}
|
||||
|
||||
const processFile = async (tenantId: number, file: typeof files.$inferSelect, result: IncomingInvoicePreparationResult) => {
|
||||
console.log(`Processing file ${file.id} for tenant ${tenantId}`)
|
||||
|
||||
let fileData: Buffer
|
||||
|
||||
try {
|
||||
fileData = await loadFileBuffer(file.path!)
|
||||
} catch (error) {
|
||||
server.log.error(error, `Datei ${file.id} konnte nicht aus S3 geladen werden.`)
|
||||
result.skipped.push({ fileId: file.id, reason: "Datei konnte nicht geladen werden" })
|
||||
return
|
||||
}
|
||||
|
||||
const electronicInvoice = await detectElectronicInvoice(fileData, file)
|
||||
|
||||
let itemInfo: any = null
|
||||
|
||||
if (electronicInvoice) {
|
||||
server.log.info({
|
||||
fileId: file.id,
|
||||
container: electronicInvoice.container,
|
||||
syntax: electronicInvoice.syntax,
|
||||
attachmentName: electronicInvoice.attachmentName,
|
||||
}, "Strukturierte E-Rechnung erkannt.")
|
||||
|
||||
try {
|
||||
const parsedInvoice = parseElectronicInvoice(electronicInvoice.xml, electronicInvoice.syntax)
|
||||
const vendor = await findElectronicInvoiceVendor(server, tenantId, parsedInvoice)
|
||||
itemInfo = mapElectronicInvoiceToIncomingInvoice(
|
||||
parsedInvoice,
|
||||
tenantId,
|
||||
vendor?.id || null,
|
||||
electronicInvoice.container,
|
||||
electronicInvoice.attachmentName,
|
||||
)
|
||||
} catch (error) {
|
||||
server.log.error(error, `E-Rechnung aus Datei ${file.id} konnte nicht verarbeitet werden.`)
|
||||
|
||||
if (electronicInvoice.container === "xml") {
|
||||
result.skipped.push({ fileId: file.id, reason: "E-Rechnung konnte nicht verarbeitet werden" })
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (!itemInfo) {
|
||||
const data = await getInvoiceDataFromGPT(server, file, tenantId, fileData)
|
||||
|
||||
if (!data) {
|
||||
server.log.warn(`GPT returned no data for file ${file.id}`)
|
||||
result.skipped.push({ fileId: file.id, reason: "Keine Belegdaten erkannt" })
|
||||
return
|
||||
}
|
||||
|
||||
itemInfo = mapGptDataToIncomingInvoice(data, tenantId)
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------
|
||||
// 4️⃣ IncomingInvoice anlegen und Datei atomar verknüpfen.
|
||||
// Die Verknüpfung greift nur, solange die Datei noch frei ist;
|
||||
// sonst wird der Beleg verworfen, damit keine Dubletten entstehen.
|
||||
// ---------------------------------------------------------
|
||||
const newInvoice = await server.db.transaction(async (tx) => {
|
||||
const [invoice] = await tx
|
||||
.insert(incominginvoices)
|
||||
.values(itemInfo)
|
||||
.returning()
|
||||
|
||||
if (!invoice) throw new Error("Eingangsbeleg konnte nicht angelegt werden")
|
||||
|
||||
const linked = await tx
|
||||
.update(files)
|
||||
.set({ incominginvoice: invoice.id })
|
||||
.where(and(eq(files.id, file.id), isNull(files.incominginvoice)))
|
||||
.returning({ id: files.id })
|
||||
|
||||
if (!linked.length) throw new FileAlreadyPreparedError()
|
||||
|
||||
return invoice
|
||||
})
|
||||
|
||||
result.created.push({ fileId: file.id, invoiceId: newInvoice.id, source: newInvoice.preparationSource ?? null })
|
||||
console.log(`IncomingInvoice ${newInvoice.id} created for file ${file.id}`)
|
||||
}
|
||||
|
||||
// Serialize runs per tenant so concurrent triggers cannot prepare the same file twice.
|
||||
const running = new Map<number, Promise<unknown>>()
|
||||
const backgroundRuns = new Map<number, IncomingInvoicePreparationStatus>()
|
||||
|
||||
return {
|
||||
run: async (tenant:number, options: { fileIds?: string[] } = {}) => {
|
||||
const previous = running.get(tenant) || Promise.resolve()
|
||||
const current = previous.catch(() => undefined).then(() => processInvoices(tenant, options.fileIds))
|
||||
running.set(tenant, current)
|
||||
try {
|
||||
const result = await current
|
||||
console.log("Incoming Invoice Preparation Completed.")
|
||||
return result
|
||||
} finally {
|
||||
if (running.get(tenant) === current) running.delete(tenant)
|
||||
}
|
||||
const run = async (tenant:number, options: { fileIds?: string[] } = {}) => {
|
||||
const previous = running.get(tenant) || Promise.resolve()
|
||||
const current = previous.catch(() => undefined).then(() => processInvoices(tenant, options.fileIds))
|
||||
running.set(tenant, current)
|
||||
try {
|
||||
const result = await current
|
||||
console.log("Incoming Invoice Preparation Completed.")
|
||||
return result
|
||||
} finally {
|
||||
if (running.get(tenant) === current) running.delete(tenant)
|
||||
}
|
||||
}
|
||||
|
||||
const status = (tenant: number): IncomingInvoicePreparationStatus =>
|
||||
backgroundRuns.get(tenant) || { running: false, startedAt: null, finishedAt: null, result: null }
|
||||
|
||||
// Startet die Vorbereitung losgelöst von der Anfrage; ein laufender Vorgang wird nicht erneut gestartet.
|
||||
const start = (tenant: number): IncomingInvoicePreparationStatus => {
|
||||
const current = status(tenant)
|
||||
if (current.running) return current
|
||||
|
||||
const next: IncomingInvoicePreparationStatus = { running: true, startedAt: new Date().toISOString(), finishedAt: null, result: null }
|
||||
backgroundRuns.set(tenant, next)
|
||||
|
||||
run(tenant)
|
||||
.then((result) => {
|
||||
backgroundRuns.set(tenant, { ...next, running: false, finishedAt: new Date().toISOString(), result })
|
||||
})
|
||||
.catch((error) => {
|
||||
server.log.error(error, `Belegvorbereitung für Mandant ${tenant} fehlgeschlagen.`)
|
||||
backgroundRuns.set(tenant, {
|
||||
...next,
|
||||
running: false,
|
||||
finishedAt: new Date().toISOString(),
|
||||
result: { filesFound: 0, created: [], skipped: [], error: "Belegvorbereitung fehlgeschlagen" },
|
||||
})
|
||||
})
|
||||
|
||||
return next
|
||||
}
|
||||
|
||||
return { run, start, status }
|
||||
|
||||
}
|
||||
|
||||
@@ -369,8 +369,11 @@ export default async function functionRoutes(server: FastifyInstance) {
|
||||
})
|
||||
|
||||
server.post('/functions/services/prepareincominginvoices', async (req, reply) => {
|
||||
return reply.code(202).send(server.services.prepareIncomingInvoices.start(req.user.tenant_id))
|
||||
})
|
||||
|
||||
await server.services.prepareIncomingInvoices.run(req.user.tenant_id)
|
||||
server.get('/functions/services/prepareincominginvoices', async (req) => {
|
||||
return server.services.prepareIncomingInvoices.status(req.user.tenant_id)
|
||||
})
|
||||
|
||||
server.get('/functions/liquidity-forecast', async (req, reply) => {
|
||||
|
||||
45
backend/tests/incomingInvoicePreparation.test.ts
Normal file
45
backend/tests/incomingInvoicePreparation.test.ts
Normal file
@@ -0,0 +1,45 @@
|
||||
import assert from "node:assert/strict"
|
||||
import test from "node:test"
|
||||
import { prepareIncomingInvoices } from "../src/modules/cron/prepareIncomingInvoices"
|
||||
|
||||
function server(tenantLookups: { count: number }) {
|
||||
let release: () => void = () => undefined
|
||||
const gate = new Promise<void>((resolve) => { release = resolve })
|
||||
const db: any = {
|
||||
select: () => ({
|
||||
from: () => ({
|
||||
where: () => ({
|
||||
orderBy: async () => {
|
||||
tenantLookups.count += 1
|
||||
await gate
|
||||
return []
|
||||
},
|
||||
}),
|
||||
}),
|
||||
}),
|
||||
}
|
||||
const log = { error: () => undefined, info: () => undefined, warn: () => undefined }
|
||||
return { server: { db, log } as any, release: () => release() }
|
||||
}
|
||||
|
||||
test("start runs the preparation in the background and does not start a second run", async () => {
|
||||
const lookups = { count: 0 }
|
||||
const { server: srv, release } = server(lookups)
|
||||
const service = prepareIncomingInvoices(srv)
|
||||
|
||||
const first = service.start(7)
|
||||
const second = service.start(7)
|
||||
assert.equal(first.running, true)
|
||||
assert.equal(second.startedAt, first.startedAt)
|
||||
assert.equal(service.status(8).running, false)
|
||||
|
||||
release()
|
||||
await new Promise((resolve) => setImmediate(resolve))
|
||||
await new Promise((resolve) => setImmediate(resolve))
|
||||
|
||||
const status = service.status(7)
|
||||
assert.equal(lookups.count, 1)
|
||||
assert.equal(status.running, false)
|
||||
assert.ok(status.finishedAt)
|
||||
assert.equal(status.result?.error, "Mandant nicht gefunden")
|
||||
})
|
||||
Reference in New Issue
Block a user