rebreak-monorepo/backend/imap-idle/index.mjs

/**
 * rebreak-imap-idle — IMAP IDLE Daemon
 *
 * Hält pro aktivem MailConnection-Eintrag eine persistente IMAP-IDLE-Session.
 * Wenn der Server "EXISTS" meldet (neue Mail), feuert der Daemon sofort
 * POST /api/mail/scan-internal gegen das lokale Backend — ohne 30min-Warte.
 *
 * Env-Vars (via Infisical-Wrapper):
 *   DATABASE_URL         — Postgres-Connection-String
 *   ADMIN_SECRET         — Header-Secret für /api/mail/scan-internal
 *   ENCRYPTION_KEY       — AES-256 Key (gleicher wie im Backend)
 *   BACKEND_URL          — z.B. http://127.0.0.1:3016 (default: 3016)
 *   NODE_ENV             — production / staging
 *
 * Starten:
 *   node index.mjs
 *
 * Prozess-Signale:
 *   SIGTERM / SIGINT → graceful shutdown (alle IMAP-Sessions schließen)
 */

import { ImapFlow } from "imapflow";
import pg from "pg";
import { createDecipheriv } from "crypto";

// ─── Config ─────────────────────────────────────────────────────────────────

const BACKEND_URL =
  process.env.BACKEND_URL ||
  (process.env.NODE_ENV === "production"
    ? "http://127.0.0.1:3015"
    : "http://127.0.0.1:3016");

const ADMIN_SECRET =
  process.env.NUXT_ADMIN_SECRET || process.env.ADMIN_SECRET || "";

const DB_REFRESH_INTERVAL_MS = 5 * 60 * 1000;   // 5 min — neue Connections entdecken
// IDLE_RENEW von 25min → 10min: GMX dropped IDLE-connections silent vor 25min
// → exists-events kommen nie an + ImapFlow.idle() hängt ohne reject. 10min
// deckt alle bekannten Provider-Timeouts ab (GMX ~10-15min, Gmail ~29min,
// iCloud ~29min, Outlook ~29min). Trade-off: alle 10min full reconnect-cycle.
const IDLE_RENEW_INTERVAL_MS = 10 * 60 * 1000;  // 10 min (war 25)
// NOOP-heartbeat alle 2min während IDLE: defensive check ob connection wirklich
// alive ist. Wenn NOOP fehlschlägt → close + reconnect-loop.
const IDLE_NOOP_INTERVAL_MS  = 2 * 60 * 1000;   // 2 min — silent-drop early-detection
const RECONNECT_DELAYS_MS    = [1000, 5000, 30_000]; // exponential backoff, danach 60s loop
const RECONNECT_LOOP_DELAY_MS = 60 * 1000;

// ─── DB-Pool (direktes pg, kein Prisma — Daemon ist kein Nitro-Kontext) ────

const pool = new pg.Pool({ connectionString: process.env.DATABASE_URL });

async function loadActiveConnections() {
  // DB-table heißt "mail_connections" + snake_case columns (Prisma @map).
  // Aliase auf camelCase damit der restliche Daemon-Code unverändert bleibt.
  const { rows } = await pool.query(
    `SELECT id,
            user_id           AS "userId",
            email,
            imap_host         AS "imapHost",
            imap_port         AS "imapPort",
            password_encrypted AS "passwordEncrypted",
            reject_unauthorized AS "rejectUnauthorized",
            use_starttls      AS "useStarttls"
     FROM rebreak.mail_connections
     WHERE is_active = true`,
  );
  return rows;
}

// ─── Crypto (analog zu server/utils/crypto.ts) ──────────────────────────────

const AES_ALGO   = "aes-256-gcm";
const KEY_LENGTH = 32;

function getKey() {
  const raw =
    process.env.NUXT_ENCRYPTION_KEY || process.env.ENCRYPTION_KEY || "";
  if (!raw || raw.length < KEY_LENGTH) {
    return Buffer.alloc(
      KEY_LENGTH,
      raw.padEnd(KEY_LENGTH, "0").slice(0, KEY_LENGTH),
    );
  }
  return Buffer.from(raw.slice(0, KEY_LENGTH), "utf8");
}

function decrypt(stored) {
  const parts = stored.split(":");
  if (parts.length !== 3) throw new Error("Invalid encrypted format");
  const [ivHex, tagHex, dataHex] = parts;
  const decipher = createDecipheriv(
    AES_ALGO,
    getKey(),
    Buffer.from(ivHex, "hex"),
  );
  decipher.setAuthTag(Buffer.from(tagHex, "hex"));
  return decipher.update(Buffer.from(dataHex, "hex")) + decipher.final("utf8");
}

