echos-ocr/src/lib/email-watcher.ts

import { ImapFlow } from "imapflow";
import { simpleParser } from "mailparser";
import { prisma } from "./db";
import { uploadBuffer } from "./storage";
import { enqueueProcessing } from "./processing-queue";

const ALLOWED_CONTENT_TYPES = [
  "application/pdf",
  "image/jpeg",
  "image/jpg",
  "image/png",
  "image/webp",
];

const MAX_ATTACHMENT_SIZE = 50 * 1024 * 1024; // 50 MB

type EmailConfig = {
  host: string;
  port: number;
  user: string;
  pass: string;
  tls: boolean;
  folder: string;
  processedAction: string;
  processedFolder: string;
};

function log(msg: string, ...args: unknown[]) {
  console.log(`[email-watcher] ${msg}`, ...args);
}

function logError(msg: string, ...args: unknown[]) {
  console.error(`[email-watcher] ${msg}`, ...args);
}

async function handleMessage(client: ImapFlow, uid: number, config: EmailConfig) {
  const result = await client.fetchOne(String(uid), { source: true }, { uid: true });
  if (!result) return;
  const rawSource = (result as unknown as { source?: Buffer }).source;
  if (!rawSource) return;

  const parsed = await simpleParser(rawSource);
  const subject = parsed.subject || "(no subject)";
  const from = parsed.from?.text || "unknown";

  const attachments = (parsed.attachments || []).filter((att) => {
    const ct = att.contentType?.toLowerCase() || "";
    return ALLOWED_CONTENT_TYPES.includes(ct);
  });

  if (attachments.length === 0) {
    log(`Skipping email from ${from} "${subject}" — no valid attachments`);
    return;
  }

  log(`Processing email from ${from} "${subject}" — ${attachments.length} attachment(s)`);

  for (const att of attachments) {
    const fileName = att.filename || `email-attachment-${Date.now()}`;
    const buffer = att.content;

    if (buffer.length > MAX_ATTACHMENT_SIZE) {
      log(`Skipping attachment "${fileName}" — exceeds 50 MB size limit (${(buffer.length / 1024 / 1024).toFixed(1)} MB)`);
      continue;
    }

    try {
      const job = await prisma.processingJob.create({
        data: {
          fileName,
          filePath: `email/${fileName}`,
          status: "queued",
        },
      });

      const sourceKey = `sources/${job.id}/${fileName}`;
      await uploadBuffer(sourceKey, buffer, att.contentType || "application/octet-stream");

      await enqueueProcessing(job.id);
      log(`Queued: ${fileName} (job ${job.id})`);
    } catch (err) {
      logError(`Failed to create job for ${fileName}:`, err);
    }
  }

  if (config.processedAction === "mark_read") {
    await client.messageFlagsAdd(String(uid), ["\\Seen"], { uid: true });
  } else if (config.processedAction === "move") {
    try {
      await client.messageMove(String(uid), config.processedFolder, { uid: true });
    } catch {
      log(`Could not move to "${config.processedFolder}", marking as read instead`);
      await client.messageFlagsAdd(String(uid), ["\\Seen"], { uid: true });
    }
  } else if (config.processedAction === "delete") {
    await client.messageDelete(String(uid), { uid: true });
  }
}

export async function scanInbox(): Promise<{ processed: number; skipped: number; error?: string }> {
  const settings = await prisma.appSettings.findUnique({ where: { id: "singleton" } });
  if (!settings?.emailImapHost || !settings?.emailImapUser || !settings?.emailImapPass) {
    return { processed: 0, skipped: 0, error: "IMAP not configured" };
  }

  if (!settings.emailWatching) {
    return { processed: 0, skipped: 0, error: "Email watching is disabled" };
  }

  const config: EmailConfig = {
    host: settings.emailImapHost,
    port: settings.emailImapPort,
    user: settings.emailImapUser,
    pass: settings.emailImapPass,
    tls: settings.emailImapTls,
    folder: settings.emailFolder,
    processedAction: settings.emailProcessed,
    processedFolder: settings.emailProcessedFolder,
  };

  const scanClient = new ImapFlow({
    host: config.host,
    port: config.port,
    secure: config.tls,
    auth: { user: config.user, pass: config.pass },
    logger: false,
    emitLogs: false,
  });

  let processed = 0;
  let skipped = 0;

  try {
    await scanClient.connect();
    log(`[scan] Connected for inbox scan`);

    const lock = await scanClient.getMailboxLock(config.folder);
    try {
      const searchResult = await scanClient.search({ seen: false }, { uid: true });
      const unseen = Array.isArray(searchResult) ? searchResult : [];
      log(`[scan] Found ${unseen.length} unseen message(s)`);

      for (const uid of unseen) {
        try {
          await handleMessage(scanClient, uid, config);
          processed++;
        } catch (err) {
          logError(`[scan] Error handling message UID ${uid}:`, err);
          skipped++;
        }
      }
    } finally {
      lock.release();
    }

    await scanClient.logout();
  } catch (err) {
    const message = err instanceof Error ? err.message : "Scan failed";
    logError("[scan] Error:", message);
    try { await scanClient.logout(); } catch {}
    return { processed, skipped, error: message };
  }

  log(`[scan] Complete: ${processed} processed, ${skipped} skipped`);
  return { processed, skipped };
}

export async function testEmailConnection(config: Omit<EmailConfig, "processedAction" | "processedFolder">): Promise<{ ok: boolean; folders?: string[]; error?: string }> {
  let testClient: ImapFlow | null = null;
  try {
    testClient = new ImapFlow({
      host: config.host,
      port: config.port,
      secure: config.tls,
      auth: { user: config.user, pass: config.pass },
      logger: false,
      emitLogs: false,
    });

    await testClient.connect();
    const folders: string[] = [];
    const list = await testClient.list();
    for (const folder of list) {
      folders.push(folder.path);
    }

    await testClient.logout();
    return { ok: true, folders };
  } catch (err) {
    const message = err instanceof Error ? err.message : "Connection failed";
    return { ok: false, error: message };
  } finally {
    if (testClient) {
      try { await testClient.logout(); } catch {}
    }
  }
}