flfeders/FEDEO
2
1
Fork 0
Code Issues 106 Pull Requests Actions Packages Projects 22 Releases Wiki Activity

Fix hanging document mailbox imports

Browse Source
This commit is contained in:
Florian Federspiel
2026-08-10 15:16:42 +02:00
parent 49cbc9314e
commit 5181ecc9f3
3 changed files with 100 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

64
backend/src/modules/document-import/document-import.service.ts
View File

@@ -33,6 +33,8 @@ type ImportSourceConnection = {
}
const activeSyncs = new Set<string>()
const syncProgress = new Map<string, { processed: number, total: number, imported: number, duplicates: number }>()
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 }
}