// ─── Logging ────────────────────────────────────────────────────────────────

function log(email, msg) {
  // NIEMALS password/credentials loggen. email ist safe (kein secret).
  console.log(`[idle/${email}] ${msg}`);
}

function logError(email, msg, err) {
  const errMsg = err?.message ?? String(err);
  // Credentials tauchen nie in err.message auf (ImapFlow maskiert sie nicht,
  // aber wir liefern pass nur an ImapFlow — nicht in eigenen log-calls).
  console.error(`[idle/${email}] ${msg}: ${errMsg}`);
}

// ─── Session-Registry ────────────────────────────────────────────────────────
// Map<connectionId, SessionHandle>

const sessions = new Map();

let shuttingDown = false;

// ─── Scan-Trigger ────────────────────────────────────────────────────────────

async function triggerScan(conn) {
  try {
    const res = await fetch(`${BACKEND_URL}/api/mail/scan-internal`, {
      method: "POST",
      headers: {
        "Content-Type": "application/json",
        "x-admin-secret": ADMIN_SECRET,
      },
      body: JSON.stringify({ userId: conn.userId }),
    });
    if (!res.ok) {
      log(conn.email, `scan-trigger HTTP ${res.status}`);
    } else {
      const data = await res.json().catch(() => ({}));
      log(
        conn.email,
        `scan-triggered → scanned=${data.scanned ?? "?"} blocked=${data.blocked ?? "?"}`,
      );
    }
  } catch (err) {
    logError(conn.email, "scan-trigger failed", err);
  }
}

// ─── IDLE-Session ────────────────────────────────────────────────────────────

/**
 * Startet eine einzelne IDLE-Session für eine MailConnection.
 * Reconnect-Loop läuft intern — diese Funktion returned nie (bis shutdown).
 */
async function runSession(conn) {
  let attempt = 0;

  while (!shuttingDown) {
    let password;
    try {
      password = decrypt(conn.passwordEncrypted);
    } catch (err) {
      logError(conn.email, "decrypt failed — session aborted", err);
      return; // Kein retry bei Decrypt-Fehler (kaputtes Credential)
    }

    const useImplicitTls = !conn.useStarttls;
    const imap = new ImapFlow({
      host: conn.imapHost,
      port: conn.imapPort,
      secure: useImplicitTls,
      ...(conn.useStarttls ? { requireTLS: true } : {}),
      auth: { user: conn.email, pass: password },
      logger: false,
      tls: { rejectUnauthorized: conn.rejectUnauthorized ?? true },
      // Outlook schließt idle-connections aggressiv — disableCompression
      // verhindert edge-cases bei partial reads nach reconnect
      disableCompression: conn.imapHost.includes("office365"),
    });

    // Referenz ablegen damit shutdown() darauf zugreifen kann
    const handle = sessions.get(conn.id);
    if (handle) handle.imap = imap;

    try {
      await imap.connect();
      log(conn.email, `connected (${conn.imapHost}:${conn.imapPort})`);
      attempt = 0; // Reset nach erfolgreicher Verbindung

      await imap.getMailboxLock("INBOX");

      // IDLE-Loop: alle IDLE_RENEW_INTERVAL_MS erneuern
      while (!shuttingDown && sessions.has(conn.id)) {
        let idleAbort;

        const idlePromise = new Promise((resolve, reject) => {
          imap
            .idle()
            .then(resolve)
            .catch(reject);

          idleAbort = () => {
            // ImapFlow.idle() bricht ab wenn die Connection getrennt wird
            imap.close();
            resolve();
          };
        });

        // exists-event → sofort scannen
        const onExists = () => {
          log(conn.email, "exists-event received (new mail)");
          triggerScan(conn); // fire-and-forget
        };
        imap.on("exists", onExists);

        // IDLE nach 10min erneuern (war 25; GMX dropped silent vor 25min)
        const renewTimer = setTimeout(() => {
          log(conn.email, "idle renewing (10min threshold)");
          imap.close(); // Unterbricht idle() → Loop iteriert → reconnect
        }, IDLE_RENEW_INTERVAL_MS);

        // NOOP-heartbeat alle 2min: detect silent-IDLE-drops (GMX-pattern).
        // Wenn NOOP fehlschlägt → close → loop iteriert → reconnect.
        const noopTimer = setInterval(async () => {
          try {
            await imap.noop();
            // Optional: verbose-log für debugging — aktuell silent
            // log(conn.email, "noop ok");
          } catch (err) {
            logError(conn.email, "noop failed — connection dead, force reconnect", err);
            imap.close();
          }
        }, IDLE_NOOP_INTERVAL_MS);

        try {
          await idlePromise;
        } finally {
          clearInterval(noopTimer);
          clearTimeout(renewTimer);
          imap.removeListener("exists", onExists);
        }

        if (shuttingDown || !sessions.has(conn.id)) break;

        // Kurze Pause vor Reconnect (idle() returned auch bei normalem timeout)
        await sleep(500);
      }

      try { await imap.logout(); } catch { /* ignore */ }
      return; // Sauberer Exit (shutdown oder Connection entfernt)

    } catch (err) {
      logError(conn.email, "connection error", err);
      try { imap.close(); } catch { /* ignore */ }
    }

    if (shuttingDown || !sessions.has(conn.id)) return;

    // Exponential backoff
    const delay =
      attempt < RECONNECT_DELAYS_MS.length
        ? RECONNECT_DELAYS_MS[attempt]
        : RECONNECT_LOOP_DELAY_MS;
    attempt++;
    log(conn.email, `reconnecting in ${delay / 1000}s (attempt ${attempt})`);
    await sleep(delay);
  }
}

