feat(documents): prepare inbox suggestions and add entity assignments

backend/src/modules/scanner-inbox-assignments.ts

@@ -0,0 +1,70 @@
+import { and, asc, eq, ilike, isNull, sql, type SQL } from "drizzle-orm"
+import type { AnyPgColumn, AnyPgTable } from "drizzle-orm/pg-core"
+import type { FastifyInstance } from "fastify"
+import { authProfiles, checks, contracts, createddocuments, customers, incominginvoices, inventoryitems, plants, products, projects, vehicles, vendors } from "../../db/schema"
+
+type Target = { label: string; table: AnyPgTable; id: AnyPgColumn; tenant: AnyPgColumn; name: AnyPgColumn; available: SQL; uuid?: boolean }
+export const assignmentTargets: Record<string, Target> = {
+    project: { label: "Projekt", table: projects, id: projects.id, tenant: projects.tenant, name: projects.name, available: eq(projects.archived, false) },
+    customer: { label: "Kunde", table: customers, id: customers.id, tenant: customers.tenant, name: customers.name, available: eq(customers.archived, false) },
+    vendor: { label: "Lieferant", table: vendors, id: vendors.id, tenant: vendors.tenant, name: vendors.name, available: eq(vendors.archived, false) },
+    vehicle: { label: "Fahrzeug", table: vehicles, id: vehicles.id, tenant: vehicles.tenant, name: vehicles.license_plate, available: eq(vehicles.archived, false) },
+    plant: { label: "Objekt", table: plants, id: plants.id, tenant: plants.tenant, name: plants.name, available: eq(plants.archived, false) },
+    contract: { label: "Vertrag", table: contracts, id: contracts.id, tenant: contracts.tenant, name: contracts.name, available: eq(contracts.archived, false) },
+    product: { label: "Produkt", table: products, id: products.id, tenant: products.tenant, name: products.name, available: eq(products.archived, false) },
+    createddocument: { label: "Ausgangsbeleg", table: createddocuments, id: createddocuments.id, tenant: createddocuments.tenant, name: createddocuments.documentNumber, available: eq(createddocuments.archived, false) },
+    incominginvoice: { label: "Eingangsrechnung", table: incominginvoices, id: incominginvoices.id, tenant: incominginvoices.tenant, name: incominginvoices.reference, available: eq(incominginvoices.archived, false) },
+    inventoryitem: { label: "Inventarartikel", table: inventoryitems, id: inventoryitems.id, tenant: inventoryitems.tenant, name: inventoryitems.name, available: eq(inventoryitems.archived, false) },
+    check: { label: "Überprüfung", table: checks, id: checks.id, tenant: checks.tenant, name: checks.name, available: eq(checks.archived, false), uuid: true },
+    authProfile: { label: "Mitarbeiter", table: authProfiles, id: authProfiles.id, tenant: authProfiles.tenant_id, name: authProfiles.full_name,
+        available: and(eq(authProfiles.active, true), isNull(authProfiles.customer_for_portal)), uuid: true },
+}
+export const assignmentKinds = Object.entries(assignmentTargets).map(([value, target]) => ({ value, label: target.label }))
+export type FileAssignments = Record<string, string | number | null>
+const invalidAssignment = () => Object.assign(new Error("Die gewählte Zuordnung ist nicht verfügbar."), { statusCode: 400 })
+const targetFor = (kind: string) => {
+    if (!Object.prototype.hasOwnProperty.call(assignmentTargets, kind)) throw invalidAssignment()
+    return assignmentTargets[kind]
+}
+export function assignmentId(kind: string, id: unknown) {
+    const target = targetFor(kind)
+    if (target.uuid) {
+        if (typeof id !== "string" || !/^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i.test(id)) throw invalidAssignment()
+        return id
+    }
+    if (!/^[1-9]\d*$/.test(String(id)) || !Number.isSafeInteger(Number(id))) throw invalidAssignment()
+    return Number(id)
+}
+type Reader = Pick<FastifyInstance["db"], "select">
+export async function assignmentOptions(db: Reader, tenantId: number, kind: string, search = "") {
+    const target = targetFor(kind)
+    const name = sql<string>`coalesce(nullif(${target.name}, ''), '#' || ${target.id}::text)`
+    const rows = await db.select({ id: target.id, name }).from(target.table)
+        .where(and(eq(target.tenant, tenantId), target.available,
+            search ? ilike(name, `%${search.replace(/[\\%_]/g, "\\$&")}%`) : undefined))
+        .orderBy(asc(target.name), asc(target.id)).limit(100)
+    return rows.map(row => ({ id: String(row.id), name: row.name }))
+}
+export async function currentAssignments(db: Reader, tenantId: number, file: Record<string, unknown>) {
+    const assignments = []
+    for (const [kind, target] of Object.entries(assignmentTargets)) {
+        if (file[kind] == null) continue
+        const [entry] = await db.select({ name: target.name }).from(target.table)
+            .where(and(eq(target.id, file[kind]), eq(target.tenant, tenantId), target.available)).limit(1)
+        assignments.push({ kind, id: String(file[kind]), name: entry?.name || (entry ? `#${file[kind]}` : "Nicht mehr verfügbar"), available: Boolean(entry) })
+    }
+    return assignments
+}
+export async function validateAssignments(tx: Reader, tenantId: number, assignments: FileAssignments) {
+    const changes: FileAssignments = {}
+    for (const [kind, value] of Object.entries(assignments)) {
+        const target = targetFor(kind)
+        if (value === null) { changes[kind] = null; continue }
+        const id = assignmentId(kind, value)
+        const [row] = await tx.select({ id: target.id }).from(target.table)
+            .where(and(eq(target.id, id), eq(target.tenant, tenantId), target.available)).for("share").limit(1)
+        if (!row) throw invalidAssignment()
+        changes[kind] = id
+    }
+    return changes
+}

