308 lines
13 KiB
TypeScript
308 lines
13 KiB
TypeScript
import { createHash } from "node:crypto"
|
|
import dayjs from "dayjs"
|
|
import { and, eq } from "drizzle-orm"
|
|
import { FastifyInstance } from "fastify"
|
|
import { ImapFlow } from "imapflow"
|
|
import { simpleParser } from "mailparser"
|
|
|
|
import {
|
|
documentImportItems,
|
|
documentImportSources,
|
|
documentImportStates,
|
|
filetags,
|
|
folders,
|
|
} from "../../../db/schema"
|
|
import { decrypt } from "../../utils/crypt"
|
|
import { saveFile } from "../../utils/files"
|
|
|
|
type ImportSourceConnection = {
|
|
id: string
|
|
tenantId: number
|
|
name: string
|
|
provider: string
|
|
enabled: boolean
|
|
mailboxAddress: string
|
|
password: string
|
|
imapHost: string
|
|
imapPort: number
|
|
imapSecure: boolean
|
|
mailboxPath: string
|
|
targetFolderId: string | null
|
|
defaultFiletypeId: string | null
|
|
markAsSeen: boolean
|
|
}
|
|
|
|
const activeSyncs = new Set<string>()
|
|
|
|
const decryptString = (value: unknown) => value ? decrypt(value as any) : ""
|
|
|
|
export function documentImportService(server: FastifyInstance) {
|
|
const loadSource = async (tenantId: number, sourceId: string): Promise<ImportSourceConnection | null> => {
|
|
const [row] = await server.db
|
|
.select()
|
|
.from(documentImportSources)
|
|
.where(and(
|
|
eq(documentImportSources.id, sourceId),
|
|
eq(documentImportSources.tenantId, tenantId),
|
|
))
|
|
.limit(1)
|
|
|
|
if (!row) return null
|
|
return {
|
|
id: row.id,
|
|
tenantId: row.tenantId,
|
|
name: row.name,
|
|
provider: row.provider,
|
|
enabled: row.enabled,
|
|
mailboxAddress: decryptString(row.mailboxAddressEncrypted),
|
|
password: decryptString(row.passwordEncrypted),
|
|
imapHost: decryptString(row.imapHostEncrypted),
|
|
imapPort: row.imapPort,
|
|
imapSecure: row.imapSecure,
|
|
mailboxPath: row.mailboxPath,
|
|
targetFolderId: row.targetFolderId,
|
|
defaultFiletypeId: row.defaultFiletypeId,
|
|
markAsSeen: row.markAsSeen,
|
|
}
|
|
}
|
|
|
|
const createImapClient = (source: ImportSourceConnection) => new ImapFlow({
|
|
host: source.imapHost,
|
|
port: source.imapPort,
|
|
secure: source.imapSecure,
|
|
auth: { user: source.mailboxAddress, pass: source.password },
|
|
logger: false,
|
|
})
|
|
|
|
const classify = async (source: ImportSourceConnection, subject = "") => {
|
|
let folderId = source.targetFolderId
|
|
let filetypeId = source.defaultFiletypeId
|
|
|
|
if (!folderId && /(Rechnung|Beleg|Invoice|Quittung)/i.test(subject)) {
|
|
const [folder] = await server.db.select({ id: folders.id }).from(folders).where(and(
|
|
eq(folders.tenant, source.tenantId),
|
|
eq(folders.function, "incomingInvoices"),
|
|
// @ts-ignore Das bestehende Schema typisiert das Jahr numerisch, verwendet es aber als Zeichenfolge.
|
|
eq(folders.year, dayjs().format("YYYY")),
|
|
)).limit(1)
|
|
folderId = folder?.id || null
|
|
|
|
if (!filetypeId) {
|
|
const [tag] = await server.db.select({ id: filetags.id }).from(filetags).where(and(
|
|
eq(filetags.tenant, source.tenantId),
|
|
eq(filetags.incomingDocumentType, "invoices"),
|
|
)).limit(1)
|
|
filetypeId = tag?.id || null
|
|
}
|
|
} else if (!filetypeId && /(Mahnung|Zahlungsaufforderung|Zahlungsverzug)/i.test(subject)) {
|
|
const [tag] = await server.db.select({ id: filetags.id }).from(filetags).where(and(
|
|
eq(filetags.tenant, source.tenantId),
|
|
eq(filetags.incomingDocumentType, "reminders"),
|
|
)).limit(1)
|
|
filetypeId = tag?.id || null
|
|
}
|
|
|
|
if (!folderId) {
|
|
const [folder] = await server.db.select({ id: folders.id }).from(folders).where(and(
|
|
eq(folders.tenant, source.tenantId),
|
|
eq(folders.function, "deposit"),
|
|
)).limit(1)
|
|
folderId = folder?.id || null
|
|
}
|
|
|
|
return { folderId, filetypeId }
|
|
}
|
|
|
|
const importAttachment = async (
|
|
source: ImportSourceConnection,
|
|
remoteMessageId: string,
|
|
subject: string,
|
|
attachment: any,
|
|
index: number,
|
|
) => {
|
|
const content = Buffer.from(attachment.content)
|
|
const checksum = createHash("sha256").update(content).digest("hex")
|
|
const attachmentKey = `${index}:${attachment.filename || "Anhang"}`
|
|
|
|
const [existing] = await server.db.select({ status: documentImportItems.status }).from(documentImportItems)
|
|
.where(and(
|
|
eq(documentImportItems.sourceId, source.id),
|
|
eq(documentImportItems.remoteMessageId, remoteMessageId),
|
|
eq(documentImportItems.attachmentKey, attachmentKey),
|
|
)).limit(1)
|
|
if (existing?.status === "imported" || existing?.status === "duplicate") return "duplicate"
|
|
|
|
const [sameContent] = await server.db.select({ id: documentImportItems.id }).from(documentImportItems)
|
|
.where(and(
|
|
eq(documentImportItems.sourceId, source.id),
|
|
eq(documentImportItems.attachmentChecksum, checksum),
|
|
eq(documentImportItems.status, "imported"),
|
|
)).limit(1)
|
|
|
|
if (sameContent) {
|
|
await server.db.insert(documentImportItems).values({
|
|
tenantId: source.tenantId,
|
|
sourceId: source.id,
|
|
remoteMessageId,
|
|
attachmentKey,
|
|
attachmentChecksum: checksum,
|
|
filename: attachment.filename || null,
|
|
status: "duplicate",
|
|
}).onConflictDoUpdate({
|
|
target: [documentImportItems.sourceId, documentImportItems.remoteMessageId, documentImportItems.attachmentKey],
|
|
set: { status: "duplicate", error: null },
|
|
})
|
|
return "duplicate"
|
|
}
|
|
|
|
try {
|
|
const target = await classify(source, subject)
|
|
const saved = await saveFile(
|
|
server,
|
|
source.tenantId,
|
|
remoteMessageId,
|
|
attachment,
|
|
target.folderId,
|
|
target.filetypeId,
|
|
)
|
|
if (!saved) throw new Error("Datei konnte nicht gespeichert werden")
|
|
|
|
await server.db.insert(documentImportItems).values({
|
|
tenantId: source.tenantId,
|
|
sourceId: source.id,
|
|
remoteMessageId,
|
|
attachmentKey,
|
|
attachmentChecksum: checksum,
|
|
filename: attachment.filename || null,
|
|
status: "imported",
|
|
fileId: saved.id,
|
|
}).onConflictDoUpdate({
|
|
target: [documentImportItems.sourceId, documentImportItems.remoteMessageId, documentImportItems.attachmentKey],
|
|
set: { status: "imported", error: null, fileId: saved.id },
|
|
})
|
|
return "imported"
|
|
} catch (error: any) {
|
|
await server.db.insert(documentImportItems).values({
|
|
tenantId: source.tenantId,
|
|
sourceId: source.id,
|
|
remoteMessageId,
|
|
attachmentKey,
|
|
attachmentChecksum: checksum,
|
|
filename: attachment.filename || null,
|
|
status: "failed",
|
|
error: error?.message || "Import fehlgeschlagen",
|
|
}).onConflictDoUpdate({
|
|
target: [documentImportItems.sourceId, documentImportItems.remoteMessageId, documentImportItems.attachmentKey],
|
|
set: { status: "failed", error: error?.message || "Import fehlgeschlagen" },
|
|
})
|
|
throw error
|
|
}
|
|
}
|
|
|
|
const testConnection = async (tenantId: number, sourceId: string) => {
|
|
const source = await loadSource(tenantId, sourceId)
|
|
if (!source) throw new Error("Importquelle wurde nicht gefunden")
|
|
if (source.provider !== "imap") throw new Error("Dieser Provider wird noch nicht unterstützt")
|
|
const client = createImapClient(source)
|
|
try {
|
|
await client.connect()
|
|
const mailbox = await client.mailboxOpen(source.mailboxPath, { readOnly: true })
|
|
return { success: true, mailbox: source.mailboxPath, messages: mailbox.exists }
|
|
} finally {
|
|
await client.logout().catch(() => client.close())
|
|
}
|
|
}
|
|
|
|
const syncSource = async (tenantId: number, sourceId: string) => {
|
|
if (activeSyncs.has(sourceId)) throw new Error("Diese Importquelle wird bereits synchronisiert")
|
|
const source = await loadSource(tenantId, sourceId)
|
|
if (!source) throw new Error("Importquelle wurde nicht gefunden")
|
|
if (!source.enabled) throw new Error("Importquelle ist deaktiviert")
|
|
if (source.provider !== "imap") throw new Error("Dieser Provider wird noch nicht unterstützt")
|
|
|
|
activeSyncs.add(sourceId)
|
|
const client = createImapClient(source)
|
|
let imported = 0
|
|
let duplicates = 0
|
|
let messages = 0
|
|
try {
|
|
await client.connect()
|
|
const lock = await client.getMailboxLock(source.mailboxPath)
|
|
try {
|
|
const opened: any = await client.mailboxOpen(source.mailboxPath)
|
|
const uidValidity = Number(opened.uidValidity || 0)
|
|
const [state] = await server.db.select().from(documentImportStates).where(and(
|
|
eq(documentImportStates.sourceId, source.id),
|
|
eq(documentImportStates.mailboxPath, source.mailboxPath),
|
|
)).limit(1)
|
|
const highestUid = state && Number(state.uidValidity) === uidValidity ? Number(state.highestUid) : 0
|
|
let processedHighestUid = highestUid
|
|
// Ungelesene Nachrichten werden immer berücksichtigt. So bleibt ein fehlgeschlagener
|
|
// Import erneut verarbeitbar, auch wenn danach bereits neuere UIDs erfolgreich waren.
|
|
const query: any = { seen: false }
|
|
|
|
for await (const message of client.fetch(query, { uid: true, envelope: true, source: true })) {
|
|
messages += 1
|
|
const parsed = await simpleParser(message.source)
|
|
const remoteMessageId = `${uidValidity}:${message.uid}`
|
|
let complete = true
|
|
for (const [index, attachment] of (parsed.attachments || []).entries()) {
|
|
if (attachment.contentDisposition === "inline" && !attachment.filename) continue
|
|
try {
|
|
const result = await importAttachment(source, remoteMessageId, parsed.subject || "", attachment, index)
|
|
if (result === "imported") imported += 1
|
|
else duplicates += 1
|
|
} catch {
|
|
complete = false
|
|
}
|
|
}
|
|
if (complete && source.markAsSeen) await client.messageFlagsAdd({ uid: message.uid }, ["\\Seen"], { uid: true })
|
|
if (complete) processedHighestUid = Math.max(processedHighestUid, Number(message.uid))
|
|
}
|
|
|
|
await server.db.insert(documentImportStates).values({
|
|
sourceId: source.id,
|
|
mailboxPath: source.mailboxPath,
|
|
uidValidity,
|
|
highestUid: processedHighestUid,
|
|
updatedAt: new Date(),
|
|
}).onConflictDoUpdate({
|
|
target: [documentImportStates.sourceId, documentImportStates.mailboxPath],
|
|
set: { uidValidity, highestUid: processedHighestUid, updatedAt: new Date() },
|
|
})
|
|
} finally {
|
|
lock.release()
|
|
}
|
|
|
|
await server.db.update(documentImportSources).set({ lastSyncedAt: new Date(), lastError: null }).where(eq(documentImportSources.id, source.id))
|
|
return { success: true, messages, imported, duplicates }
|
|
} catch (error: any) {
|
|
await server.db.update(documentImportSources).set({ lastError: error?.message || "Synchronisierung fehlgeschlagen" }).where(eq(documentImportSources.id, source.id))
|
|
throw error
|
|
} finally {
|
|
activeSyncs.delete(sourceId)
|
|
if (client.usable) await client.logout().catch(() => client.close())
|
|
}
|
|
}
|
|
|
|
const syncAll = async () => {
|
|
const sources = await server.db.select({
|
|
id: documentImportSources.id,
|
|
tenantId: documentImportSources.tenantId,
|
|
}).from(documentImportSources).where(eq(documentImportSources.enabled, true))
|
|
|
|
const results = []
|
|
for (const source of sources) {
|
|
try {
|
|
results.push({ sourceId: source.id, ...(await syncSource(source.tenantId, source.id)) })
|
|
} catch (error: any) {
|
|
server.log.error({ sourceId: source.id, error: error?.message }, "Dokumentenimport fehlgeschlagen")
|
|
results.push({ sourceId: source.id, success: false, error: error?.message || "Import fehlgeschlagen" })
|
|
}
|
|
}
|
|
return results
|
|
}
|
|
|
|
return { testConnection, syncSource, syncAll }
|
|
}
|