// ─── Session-Management ──────────────────────────────────────────────────────

function startSession(conn) {
  if (sessions.has(conn.id)) return; // bereits aktiv
  log(conn.email, "session starting");
  const handle = { conn, imap: null, promise: null };
  sessions.set(conn.id, handle);
  handle.promise = runSession(conn).catch((err) => {
    logError(conn.email, "session crashed (unhandled)", err);
    sessions.delete(conn.id);
  });
}

async function stopSession(connectionId, email) {
  const handle = sessions.get(connectionId);
  if (!handle) return;
  log(email ?? connectionId, "session stopping");
  sessions.delete(connectionId);
  if (handle.imap) {
    try { handle.imap.close(); } catch { /* ignore */ }
  }
  await handle.promise.catch(() => {});
}

// ─── DB-Refresh-Loop ─────────────────────────────────────────────────────────

async function refreshConnections() {
  let rows;
  try {
    rows = await loadActiveConnections();
  } catch (err) {
    console.error("[idle/db] loadActiveConnections failed:", err.message);
    return;
  }

  const activeIds = new Set(rows.map((r) => r.id));

  // Neue Connections starten
  for (const row of rows) {
    if (!sessions.has(row.id)) {
      startSession(row);
    }
  }

  // Entfernte Connections stoppen
  for (const [id, handle] of sessions.entries()) {
    if (!activeIds.has(id)) {
      await stopSession(id, handle.conn?.email);
    }
  }

  console.log(
    `[idle/db] refreshed — ${activeIds.size} active connections, ${sessions.size} sessions`,
  );
}

// ─── Graceful Shutdown ───────────────────────────────────────────────────────

async function shutdown(signal) {
  console.log(`[idle] received ${signal} — shutting down gracefully`);
  shuttingDown = true;

  const stopPromises = [];
  for (const [id, handle] of sessions.entries()) {
    stopPromises.push(stopSession(id, handle.conn?.email));
  }
  await Promise.allSettled(stopPromises);

  await pool.end().catch(() => {});
  console.log("[idle] shutdown complete");
  process.exit(0);
}

process.on("SIGTERM", () => shutdown("SIGTERM"));
process.on("SIGINT",  () => shutdown("SIGINT"));

// ─── Startup ─────────────────────────────────────────────────────────────────

function sleep(ms) {
  return new Promise((resolve) => setTimeout(resolve, ms));
}

function assertEnv() {
  const missing = [];
  if (!process.env.DATABASE_URL) missing.push("DATABASE_URL");
  if (!ADMIN_SECRET) missing.push("ADMIN_SECRET / NUXT_ADMIN_SECRET");
  if (missing.length > 0) {
    console.error(
      `[idle] FEHLER: fehlende Env-Vars: ${missing.join(", ")}. Daemon startet nicht.`,
    );
    process.exit(1);
  }
}

async function main() {
  assertEnv();

  console.log(
    `[idle] starting — backend=${BACKEND_URL} env=${process.env.NODE_ENV ?? "unknown"}`,
  );

  // Initialer Load
  await refreshConnections();

  // Periodischer DB-Refresh
  setInterval(() => {
    if (!shuttingDown) refreshConnections();
  }, DB_REFRESH_INTERVAL_MS);
}

main().catch((err) => {
  console.error("[idle] fatal startup error:", err);
  process.exit(1);
});