backend/src/modules/scanner-inbox.service.ts

@@ -1,12 +1,13 @@
 import { createHash, randomBytes, randomUUID, timingSafeEqual } from "node:crypto"
-import { and, asc, count, desc, eq, ilike, isNull, sql } from "drizzle-orm"
+import { and, asc, count, desc, eq, ilike, inArray, isNull, sql } from "drizzle-orm"
 import type { FastifyInstance } from "fastify"
 import { DeleteObjectCommand, PutObjectCommand } from "@aws-sdk/client-s3"
 import { files, filetags, folders, scannerInboxConnections, scannerInboxDocuments, tenants } from "../../db/schema"
 import { sanitizeFilename } from "../utils/filename"
 import { s3 } from "../utils/s3"
 import { secrets } from "../utils/secrets"
-import { suggestDocumentFiling, validateFilingSuggestion, type FilingSuggestionResult } from "./scanner-inbox-suggestions"
+import { suggestDocumentFiling, validateFilingSuggestion } from "./scanner-inbox-suggestions"
+import { assignmentKinds, assignmentOptions, currentAssignments, validateAssignments, type FileAssignments } from "./scanner-inbox-assignments"
 
 export const SCANNER_UPLOAD_LIMIT = 100 * 1024 * 1024
 export const SCANNER_DAV_PATH = "/scanner-inbox/"
