import type { ImportJob, Prisma } from "@prisma/client"; import { writeAuditLog } from "@/server/audit"; import { emitEvent } from "@/server/events"; import type { DocumentExtractionProvider, WorkOrderExtraction } from "@/server/ai/providers"; import { ServiceError, type ServiceCtx } from "@/server/services/context"; import { emptyExtraction, EXTRACTION_VERSION, type PlausibilityHint, type StoredExtraction, } from "@/lib/imports/extraction"; import { checkPlausibility } from "@/lib/imports/plausibility"; // TODO(L3→L1): replace with "@/lib/customers/duplicates" after the L1 merge (same interface). import { findDuplicateCustomers, findSiteCandidates } from "./duplicates-stub"; export type ProcessDeps = { provider: DocumentExtractionProvider | null; loadBytes: (doc: { storageKey: string }) => Promise; now?: Date; }; /** * Extraction pipeline (spec §9.3 steps 3–7): processing → provider (text + fields) → * plausibility → duplicate customers + site candidates → review_required. * Without provider: review_required with empty extraction and hint "manual_entry". * Any error: failed + event import.failed. Never creates customers or work orders. * Idempotent: only jobs in uploaded/processing are processed. */ export async function processImport(ctx: ServiceCtx, importId: string, deps: ProcessDeps): Promise { const job = await ctx.db.importJob.findFirst({ where: { id: importId } }); if (!job) throw new ServiceError("not_found", "import not found"); if (job.status !== "uploaded" && job.status !== "processing") return job; const actorId = job.importedById ?? undefined; await ctx.db.importJob.update({ where: { id: job.id }, data: { status: "processing", errorMessage: null } }); try { const doc = await ctx.db.document.findFirst({ where: { id: job.documentId } }); if (!doc) throw new Error("document_missing"); let fields: WorkOrderExtraction; let hints: PlausibilityHint[] = []; let text: string | null = null; let providerName: string | null = null; let model: string | null = null; if (!deps.provider) { fields = emptyExtraction(); hints = [{ field: null, code: "manual_entry" }]; } else { const bytes = await deps.loadBytes({ storageKey: doc.storageKey }); if (!bytes) throw new Error("file_unavailable"); const result = await deps.provider.extract({ bytes, mimeType: doc.mimeType, fileName: doc.fileName }); const checked = checkPlausibility(result.extraction, deps.now); fields = checked.extraction; hints = checked.hints; text = result.text; providerName = result.meta.provider; model = result.meta.model; await ctx.db.aiGeneration.create({ data: { tenantId: ctx.tenantId, kind: "import_extraction", provider: result.meta.provider, model: result.meta.model, entityType: "import_job", entityId: job.id, input: { documentId: doc.id, fileName: doc.fileName, mimeType: doc.mimeType, fileSize: doc.fileSize, checksum: doc.checksum }, output: { extraction: result.extraction, hints } as unknown as Prisma.InputJsonValue, inputTokens: result.meta.inputTokens ?? null, outputTokens: result.meta.outputTokens ?? null, createdById: actorId ?? null, }, }); } const address = fields.customerAddress.value ?? {}; const duplicates = await findDuplicateCustomers(ctx.db, { customerNumber: fields.customerNumber.value, companyName: fields.companyName.value, firstName: fields.customerFirstName.value, lastName: fields.customerLastName.value, street: address.street, houseNumber: address.houseNumber, postalCode: address.postalCode, city: address.city, email: fields.email.value, phone: fields.phone.value, }); const siteAddress = fields.siteAddress.value ?? (fields.siteName.value ? {} : null); const siteCandidates = await findSiteCandidates( ctx.db, duplicates.map((d) => d.customerId), siteAddress ? { ...siteAddress, name: fields.siteName.value } : null, ); const stored: StoredExtraction = { fields, hints, siteCandidates }; const updated = await ctx.db.importJob.update({ where: { id: job.id }, data: { status: "review_required", errorMessage: null, extractedText: text, extraction: stored as unknown as Prisma.InputJsonValue, extractionVersion: EXTRACTION_VERSION, extractionModel: model, provider: providerName, duplicateCandidates: duplicates as unknown as Prisma.InputJsonValue, }, }); await writeAuditLog({ tenantId: ctx.tenantId, actorId, action: "update", entity: "import_job", entityId: job.id, before: { status: job.status }, after: { status: updated.status, provider: providerName, model, hints: hints.length, duplicateCandidates: duplicates.length }, }); await emitEvent(ctx, { type: "import.ready_for_review", entityType: "import_job", entityId: job.id, data: { manual: !deps.provider, duplicateCandidates: duplicates.length }, }); return updated; } catch (err) { const message = ((err as Error).message || "extraction_failed").slice(0, 500); console.error(`[imports] processing ${job.id} failed:`, message); const failed = await ctx.db.importJob.update({ where: { id: job.id }, data: { status: "failed", errorMessage: message } }); await writeAuditLog({ tenantId: ctx.tenantId, actorId, action: "update", entity: "import_job", entityId: job.id, before: { status: job.status }, after: { status: "failed", errorMessage: message }, }); await emitEvent(ctx, { type: "import.failed", entityType: "import_job", entityId: job.id, data: { reason: message.slice(0, 120) } }); return failed; } }