inbound-email/server.js

const crypto = require('crypto');
const fs = require('fs');
const path = require('path');
const SMTPServer = require('smtp-server').SMTPServer;
const config = require('./config');
const { parseEmail } = require('./services/emailParser');
const { sendToWebhook } = require('./services/webhookService');
require('aws-sdk/lib/maintenance_mode_message').suppress = true;
const Queue = require('better-queue');
const winston = require('winston');
require('winston-daily-rotate-file');

const logger = winston.createLogger({
  level: 'info',
  format: winston.format.combine(
    winston.format.timestamp(),
    winston.format.json()
  ),
  transports: [
    // Supervisor captures stdout/stderr. A regular file avoids EPIPE loops
    // when Supervisor rotates its child logs.
    new winston.transports.DailyRotateFile({
      filename: path.join(config.LOG_DIR, 'application-%DATE%.log'),
      datePattern: 'YYYY-MM-DD',
      zippedArchive: true,
      maxSize: '20m',
      maxFiles: '90d'
    })
  ]
});

function validateConfig() {
  if (!config.WEBHOOK_URL) {
    throw new Error('Missing required configuration: WEBHOOK_URL');
  }
  if (!/^https?:\/\//i.test(config.WEBHOOK_URL)) {
    throw new Error('WEBHOOK_URL must use http:// or https://');
  }
  if (!Number.isInteger(config.PORT) || config.PORT < 1 || config.PORT > 65535) {
    throw new Error('PORT must be an integer between 1 and 65535');
  }
  if (!Number.isInteger(config.MAX_FILE_SIZE) || config.MAX_FILE_SIZE < 0) {
    throw new Error('MAX_FILE_SIZE must be a non-negative integer');
  }
  if (!Number.isInteger(config.MAX_MESSAGE_SIZE) || config.MAX_MESSAGE_SIZE < 0) {
    throw new Error('MAX_MESSAGE_SIZE must be a non-negative integer');
  }
  if (!Number.isInteger(config.MAX_SMTP_CLIENTS) || config.MAX_SMTP_CLIENTS < 1) {
    throw new Error('MAX_SMTP_CLIENTS must be a positive integer');
  }
  if (!Number.isInteger(config.WEBHOOK_CONCURRENCY) || config.WEBHOOK_CONCURRENCY < 1) {
    throw new Error('WEBHOOK_CONCURRENCY must be a positive integer');
  }
  if (!Number.isInteger(config.WEBHOOK_QUEUE_RETRIES) || config.WEBHOOK_QUEUE_RETRIES < 0) {
    throw new Error('WEBHOOK_QUEUE_RETRIES must be a non-negative integer');
  }
  if (!Number.isInteger(config.WEBHOOK_QUEUE_RETRY_DELAY) || config.WEBHOOK_QUEUE_RETRY_DELAY < 1000) {
    throw new Error('WEBHOOK_QUEUE_RETRY_DELAY must be at least 1000 milliseconds');
  }
  if (!Number.isInteger(config.WEBHOOK_QUEUE_TIMEOUT) || config.WEBHOOK_QUEUE_TIMEOUT < 1000) {
    throw new Error('WEBHOOK_QUEUE_TIMEOUT must be at least 1000 milliseconds');
  }
  if (config.SMTP_SECURE && (!config.TLS_KEY_PATH || !config.TLS_CERT_PATH)) {
    throw new Error('SMTP_SECURE=true requires TLS_KEY_PATH and TLS_CERT_PATH');
  }
}

try {
  validateConfig();
} catch (error) {
  logger.error('Configuration error:', { message: error.message });
  process.exit(1);
}

const pendingDir = path.join(config.SPOOL_DIR, 'pending');
const failedDir = path.join(config.SPOOL_DIR, 'failed');
let shuttingDown = false;
let spoolScanTimer = null;
let activeJobs = 0;
const activeJobWaiters = new Set();
const enqueuedJobIds = new Set();

try {
  fs.mkdirSync(pendingDir, { recursive: true, mode: 0o700 });
  fs.mkdirSync(failedDir, { recursive: true, mode: 0o700 });
} catch (error) {
  logger.error('Failed to initialize spool directories:', {
    message: error.message,
    stack: error.stack
  });
  process.exit(1);
}