@@ -47,7 +48,6 @@ export function validateScannerFile(filename: string, body: Buffer) {
 }
 
 export function createScannerInboxService(db: FastifyInstance["db"], storage: ScannerStorage = objectStorage, analyze = suggestDocumentFiling) {
-    const pendingSuggestions = new Map<string, Promise<FilingSuggestionResult>>()
     const publicConnection = {
         username: scannerInboxConnections.username,
         active: scannerInboxConnections.active,
@@ -93,13 +93,19 @@ export function createScannerInboxService(db: FastifyInstance["db"], storage: Sc
             isNull(scannerInboxDocuments.filedAt), eq(files.archived, false),
             search ? ilike(scannerInboxDocuments.remoteFilename, `%${search.replace(/[\\%_]/g, "\\$&")}%`) : undefined)
         const documents = await db.select({ id: files.id, name: scannerInboxDocuments.remoteFilename,
-            mimeType: files.mimeType, size: files.size, receivedAt: scannerInboxDocuments.receivedAt })
+            mimeType: files.mimeType, size: files.size, receivedAt: scannerInboxDocuments.receivedAt, suggestionStatus: scannerInboxDocuments.suggestionStatus })
             .from(scannerInboxDocuments).innerJoin(files, eq(files.id, scannerInboxDocuments.fileId))
             .where(condition).orderBy(desc(scannerInboxDocuments.receivedAt), desc(files.id)).limit(limit).offset(offset)
         const [total] = await db.select({ value: count() }).from(scannerInboxDocuments)
             .innerJoin(files, eq(files.id, scannerInboxDocuments.fileId)).where(condition)
         return { documents, total: total.value }
     }
