import crypto from 'node:crypto'; import { unlink } from 'node:fs/promises'; import path from 'node:path'; import { clearInboundMaildirMigrationStaging, commitInboundMaildirMigrationStaging, createImportedInboundMessage, createImportedInboundMessageWithWebhook, listInboundFolders, listInboundMailboxesForStorage, listInboundMaildirIndex, listInboundMaildirMigrationCandidates, listInboundMaildirMigrationStaging, markMissingInboundMaildirMessages, stageInboundMessageMaildirStorageBatch, syncInboundMaildirFolders, syncInboundMessageMaildirMetadata } from './db.js'; import { parseInboundMessage } from './inbound-mail.js'; import { ensureMaildirMailbox, flushMaildirFiles, listMaildirFolders, maildirHomePath, readMaildirMessage, scanMaildirMailbox, sqliteMaildirStorageKey, writeMaildirMessage } from './maildir-store.js'; export async function migrateInboundMessagesToMaildir({ root, batchSize = 250, onProgress = null } = {}) { const cleanBatchSize = Math.min(1000, Math.max(1, Number(batchSize) || 250)); let lastId = 0; let processed = 0; let written = 0; let reused = 0; let looseFiles = []; let looseDirectories = []; try { await cleanupMaildirMigrationStaging({ root, batchSize: cleanBatchSize }); for (const mailbox of listInboundMailboxesForStorage()) { await ensureMaildirMailbox(root, mailbox.address, listInboundFolders(mailbox), { durable: true }); const entries = new Map( (await scanMaildirMailbox({ root, address: mailbox.address })) .map((entry) => [entry.storageKey, entry]) ); let mailboxAfterId = 0; while (true) { const candidates = listInboundMaildirMigrationCandidates( mailboxAfterId, cleanBatchSize, { mailboxId: mailbox.id } ); if (!candidates.length) break; const batchRecords = []; const batchFiles = []; const batchDirectories = []; looseFiles = batchFiles; looseDirectories = batchDirectories; for (const candidate of candidates) { mailboxAfterId = candidate.id; lastId = Math.max(lastId, candidate.id); const storageKey = validStorageKey(candidate.storageKey) ? candidate.storageKey : sqliteMaildirStorageKey(candidate.id); const existing = entries.get(storageKey); if (existing) { const { bytes } = await readMaildirMessage(existing); if (!bytes.equals(Buffer.from(candidate.rawMessageBytes))) { throw new Error(`邮件 ${candidate.id} 的 Maildir 存储标识已被其他内容占用。`); } batchDirectories.push(path.dirname(existing.filePath)); await unlink(existing.filePath); entries.delete(storageKey); reused += 1; } const storage = await writeMaildirMessage({ root, address: candidate.mailboxAddress, rawMessageBytes: candidate.rawMessageBytes, folder: candidate.folder, flags: candidate.flags, keywords: candidate.keywords, read: candidate.read, receivedAt: candidate.receivedAt, storageKey, durable: false }); const filePath = maildirFilePath(root, candidate.mailboxAddress, storage.relpath); entries.set(storageKey, { storageKey, filePath, relpath: storage.relpath, size: storage.size, mtimeMs: storage.mtimeMs, folder: candidate.folder, flags: candidate.flags, keywords: candidate.keywords, read: candidate.read }); written += 1; batchFiles.push(filePath); batchRecords.push({ id: candidate.id, storage }); processed += 1; await onProgress?.({ processed, written, reused, lastId: candidate.id }); } stageInboundMessageMaildirStorageBatch(batchRecords); await flushMaildirFiles(batchFiles, { root, directories: batchDirectories }); looseFiles = []; looseDirectories = []; } } const switched = commitInboundMaildirMigrationStaging(processed); if (switched !== processed) throw new Error('Maildir 批量切换未完整提交。'); } catch (error) { await removeMaildirFiles(looseFiles); await flushMaildirFiles([], { root, directories: looseDirectories }).catch(() => null); await cleanupMaildirMigrationStaging({ root, batchSize: cleanBatchSize }).catch(() => null); throw error; } return { processed, written, reused, lastId }; } export async function reconcileMaildirMailbox({ root, mailbox, missingCounts = new Map() }) { await ensureMaildirMailbox(root, mailbox.address); const folderReport = syncInboundMaildirFolders( mailbox, await listMaildirFolders({ root, address: mailbox.address }) ); const index = listInboundMaildirIndex(mailbox.id); const entries = assignStorageKeys( await scanMaildirMailbox({ root, address: mailbox.address }), index ); const indexByKey = new Map(index.map((message) => [message.storageKey, message])); const present = new Set(); let created = 0; let updated = 0; for (const entry of entries) { present.add(entry.storageKey); missingCounts.delete(missingCounterKey(mailbox.id, entry.storageKey)); const existing = indexByKey.get(entry.storageKey); const storage = storageFromEntry(entry); if (existing) { if ( !existing.deleted && existing.storageRelpath === entry.relpath && existing.storageSize === entry.size && existing.storageMtimeMs === entry.mtimeMs && existing.folder === entry.folder && existing.read === entry.read && sameStringSet(existing.flags, entry.flags) && sameStringSet(existing.keywords, entry.keywords) ) continue; if (syncInboundMessageMaildirMetadata(mailbox, storage, entry)) updated += 1; continue; } const { bytes, sha256 } = await readMaildirMessage(entry); let parsed = {}; try { parsed = await parseInboundMessage(bytes, [mailbox.address]); } catch { // Dovecot must still expose malformed historic messages verbatim. The // management index falls back to minimal metadata instead of dropping it. } const createIndex = entry.folder === 'INBOX' && entry.storageKey.startsWith('mhsmtp-') ? createImportedInboundMessageWithWebhook : createImportedInboundMessage; const result = createIndex(mailbox, { ...parsed, importSource: 'maildir-index', sourceKey: `${mailbox.id}:${entry.storageKey}`, folder: entry.folder, flags: entry.flags, keywords: entry.keywords, read: entry.read, receivedAt: entry.receivedAt, rawMessageBytes: bytes, recipients: parsed.recipients?.length ? parsed.recipients : [mailbox.address], storage: storageFromEntry(entry, { sha256, size: bytes.length }) }); if (result.created) { created += 1; } else if (syncInboundMessageMaildirMetadata(mailbox, storage, entry)) updated += 1; } const protectedKeys = new Set(present); for (const message of index) { if (present.has(message.storageKey) || message.deleted) continue; const counterKey = missingCounterKey(mailbox.id, message.storageKey); const count = (missingCounts.get(counterKey) || 0) + 1; missingCounts.set(counterKey, count); if (count < 2) protectedKeys.add(message.storageKey); } const deleted = markMissingInboundMaildirMessages(mailbox.id, protectedKeys); return { mailboxId: mailbox.id, files: entries.length, foldersCreatedOrRestored: folderReport.createdOrRestored, foldersDeleted: folderReport.deleted, created, updated, deleted }; } export async function reconcileAllMaildirs({ root, missingCounts = new Map() } = {}) { const reports = []; for (const mailbox of listInboundMailboxesForStorage()) { reports.push(await reconcileMaildirMailbox({ root, mailbox, missingCounts })); } return reports; } export function startMaildirReconciler({ root, enabled = true, intervalMs = 300_000, logger = console } = {}) { if (!enabled) return { stop() {} }; const delay = Math.max(1_000, Number(intervalMs) || 300_000); const missingCounts = new Map(); let stopped = false; let timer = null; const schedule = () => { if (stopped) return; timer = setTimeout(run, delay); timer.unref?.(); }; const run = async () => { try { await reconcileAllMaildirs({ root, missingCounts }); } catch (error) { logger.warn?.(`Maildir index synchronization failed: ${error.message || error}`); } finally { schedule(); } }; schedule(); return { async runNow() { return reconcileAllMaildirs({ root, missingCounts }); }, stop() { stopped = true; clearTimeout(timer); } }; } function assignStorageKeys(entries, index) { const indexByRelpath = new Map(index.map((message) => [message.storageRelpath, message])); const used = new Set(); const pending = []; const output = []; for (const entry of entries) { const existing = indexByRelpath.get(entry.relpath); if (existing && !used.has(existing.storageKey)) { used.add(existing.storageKey); output.push({ ...entry, storageKey: existing.storageKey }); } else { pending.push(entry); } } for (const entry of pending) { let storageKey = entry.storageKey; if (used.has(storageKey)) { storageKey = `mhcopy-${crypto.createHash('sha256') .update(`${entry.storageKey}\0${entry.relpath}`) .digest('hex') .slice(0, 40)}`; } while (used.has(storageKey)) { storageKey = `mhcopy-${crypto.createHash('sha256') .update(`${storageKey}\0${entry.relpath}`) .digest('hex') .slice(0, 40)}`; } used.add(storageKey); output.push({ ...entry, storageKey }); } return output.sort((left, right) => left.relpath.localeCompare(right.relpath)); } function storageFromEntry(entry, overrides = {}) { return { backend: 'maildir', key: entry.storageKey, relpath: entry.relpath, sha256: overrides.sha256 || '', size: overrides.size ?? entry.size, mtimeMs: entry.mtimeMs, indexedAt: new Date().toISOString() }; } function validStorageKey(value) { return /^[a-z0-9][a-z0-9._-]{0,191}$/.test(String(value || '')); } function sameStringSet(left, right) { return [...new Set(left || [])].sort().join('\0') === [...new Set(right || [])].sort().join('\0'); } function missingCounterKey(mailboxId, storageKey) { return `${mailboxId}:${storageKey}`; } async function cleanupMaildirMigrationStaging({ root, batchSize }) { let afterMessageId = 0; while (true) { const staged = listInboundMaildirMigrationStaging(afterMessageId, batchSize); if (!staged.length) break; const files = staged.map((entry) => ( maildirFilePath(root, entry.mailboxAddress, entry.storageRelpath) )); await removeMaildirFiles(files); await flushMaildirFiles([], { root, directories: files.map((filePath) => path.dirname(filePath)) }); afterMessageId = staged.at(-1).messageId; } clearInboundMaildirMigrationStaging(); } async function removeMaildirFiles(filePaths) { const files = [...new Set(filePaths || [])]; for (let offset = 0; offset < files.length; offset += 32) { await Promise.all(files.slice(offset, offset + 32).map(async (filePath) => { try { await unlink(filePath); } catch (error) { if (error?.code !== 'ENOENT') throw error; } })); } } function maildirFilePath(root, address, relpath) { const home = maildirHomePath(root, address); const filePath = path.resolve(home, ...String(relpath || '').split('/')); if (!filePath.startsWith(`${home}${path.sep}`)) throw new Error('Maildir 存储路径不正确。'); return filePath; }