function notifyActiveJobWaiters() {
  if (activeJobs !== 0) return;
  for (const waiter of activeJobWaiters) waiter();
  activeJobWaiters.clear();
}

function waitForActiveJobs(timeout) {
  if (activeJobs === 0) return Promise.resolve();

  return new Promise(resolve => {
    let timer;
    const onDone = () => {
      clearTimeout(timer);
      activeJobWaiters.delete(onDone);
      resolve();
    };

    timer = setTimeout(onDone, timeout);
    activeJobWaiters.add(onDone);
  });
}

function delay(milliseconds) {
  return new Promise(resolve => setTimeout(resolve, milliseconds));
}

function logWebhookError(error) {
  const message = error instanceof Error ? error.message : String(error);
  const stack = error instanceof Error ? error.stack : undefined;
  logger.error('Webhook error:', { message, stack });

  if (error && error.response) {
    let data = error.response.data;
    if (typeof data === 'string') data = data.slice(0, 500);
    logger.error('Webhook response error:', {
      status: error.response.status,
      data
    });
  }
}

async function sendWithRetry(parsed) {
  const maxRetries = 3;
  let retries = 0;

  while (true) {
    try {
      await sendToWebhook(parsed);
      return;
    } catch (error) {
      logWebhookError(error);
      if (retries >= maxRetries) throw error;

      retries++;
      logger.info(`Retrying webhook (attempt ${retries}/${maxRetries})`);
      await delay(1000 * retries);
    }
  }
}

async function persistEmail(parsed) {
  const id = `${Date.now()}-${process.pid}-${crypto.randomUUID()}`;
  const filePath = path.join(pendingDir, `${id}.json`);
  const temporaryPath = `${filePath}.tmp`;

  let fileHandle;
  try {
    fileHandle = await fs.promises.open(temporaryPath, 'w', 0o600);
    await fileHandle.writeFile(JSON.stringify(parsed), 'utf8');
    await fileHandle.sync();
    await fileHandle.close();
    fileHandle = undefined;
    await fs.promises.rename(temporaryPath, filePath);
    return { id, filePath };
  } catch (error) {
    if (fileHandle) {
      await fileHandle.close().catch(() => {});
    }
    await fs.promises.unlink(temporaryPath).catch(() => {});
    throw error;
  }
}

async function quarantineFile(filePath) {
  const target = path.join(
    failedDir,
    `${path.basename(filePath)}.${Date.now()}.failed`
  );
  await fs.promises.rename(filePath, target);
  return target;
}

async function processWebhookJob(job, cb) {
  activeJobs++;

  try {
    let parsed;
    try {
      parsed = JSON.parse(await fs.promises.readFile(job.filePath, 'utf8'));
    } catch (error) {
      if (error.code === 'ENOENT') {
        logger.warn('Spool job disappeared before processing', { jobId: job.id });
        cb(null);
        return;
      }

      try {
        const failedPath = await quarantineFile(job.filePath);
        logger.error('Invalid spool job moved to failed spool', {
          jobId: job.id,
          failedPath,
          message: error.message
        });
        cb(null);
      } catch (quarantineError) {
        cb(quarantineError);
      }
      return;
    }

    await sendWithRetry(parsed);
    await fs.promises.unlink(job.filePath);
    logger.info('Successfully sent to webhook', { jobId: job.id });
    cb(null);
  } catch (error) {
    logWebhookError(error);
    cb(error);
  } finally {
    activeJobs--;
    notifyActiveJobWaiters();
  }
}

const webhookQueue = new Queue(processWebhookJob, {
  concurrent: config.WEBHOOK_CONCURRENCY,
  maxRetries: config.WEBHOOK_QUEUE_RETRIES,
  retryDelay: config.WEBHOOK_QUEUE_RETRY_DELAY,
  maxTimeout: config.WEBHOOK_QUEUE_TIMEOUT
});

webhookQueue.on('error', error => {
  logger.error('Webhook queue error:', { message: error.message, stack: error.stack });
});

webhookQueue.on('task_failed', (jobId, error) => {
  logger.error('Webhook queue task exhausted retries:', {
    jobId,
    message: error && error.message ? error.message : String(error)
  });
});

