MailSend/src/maildir-sync.js

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;
}