From 5181ecc9f3894b389aa26f9d1c9330d2288cef15 Mon Sep 17 00:00:00 2001 From: flfeders Date: Mon, 10 Aug 2026 15:16:42 +0200 Subject: [PATCH] Fix hanging document mailbox imports --- .../document-import.service.ts | 64 +++++++++++++++---- backend/src/routes/documentImports.ts | 6 +- .../pages/settings/document-imports/index.vue | 56 ++++++++++++---- 3 files changed, 100 insertions(+), 26 deletions(-) diff --git a/backend/src/modules/document-import/document-import.service.ts b/backend/src/modules/document-import/document-import.service.ts index 0311ab8..1eb06bd 100644 --- a/backend/src/modules/document-import/document-import.service.ts +++ b/backend/src/modules/document-import/document-import.service.ts @@ -33,6 +33,8 @@ type ImportSourceConnection = { } const activeSyncs = new Set() +const syncProgress = new Map() +const IMPORT_BATCH_SIZE = 25 const decryptString = (value: unknown) => value ? decrypt(value as any) : "" @@ -215,17 +217,18 @@ export function documentImportService(server: FastifyInstance) { 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 client: ImapFlow | null = null let imported = 0 let duplicates = 0 let messages = 0 try { + 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") + + client = createImapClient(source) await client.connect() const lock = await client.getMailboxLock(source.mailboxPath) try { @@ -237,11 +240,19 @@ export function documentImportService(server: FastifyInstance) { )).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 } + // Pro Lauf nur einen begrenzten Stapel laden. Der Worker verarbeitet verbliebene + // ungelesene Nachrichten in den folgenden Läufen weiter. + const unreadUids = await client.search({ seen: false }, { uid: true }) || [] + const batchUids = unreadUids.slice(0, IMPORT_BATCH_SIZE) + syncProgress.set(sourceId, { + processed: 0, + total: batchUids.length, + imported: 0, + duplicates: 0, + }) + const uidsToMarkSeen: number[] = [] - for await (const message of client.fetch(query, { uid: true, envelope: true, source: true })) { + for await (const message of client.fetch(batchUids, { uid: true, envelope: true, source: true }, { uid: true })) { messages += 1 const parsed = await simpleParser(message.source) const remoteMessageId = `${uidValidity}:${message.uid}` @@ -256,8 +267,19 @@ export function documentImportService(server: FastifyInstance) { complete = false } } - if (complete && source.markAsSeen) await client.messageFlagsAdd({ uid: message.uid }, ["\\Seen"], { uid: true }) + // Während eines laufenden fetch dürfen keine weiteren IMAP-Befehle ausgeführt + // werden, da ImapFlow sonst auf das Ende des eigenen fetch wartet. + if (complete && source.markAsSeen) uidsToMarkSeen.push(message.uid) if (complete) processedHighestUid = Math.max(processedHighestUid, Number(message.uid)) + syncProgress.set(sourceId, { + processed: messages, + total: batchUids.length, + imported, + duplicates, + }) + } + if (uidsToMarkSeen.length > 0) { + await client.messageFlagsAdd(uidsToMarkSeen, ["\\Seen"], { uid: true }) } await server.db.insert(documentImportStates).values({ @@ -277,14 +299,28 @@ export function documentImportService(server: FastifyInstance) { 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)) + await server.db.update(documentImportSources).set({ lastError: error?.message || "Synchronisierung fehlgeschlagen" }).where(eq(documentImportSources.id, sourceId)) throw error } finally { activeSyncs.delete(sourceId) - if (client.usable) await client.logout().catch(() => client.close()) + syncProgress.delete(sourceId) + if (client?.usable) await client.logout().catch(() => client.close()) } } + const startSync = (tenantId: number, sourceId: string) => { + if (activeSyncs.has(sourceId)) throw new Error("Diese Importquelle wird bereits synchronisiert") + void syncSource(tenantId, sourceId).catch((error: any) => { + server.log.error({ sourceId, error: error?.message }, "Manueller Dokumentenimport fehlgeschlagen") + }) + return { success: true, started: true } + } + + const getSyncStatus = (sourceId: string) => ({ + isSyncing: activeSyncs.has(sourceId), + progress: syncProgress.get(sourceId) || null, + }) + const syncAll = async () => { const sources = await server.db.select({ id: documentImportSources.id, @@ -303,5 +339,5 @@ export function documentImportService(server: FastifyInstance) { return results } - return { testConnection, syncSource, syncAll } + return { testConnection, syncSource, startSync, getSyncStatus, syncAll } } diff --git a/backend/src/routes/documentImports.ts b/backend/src/routes/documentImports.ts index 4083a1e..4a92ae2 100644 --- a/backend/src/routes/documentImports.ts +++ b/backend/src/routes/documentImports.ts @@ -29,6 +29,7 @@ export default async function documentImportRoutes(server: FastifyInstance) { lastSyncedAt: row.lastSyncedAt, lastError: row.lastError, createdAt: row.createdAt, + ...server.services.documentImports.getSyncStatus(row.id), }) server.get("/document-imports", async (req) => { @@ -101,7 +102,10 @@ export default async function documentImportRoutes(server: FastifyInstance) { server.post("/document-imports/:id/sync", async (req, reply) => { try { - return await server.services.documentImports.syncSource(tenantId(req), (req.params as { id: string }).id) + return reply.code(202).send(server.services.documentImports.startSync( + tenantId(req), + (req.params as { id: string }).id, + )) } catch (error: any) { return reply.code(400).send({ error: error?.message || "Synchronisierung fehlgeschlagen" }) } diff --git a/frontend/pages/settings/document-imports/index.vue b/frontend/pages/settings/document-imports/index.vue index b60e270..e648063 100644 --- a/frontend/pages/settings/document-imports/index.vue +++ b/frontend/pages/settings/document-imports/index.vue @@ -13,6 +13,13 @@ type ImportSource = { hasPassword: boolean lastSyncedAt?: string | null lastError?: string | null + isSyncing: boolean + progress?: { + processed: number + total: number + imported: number + duplicates: number + } | null } const api = useNuxtApp().$api @@ -23,6 +30,7 @@ const saving = ref(false) const activeAction = ref(null) const editingId = ref(null) const showForm = ref(false) +let pollTimer: ReturnType | null = null const emptyForm = () => ({ name: "", @@ -38,23 +46,33 @@ const emptyForm = () => ({ }) const form = ref(emptyForm()) -const load = async () => { - loading.value = true +const load = async (silent = false) => { + loading.value = sources.value.length === 0 try { const result = await api("/api/document-imports") sources.value = Array.isArray(result) ? result : [] } catch (error: any) { - sources.value = [] - toast.add({ - title: "Importquellen konnten nicht geladen werden", - description: error?.data?.error || error?.message || "Bitte versuche es später erneut.", - color: "error", - }) + if (!silent) { + toast.add({ + title: "Importquellen konnten nicht geladen werden", + description: error?.data?.error || error?.message || "Bitte versuche es später erneut.", + color: "error", + }) + } } finally { loading.value = false } } +const pollWhileSyncing = () => { + if (pollTimer || !sources.value.some(source => source.isSyncing)) return + pollTimer = setTimeout(async () => { + pollTimer = null + await load(true) + pollWhileSyncing() + }, 2000) +} + const create = () => { editingId.value = null form.value = emptyForm() @@ -100,6 +118,13 @@ const runAction = async (id: string, action: "test" | "sync") => { activeAction.value = `${id}:${action}` try { const result = await api(`/api/document-imports/${id}/${action}`, { method: "POST" }) + if (action === "sync" && result.started) { + const source = sources.value.find(item => item.id === id) + if (source) source.isSyncing = true + toast.add({ title: "Import gestartet", description: "Die Nachrichten werden im Hintergrund verarbeitet.", color: "success" }) + pollWhileSyncing() + return + } toast.add({ title: action === "test" ? "Verbindung erfolgreich" : "Import abgeschlossen", description: action === "test" @@ -120,7 +145,13 @@ const runAction = async (id: string, action: "test" | "sync") => { } } -onMounted(load) +onMounted(async () => { + await load() + pollWhileSyncing() +}) +onBeforeUnmount(() => { + if (pollTimer) clearTimeout(pollTimer) +})