+    const openCount = async (tenantId: number) => {
+        const [row] = await db.select({ total: count() }).from(scannerInboxDocuments)
+            .innerJoin(files, eq(files.id, scannerInboxDocuments.fileId))
+            .where(and(eq(scannerInboxDocuments.tenantId, tenantId), eq(files.tenant, tenantId), isNull(scannerInboxDocuments.filedAt), eq(files.archived, false)))
+        return row
+    }
     const filingOptions = async (tenantId: number) => {
         const [availableFolders, types] = await Promise.all([
             db.select({ id: folders.id, name: folders.name, parent: folders.parent,
@@ -108,33 +114,77 @@ export function createScannerInboxService(db: FastifyInstance["db"], storage: Sc
             db.select({ id: filetags.id, name: filetags.name }).from(filetags)
                 .where(and(eq(filetags.tenant, tenantId), eq(filetags.archived, false))).orderBy(asc(filetags.name)),
         ])
-        return { folders: availableFolders, types }
+        return { folders: availableFolders, types, assignmentKinds }
     }
     const suggestionDocument = async (tenantId: number, fileId: string) => {
-        const [document] = await db.select({ path: files.path, name: scannerInboxDocuments.remoteFilename, mimeType: files.mimeType, size: files.size })
+        const [document] = await db.select({ path: files.path, name: scannerInboxDocuments.remoteFilename, mimeType: files.mimeType, size: files.size,
+            status: scannerInboxDocuments.suggestionStatus, suggestion: scannerInboxDocuments.suggestion, message: scannerInboxDocuments.suggestionMessage })
             .from(scannerInboxDocuments).innerJoin(files, eq(files.id, scannerInboxDocuments.fileId))
             .where(and(eq(scannerInboxDocuments.tenantId, tenantId), eq(files.tenant, tenantId), eq(files.id, fileId),
                 isNull(scannerInboxDocuments.filedAt), eq(files.archived, false))).limit(1)
         if (!document?.path) throw scannerError(404, "Dokument nicht in der Ablage gefunden.")
         return document
     }
-    const suggestFiling = async (tenantId: number, fileId: string): Promise<FilingSuggestionResult> => {
-        // Check tenant ownership before reading storage or submitting anything to the AI provider.
+    const getSuggestion = async (tenantId: number, fileId: string) => {
         const document = await suggestionDocument(tenantId, fileId)
-        const key = `${tenantId}:${fileId}`
-        if (pendingSuggestions.has(key)) return pendingSuggestions.get(key)!
-        if (pendingSuggestions.size >= 4) throw scannerError(429, "Die KI verarbeitet gerade weitere Dokumente. Bitte gleich erneut versuchen.")
-        const operation = (async () => {
-            const result = await analyze(document, await filingOptions(tenantId))
-            await suggestionDocument(tenantId, fileId)
-            const suggestion = validateFilingSuggestion(result, await filingOptions(tenantId))
-            return { suggestion, message: suggestion ? undefined : result.message || "Keine passende Zuordnung verfügbar. Bitte manuell auswählen." }
-        })()
-        pendingSuggestions.set(key, operation)
-        try { return await operation }
-        finally { pendingSuggestions.delete(key) }
+        const suggestion = document.status === "ready"
+            ? validateFilingSuggestion({ suggestion: document.suggestion }, await filingOptions(tenantId)) : null
+        return { status: document.status, suggestion, message: document.status === "ready" && !suggestion
+            ? document.message || "Keine passende Zuordnung verfügbar. Bitte manuell auswählen." : document.message }
     }
-    const fileDocument = async (tenantId: number, userId: string, fileId: string, folderId: string | null, typeId: string) => db.transaction(async tx => {
+    const queueSuggestion = async (tenantId: number, fileId: string) => {
+        await suggestionDocument(tenantId, fileId)
+        await db.update(scannerInboxDocuments).set({ suggestionStatus: "pending", suggestion: null, suggestionMessage: null,
+            suggestionClaim: null, suggestionUpdatedAt: new Date() })
+            .where(and(eq(scannerInboxDocuments.tenantId, tenantId), eq(scannerInboxDocuments.fileId, fileId),
+                isNull(scannerInboxDocuments.filedAt), inArray(scannerInboxDocuments.suggestionStatus, ["ready", "failed"])))
+        return getSuggestion(tenantId, fileId)
+    }
+    const processNextSuggestion = async () => {
+        const claim = randomUUID()
+        // Claim without holding a transaction open while OCR/the provider runs. Expired claims survive restarts.
+        const claimed = await db.execute<{ fileId: string; tenantId: string }>(sql`
+            UPDATE scanner_inbox_documents AS job
+            SET suggestion_status = 'processing', suggestion_claim = ${claim}, suggestion_updated_at = now()
+            WHERE job.file_id = (
+                SELECT pending.file_id FROM scanner_inbox_documents AS pending
+                JOIN files ON files.id = pending.file_id AND files.tenant = pending.tenant_id
+                JOIN tenants ON tenants.id = pending.tenant_id
+                WHERE pending.filed_at IS NULL AND files.archived = false
+                    AND tenants.locked IS DISTINCT FROM 'maintenance_tenant'
+                    AND (pending.suggestion_status = 'pending' OR
+                        (pending.suggestion_status = 'processing' AND pending.suggestion_updated_at < now() - interval '5 minutes'))
+                ORDER BY pending.received_at, pending.file_id FOR UPDATE OF pending SKIP LOCKED LIMIT 1
+            ) RETURNING job.file_id AS "fileId", job.tenant_id AS "tenantId"`)
+        const job = claimed.rows[0]
+        if (!job) return false
+        const tenantId = Number(job.tenantId)
+        const condition = and(eq(scannerInboxDocuments.fileId, job.fileId), eq(scannerInboxDocuments.tenantId, tenantId),
+            eq(scannerInboxDocuments.suggestionClaim, claim), isNull(scannerInboxDocuments.filedAt))
+        try {
+            const document = await suggestionDocument(tenantId, job.fileId)
+            const { path, name, mimeType, size } = document
+            const result = await analyze({ path, name, mimeType, size }, await filingOptions(tenantId))
+            await suggestionDocument(tenantId, job.fileId)
+            const suggestion = validateFilingSuggestion(result, await filingOptions(tenantId))
+            await db.update(scannerInboxDocuments).set({ suggestionStatus: "ready", suggestion,
+                suggestionMessage: suggestion ? null : result.message || "Keine passende Zuordnung gefunden. Bitte manuell auswählen.",
+                suggestionClaim: null, suggestionUpdatedAt: new Date() }).where(condition)
+        } catch (error: any) {
+            const message = error?.statusCode === 503 ? "Die KI-Erkennung ist noch nicht konfiguriert. Bitte manuell zuordnen oder später erneut prüfen."
+                : "Die KI-Vorbereitung ist fehlgeschlagen. Bitte erneut prüfen oder manuell zuordnen."
+            await db.update(scannerInboxDocuments).set({ suggestionStatus: "failed", suggestion: null, suggestionMessage: message,
+                suggestionClaim: null, suggestionUpdatedAt: new Date() }).where(condition)
+        }
+        return true
+    }
+    const getAssignments = async (tenantId: number, fileId: string) => {
+        await suggestionDocument(tenantId, fileId)
+        const [file] = await db.select().from(files).where(and(eq(files.id, fileId), eq(files.tenant, tenantId), eq(files.archived, false))).limit(1)
+        if (!file) throw scannerError(404, "Dokument nicht gefunden.")
+        return currentAssignments(db, tenantId, file)
+    }
+    const fileDocument = async (tenantId: number, userId: string, fileId: string, folderId: string | null, typeId: string, assignments: FileAssignments = {}) => db.transaction(async tx => {
         const [tenant] = await tx.select({ locked: tenants.locked }).from(tenants)
             .where(eq(tenants.id, tenantId)).for("share")
         if (!tenant) throw scannerError(404, "Mandant nicht gefunden.")
@@ -157,11 +207,12 @@ export function createScannerInboxService(db: FastifyInstance["db"], storage: Sc
         const [type] = await tx.select({ id: filetags.id }).from(filetags)
             .where(and(eq(filetags.id, typeId), eq(filetags.tenant, tenantId), eq(filetags.archived, false))).for("share")
         if (!type) throw scannerError(400, "Der gewählte Dokumenttyp ist nicht verfügbar.")
+        const assignmentChanges = await validateAssignments(tx, tenantId, assignments)
         const now = new Date()
-        await tx.update(files).set({ folder: folderId, type: typeId, updatedAt: now, updatedBy: userId })
+        await tx.update(files).set({ ...assignmentChanges, folder: folderId, type: typeId, updatedAt: now, updatedBy: userId })
             .where(and(eq(files.id, fileId), eq(files.tenant, tenantId)))
         // Keep the receipt so a scanner retry cannot recreate a document after filing.
-        await tx.update(scannerInboxDocuments).set({ filedAt: now })
+        await tx.update(scannerInboxDocuments).set({ filedAt: now, suggestionClaim: null })
             .where(and(eq(scannerInboxDocuments.fileId, fileId), eq(scannerInboxDocuments.tenantId, tenantId)))
         return { id: fileId, folder: folderId, type: typeId }
     })
@@ -225,7 +276,8 @@ export function createScannerInboxService(db: FastifyInstance["db"], storage: Sc
             throw error
         }
     }
-    return { connection, credentials, disable, authenticate, find, list, filingOptions, suggestFiling, fileDocument, probe, accept }
+    return { connection, credentials, disable, authenticate, find, list, openCount, filingOptions, getSuggestion, queueSuggestion, processNextSuggestion,
+        getAssignments, assignmentOptions: (tenantId: number, kind: string, search: string) => assignmentOptions(db, tenantId, kind, search), fileDocument, probe, accept }
 }
 
 export type ScannerInboxService = ReturnType<typeof createScannerInboxService>

backend/src/modules/scanner-inbox.worker.ts

@@ -0,0 +1,27 @@
+import type { FastifyInstance } from "fastify"
+import { createScannerInboxService, type ScannerInboxService } from "./scanner-inbox.service"
+
+export function startScannerInboxWorker(server: FastifyInstance, service: Pick<ScannerInboxService, "processNextSuggestion"> = createScannerInboxService(server.db)) {
+    let stopped = false
+    let timer: ReturnType<typeof setTimeout> | undefined
+    let running: Promise<void> | undefined
+    const run = async () => {
+        try {
+            // One analysis per process; database claims coordinate multiple backend instances.
+            while (!stopped && await service.processNextSuggestion()) { /* drain the durable queue */ }
+        } catch {
+            server.log.error("KI-Vorbereitung der Dokumentenablage unterbrochen; erneuter Versuch folgt.")
+        } finally {
+            if (!stopped) {
+                timer = setTimeout(() => { running = run() }, 2000)
+                timer.unref()
+            }
+        }
+    }
+    server.addHook("onReady", async () => { running = run() })
+    server.addHook("onClose", async () => {
+        stopped = true
+        clearTimeout(timer)
+        await running
+    })
+}