function enqueueSpoolFile(filePath) {
  const id = path.basename(filePath, '.json');
  if (enqueuedJobIds.has(id)) return;

  enqueuedJobIds.add(id);
  const ticket = webhookQueue.push({ id, filePath });
  ticket.once('finish', () => enqueuedJobIds.delete(id));
  ticket.once('failed', () => enqueuedJobIds.delete(id));
}

async function scanSpool() {
  if (shuttingDown) return;

  try {
    const entries = await fs.promises.readdir(pendingDir, { withFileTypes: true });
    for (const entry of entries) {
      if (entry.isFile() && entry.name.endsWith('.json')) {
        enqueueSpoolFile(path.join(pendingDir, entry.name));
      }
    }
  } catch (error) {
    logger.error('Failed to scan spool:', { message: error.message, stack: error.stack });
  }
}

const smtpOptions = {
  maxClients: config.MAX_SMTP_CLIENTS,
  socketTimeout: config.SMTP_SOCKET_TIMEOUT,
  closeTimeout: config.SMTP_CLOSE_TIMEOUT,
  size: config.MAX_MESSAGE_SIZE || undefined,
  onConnect(session, callback) {
    if (
      config.SMTP_ALLOWED_IPS.length > 0 &&
      !config.SMTP_ALLOWED_IPS.includes(session.remoteAddress)
    ) {
      const error = new Error('Connection not allowed');
      error.responseCode = 421;
      callback(error);
      return;
    }
    callback();
  },
  onData(stream, session, callback) {
    parseEmail(stream)
      .then(async parsed => {
        const job = await persistEmail(parsed);
        enqueueSpoolFile(job.filePath);
        logger.info('Email persisted and added to queue', {
          jobId: job.id,
          queued: webhookQueue.length
        });
        callback();
      })
      .catch(error => {
        logger.error('Parsing or persistence error:', {
          message: error.message,
          stack: error.stack
        });
        callback(new Error('Failed to persist email'));
      });
  },
  disabledCommands: ['AUTH'],
  secure: config.SMTP_SECURE
};

if (config.SMTP_SECURE) {
  smtpOptions.key = fs.readFileSync(config.TLS_KEY_PATH);
  smtpOptions.cert = fs.readFileSync(config.TLS_CERT_PATH);
}

const server = new SMTPServer(smtpOptions);

let serverReady = false;
server.on('error', error => {
  logger.error('SMTP server error:', { message: error.message, stack: error.stack });
  if (!serverReady) {
    // Let the file transport flush the startup error before exiting. Since the
    // listener never came up, there are no active handles keeping the process
    // alive; the non-zero exit code still lets Supervisor restart it.
    process.exitCode = 1;
  }
});

function gracefulShutdown(reason) {
  if (shuttingDown) return;
  shuttingDown = true;
  logger.info(`Shutting down: ${reason}`);
  clearInterval(spoolScanTimer);
  webhookQueue.pause();

  const cleanShutdown = reason === 'SIGTERM signal received' || reason === 'SIGINT signal received';
  const exitCode = cleanShutdown ? 0 : 1;
  const shutdownDeadline = setTimeout(() => {
    logger.error('Shutdown timeout reached; exiting with pending spool jobs preserved');
    process.exit(exitCode);
  }, 15000);

  server.close(async () => {
    await waitForActiveJobs(10000);
    clearTimeout(shutdownDeadline);
    logger.info('Server closed. Exiting process.');
    process.exit(exitCode);
  });
}

process.on('uncaughtException', error => {
  logger.error('Uncaught exception:', { message: error.message, stack: error.stack });
  gracefulShutdown('Uncaught exception');
});

process.on('unhandledRejection', reason => {
  const error = reason instanceof Error
    ? { message: reason.message, stack: reason.stack }
    : { message: String(reason) };
  logger.error('Unhandled Rejection:', error);
  gracefulShutdown('Unhandled rejection');
});

process.on('SIGTERM', () => gracefulShutdown('SIGTERM signal received'));
process.on('SIGINT', () => gracefulShutdown('SIGINT signal received'));

server.listen(config.PORT, '0.0.0.0', () => {
  serverReady = true;
  logger.info(`SMTP server listening on port ${config.PORT} on all interfaces`);
  void scanSpool();
  spoolScanTimer = setInterval(() => void scanSpool(), 60 * 1000);
  spoolScanTimer.unref();
});