307 lines
22 KiB
TypeScript
307 lines
22 KiB
TypeScript
import { createHash, randomBytes, randomUUID, timingSafeEqual } from "node:crypto"
|
|
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 } 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/"
|
|
export const scannerError = (statusCode: number, message: string) => Object.assign(new Error(message), { statusCode })
|
|
const digest = (value: string | Buffer) => createHash("sha256").update(value).digest("hex")
|
|
|
|
export type ScannerIdentity = { tenantId: number; username: string; passwordHash: string }
|
|
export interface ScannerStorage {
|
|
put(key: string, body: Buffer, contentType: string): Promise<void>
|
|
remove(key: string): Promise<void>
|
|
}
|
|
|
|
const objectStorage: ScannerStorage = {
|
|
async put(key, body, contentType) {
|
|
await s3.send(new PutObjectCommand({ Bucket: secrets.S3_BUCKET, Key: key, Body: body, ContentType: contentType, ContentLength: body.length }))
|
|
},
|
|
async remove(key) {
|
|
await s3.send(new DeleteObjectCommand({ Bucket: secrets.S3_BUCKET, Key: key }))
|
|
},
|
|
}
|
|
|
|
function validateScannerFilename(filename: string) {
|
|
if (!filename || filename.length > 200 || /[\\/\u0000-\u001f\u007f]/.test(filename) || filename.startsWith(".")) {
|
|
throw scannerError(400, "Bitte einen einzelnen Dateinamen ohne Unterordner verwenden.")
|
|
}
|
|
}
|
|
|
|
export function validateScannerFile(filename: string, body: Buffer) {
|
|
validateScannerFilename(filename)
|
|
if (!body.length) throw scannerError(400, "Die Datei ist leer.")
|
|
if (body.length > SCANNER_UPLOAD_LIMIT) throw scannerError(413, "Die Datei überschreitet 100 MB.")
|
|
const extension = filename.split(".").pop()?.toLowerCase()
|
|
if (extension === "pdf" && body.subarray(0, 5).toString() === "%PDF-" && body.subarray(-1024).includes(Buffer.from("%%EOF"))) return "application/pdf"
|
|
if (["jpg", "jpeg"].includes(extension) && body[0] === 0xff && body[1] === 0xd8 && body[2] === 0xff) return "image/jpeg"
|
|
if (extension === "png" && body.subarray(0, 8).equals(Buffer.from([137, 80, 78, 71, 13, 10, 26, 10]))) return "image/png"
|
|
if (["tif", "tiff"].includes(extension) && ["49492a00", "4d4d002a"].includes(body.subarray(0, 4).toString("hex"))) return "image/tiff"
|
|
throw scannerError(415, "Bitte eine vollständige PDF-, JPEG-, PNG- oder TIFF-Datei übertragen.")
|
|
}
|
|
|
|
export function createScannerInboxService(db: FastifyInstance["db"], storage: ScannerStorage = objectStorage, analyze = suggestDocumentFiling) {
|
|
const publicConnection = {
|
|
username: scannerInboxConnections.username,
|
|
active: scannerInboxConnections.active,
|
|
createdAt: scannerInboxConnections.createdAt,
|
|
}
|
|
const connection = async (tenantId: number) => {
|
|
const [row] = await db.select(publicConnection).from(scannerInboxConnections)
|
|
.where(eq(scannerInboxConnections.tenantId, tenantId)).limit(1)
|
|
return row || null
|
|
}
|
|
const credentials = async (tenantId: number) => {
|
|
// Epson ES-580W destination passwords are limited to 20 characters (120 random bits).
|
|
const password = randomBytes(15).toString("base64url")
|
|
const [row] = await db.insert(scannerInboxConnections).values({
|
|
tenantId, username: `scan_${randomBytes(8).toString("hex")}`, passwordHash: digest(password),
|
|
}).onConflictDoUpdate({
|
|
target: scannerInboxConnections.tenantId,
|
|
set: { passwordHash: digest(password), active: true, updatedAt: new Date() },
|
|
}).returning(publicConnection)
|
|
return { ...row, password }
|
|
}
|
|
const disable = async (tenantId: number) => {
|
|
await db.update(scannerInboxConnections).set({ active: false, updatedAt: new Date() })
|
|
.where(eq(scannerInboxConnections.tenantId, tenantId))
|
|
}
|
|
const authenticate = async (username: string, password: string): Promise<ScannerIdentity | null> => {
|
|
const [row] = await db.select().from(scannerInboxConnections).where(and(
|
|
eq(scannerInboxConnections.username, username), eq(scannerInboxConnections.active, true),
|
|
)).limit(1)
|
|
const expected = row?.passwordHash || "0".repeat(64)
|
|
const valid = timingSafeEqual(Buffer.from(digest(password)), Buffer.from(expected))
|
|
return row && valid ? { tenantId: row.tenantId, username: row.username, passwordHash: row.passwordHash } : null
|
|
}
|
|
const find = async (tenantId: number, filename: string) => {
|
|
const [row] = await db.select({ size: files.size, mimeType: files.mimeType, checksum: scannerInboxDocuments.checksum,
|
|
receivedAt: scannerInboxDocuments.receivedAt }).from(scannerInboxDocuments)
|
|
.innerJoin(files, and(eq(files.id, scannerInboxDocuments.fileId), eq(files.tenant, scannerInboxDocuments.tenantId)))
|
|
.where(and(eq(scannerInboxDocuments.tenantId, tenantId), eq(scannerInboxDocuments.remoteFilename, filename))).limit(1)
|
|
return row || null
|
|
}
|
|
const list = async (tenantId: number, limit: number, offset: number, search: string) => {
|
|
const condition = and(eq(scannerInboxDocuments.tenantId, tenantId), eq(files.tenant, tenantId),
|
|
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, 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,
|
|
standardFiletype: folders.standardFiletype, standardFiletypeIsOptional: folders.standardFiletypeIsOptional })
|
|
.from(folders).where(and(eq(folders.tenant, tenantId), eq(folders.archived, false))).orderBy(asc(folders.name)),
|
|
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, 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,
|
|
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 getSuggestion = async (tenantId: number, fileId: string) => {
|
|
const document = await suggestionDocument(tenantId, fileId)
|
|
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 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 archiveDocument = async (tenantId: number, userId: string, fileId: string) => 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.")
|
|
if (tenant.locked === "maintenance_tenant") throw scannerError(423, "Der Mandant wird gerade gewartet.")
|
|
// Use the same lock order as filing so both actions cannot succeed concurrently.
|
|
const [entry] = await tx.select({ filedAt: scannerInboxDocuments.filedAt }).from(scannerInboxDocuments)
|
|
.where(and(eq(scannerInboxDocuments.tenantId, tenantId), eq(scannerInboxDocuments.fileId, fileId))).for("update")
|
|
if (!entry) throw scannerError(404, "Dokument nicht in der Ablage gefunden.")
|
|
if (entry.filedAt) throw scannerError(409, "Dieses Dokument wurde bereits abgelegt. Bitte den Eingang aktualisieren.")
|
|
const [file] = await tx.select({ archived: files.archived }).from(files)
|
|
.where(and(eq(files.id, fileId), eq(files.tenant, tenantId))).for("update")
|
|
if (!file) throw scannerError(404, "Dokument nicht gefunden.")
|
|
if (!file.archived) {
|
|
await tx.update(files).set({ archived: true, updatedAt: new Date(), updatedBy: userId })
|
|
.where(and(eq(files.id, fileId), eq(files.tenant, tenantId)))
|
|
}
|
|
// Keep the receipt for scanner retries; invalidate any analysis already in flight.
|
|
await tx.update(scannerInboxDocuments).set({ suggestionClaim: null,
|
|
suggestionStatus: sql`CASE WHEN ${scannerInboxDocuments.suggestionStatus} = 'processing' THEN 'pending' ELSE ${scannerInboxDocuments.suggestionStatus} END` })
|
|
.where(and(eq(scannerInboxDocuments.fileId, fileId), eq(scannerInboxDocuments.tenantId, tenantId)))
|
|
return { id: fileId, archived: true }
|
|
})
|
|
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.")
|
|
if (tenant.locked === "maintenance_tenant") throw scannerError(423, "Der Mandant wird gerade gewartet.")
|
|
const [entry] = await tx.select({ filedAt: scannerInboxDocuments.filedAt }).from(scannerInboxDocuments)
|
|
.where(and(eq(scannerInboxDocuments.tenantId, tenantId), eq(scannerInboxDocuments.fileId, fileId))).for("update")
|
|
if (!entry) throw scannerError(404, "Dokument nicht in der Ablage gefunden.")
|
|
if (entry.filedAt) throw scannerError(409, "Dieses Dokument wurde bereits abgelegt. Bitte den Eingang aktualisieren.")
|
|
const [file] = await tx.select({ id: files.id }).from(files)
|
|
.where(and(eq(files.id, fileId), eq(files.tenant, tenantId), eq(files.archived, false))).for("update")
|
|
if (!file) throw scannerError(404, "Dokument nicht gefunden oder archiviert.")
|
|
if (folderId) {
|
|
const [folder] = await tx.select({ standardFiletype: folders.standardFiletype, optional: folders.standardFiletypeIsOptional })
|
|
.from(folders).where(and(eq(folders.id, folderId), eq(folders.tenant, tenantId), eq(folders.archived, false))).for("share")
|
|
if (!folder) throw scannerError(400, "Der gewählte Ordner ist nicht verfügbar.")
|
|
if (folder.standardFiletype && !folder.optional && folder.standardFiletype !== typeId) {
|
|
throw scannerError(400, "Dieser Ordner gibt einen festen Dokumenttyp vor.")
|
|
}
|
|
}
|
|
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({ ...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, suggestionClaim: null })
|
|
.where(and(eq(scannerInboxDocuments.fileId, fileId), eq(scannerInboxDocuments.tenantId, tenantId)))
|
|
return { id: fileId, folder: folderId, type: typeId }
|
|
})
|
|
const assertWritable = async (tx: Pick<FastifyInstance["db"], "select">, identity: ScannerIdentity) => {
|
|
// Hold these locks until the write or write probe has finished.
|
|
const [tenant] = await tx.select({ locked: tenants.locked }).from(tenants)
|
|
.where(eq(tenants.id, identity.tenantId)).for("share")
|
|
if (!tenant) throw scannerError(401, "Scanner-Zugang ungültig.")
|
|
if (tenant.locked === "maintenance_tenant") throw scannerError(423, "Der Mandant wird gerade gewartet.")
|
|
const [active] = await tx.select({ tenantId: scannerInboxConnections.tenantId }).from(scannerInboxConnections)
|
|
.where(and(eq(scannerInboxConnections.tenantId, identity.tenantId), eq(scannerInboxConnections.username, identity.username),
|
|
eq(scannerInboxConnections.passwordHash, identity.passwordHash), eq(scannerInboxConnections.active, true))).for("share")
|
|
if (!active) throw scannerError(401, "Scanner-Zugang ungültig.")
|
|
}
|
|
const probe = async (identity: ScannerIdentity, filename: string, onlyIfNew = false) => {
|
|
validateScannerFilename(filename)
|
|
if (!/\.(pdf|jpe?g|png|tiff?)$/i.test(filename)) {
|
|
throw scannerError(415, "Bitte eine vollständige PDF-, JPEG-, PNG- oder TIFF-Datei übertragen.")
|
|
}
|
|
await db.transaction(async tx => {
|
|
await assertWritable(tx, identity)
|
|
await tx.execute(sql`SELECT pg_advisory_xact_lock(hashtextextended(${`${identity.tenantId}:${filename}`}, 0))`)
|
|
const [existing] = await tx.select({ fileId: scannerInboxDocuments.fileId }).from(scannerInboxDocuments)
|
|
.where(and(eq(scannerInboxDocuments.tenantId, identity.tenantId), eq(scannerInboxDocuments.remoteFilename, filename))).limit(1)
|
|
if (existing) throw scannerError(onlyIfNew ? 412 : 409, "Der Dateiname ist bereits vorhanden.")
|
|
// Epson probes writability with an empty PUT. No file, receipt or reservation is stored.
|
|
})
|
|
}
|
|
const accept = async (identity: ScannerIdentity, filename: string, body: Buffer, onlyIfNew = false) => {
|
|
const contentType = validateScannerFile(filename, body)
|
|
const checksum = digest(body)
|
|
let uploadedKey: string | undefined
|
|
try {
|
|
return await db.transaction(async tx => {
|
|
await assertWritable(tx, identity)
|
|
|
|
// Serialise retries across processes; a lost HTTP response must not duplicate a document.
|
|
await tx.execute(sql`SELECT pg_advisory_xact_lock(hashtextextended(${`${identity.tenantId}:${filename}`}, 0))`)
|
|
const [existing] = await tx.select().from(scannerInboxDocuments).where(and(
|
|
eq(scannerInboxDocuments.tenantId, identity.tenantId), eq(scannerInboxDocuments.remoteFilename, filename),
|
|
)).limit(1)
|
|
if (existing) {
|
|
if (onlyIfNew) throw scannerError(412, "Der Dateiname ist bereits vorhanden.")
|
|
if (existing.checksum !== checksum) throw scannerError(409, "Der Dateiname ist bereits vergeben. Bitte einen eindeutigen Dateinamen verwenden.")
|
|
return { created: false, checksum }
|
|
}
|
|
const id = randomUUID()
|
|
const key = `${identity.tenantId}/filesbyid/${id}/${sanitizeFilename(filename)}`
|
|
uploadedKey = key
|
|
await storage.put(key, body, contentType)
|
|
// Bypass saveFile's text extraction: the inbox only stores the original document.
|
|
await tx.insert(files).values({ id, tenant: identity.tenantId, name: filename, path: key, mimeType: contentType, size: body.length })
|
|
await tx.insert(scannerInboxDocuments).values({ fileId: id, tenantId: identity.tenantId, remoteFilename: filename, checksum })
|
|
return { created: true, checksum }
|
|
})
|
|
} catch (error) {
|
|
if (uploadedKey) {
|
|
try { await storage.remove(uploadedKey) }
|
|
catch (cleanupError) { console.error("Scanner-Upload konnte nicht bereinigt werden", { key: uploadedKey, cleanupError }) }
|
|
}
|
|
throw error
|
|
}
|
|
}
|
|
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), archiveDocument, fileDocument, probe, accept }
|
|
}
|
|
|
|
export type ScannerInboxService = ReturnType<typeof createScannerInboxService>
|