email-unlimit/backend/app.js

const express = require('express');
const cors = require('cors');
const http = require('http');
const { WebSocketServer } = require('ws');
const db = require('./db');
const { SMTPServer } = require('smtp-server');
const { saveEmail } = require('./saveEmail');
const emitter = require('./eventEmitter');
const logger = require('./logger');
const morgan = require('morgan');
const { validateConnection } = require('./connectionValidator');

const app = express();
const apiPort = 5182;
const smtpPort = 25;

// Setup morgan to log HTTP requests to winston
app.use(morgan('combined', { stream: { write: message => logger.info(message.trim()) } }));

app.use(cors());
app.use(express.json());

// API to get messages for a recipient
app.get('/api/messages', async (req, res) => {
  const { recipient } = req.query;
  if (!recipient) {
    logger.warn('Attempted to get messages without a recipient');
    return res.status(400).send('Recipient is required');
  }

  try {
    const [rows] = await db.execute(
      'SELECT id, sender, recipient, subject, body, CAST(received_at AS CHAR) as received_at FROM emails WHERE recipient LIKE ? ORDER BY received_at DESC',
      [`%${recipient}%`]
    );
    res.json(rows);
  } catch (error) {
    logger.error('Failed to fetch emails:', error);
    res.status(500).send('Failed to fetch emails');
  }
});

// API to get attachments for a message
app.get('/api/messages/:id/attachments', async (req, res) => {
  const { id } = req.params;
  try {
    const [rows] = await db.execute(
      'SELECT id, filename, content_type FROM email_attachments WHERE email_id = ?',
      [id]
    );
    res.json(rows);
  } catch (error) {
    logger.error('Failed to fetch attachments:', { error, emailId: id });
    res.status(500).send('Failed to fetch attachments');
  }
});

// API to download an attachment
app.get('/api/attachments/:attachmentId', async (req, res) => {
  const { attachmentId } = req.params;
  try {
    const [rows] = await db.execute(
      'SELECT filename, content_type, content FROM email_attachments WHERE id = ?',
      [attachmentId]
    );
    if (rows.length === 0) {
      return res.status(404).send('Attachment not found');
    }
    const attachment = rows[0];
    res.setHeader('Content-Disposition', `attachment; filename="${attachment.filename}"`);
    res.setHeader('Content-Type', attachment.content_type);
    res.send(attachment.content);
  } catch (error) {
    logger.error('Failed to download attachment:', { error, attachmentId });
    res.status(500).send('Failed to download attachment');
  }
});

// API to delete a message
app.delete('/api/messages/:id', async (req, res) => {
  const { id } = req.params;
  try {
    const [result] = await db.execute('DELETE FROM emails WHERE id = ?', [id]);
    if (result.affectedRows === 0) {
      logger.warn('Attempted to delete a message that was not found', { id });
      return res.status(404).send('Message not found');
    }
    logger.info('Message deleted successfully', { id });
    res.status(204).send(); // No Content
  } catch (error) {
    logger.error('Failed to delete message:', { error, id });
    res.status(500).send('Failed to delete message');
  }
});

// API to delete all messages for a recipient
app.delete('/api/messages', async (req, res) => {
  const { recipient } = req.query;
  if (!recipient) {
    logger.warn('Attempted to delete all messages without a recipient');
    return res.status(400).send('Recipient is required');
  }
  try {
    await db.execute('DELETE FROM emails WHERE recipient LIKE ?', [`%${recipient}%`]);
    logger.info('All messages for recipient deleted successfully', { recipient });
    res.status(204).send(); // No Content
  } catch (error) {
    logger.error('Failed to delete all messages:', { error, recipient });
    res.status(500).send('Failed to delete all messages');
  }
});

// Create HTTP server
const server = http.createServer(app);

// Create WebSocket server
const wss = new WebSocketServer({ server });

// Map to store connections for each recipient
const clients = new Map();

wss.on('connection', (ws, req) => {
  const url = new URL(req.url, `http://${req.headers.host}`);
  const recipient = url.searchParams.get('recipient');

  if (recipient) {
    logger.info(`WebSocket client connected`, { recipient });
    clients.set(recipient, ws);

    ws.on('close', () => {
      logger.info(`WebSocket client disconnected`, { recipient });
      clients.delete(recipient);
    });

    ws.on('error', (error) => {
      logger.error(`WebSocket error`, { recipient, error });
      clients.delete(recipient);
    });
  } else {
    logger.warn('WebSocket client connected without recipient, closing.');
    ws.close();
  }
});

emitter.on('newEmail', (email) => {
  const dbRecipient = email.recipient;  // 比如 "6471" <6471@shenjianl.cn>

  // 遍历所有 clients，用 includes 判断
  for (const [clientRecipient, ws] of clients.entries()) {
    if (dbRecipient.includes(clientRecipient)) {
      logger.info(`Sending new email notification to`, { recipient: clientRecipient });
      ws.send(JSON.stringify(email));
      break;  // 如果只想发给第一个匹配的就 break
    }
  }
});


// Start API server
server.listen(apiPort, () => {
  logger.info(`Backend API and WebSocket server listening at http://localhost:${apiPort}`);
});

// Configure and start SMTP server
const smtpServer = new SMTPServer({
  secure: false, // Enable STARTTLS
  authOptional: true,
  disabledCommands: ['AUTH'],
  onConnect(session, callback) {
    logger.info('Connection received from', { remoteAddress: session.remoteAddress });
    return validateConnection(session, callback);
  },
  onData(stream, session, callback) {
    logger.info('Receiving email...', { session });
    saveEmail(stream)
      .then(() => {
        logger.info('Email processed and saved successfully.');
        callback(); // Accept the message
      })
      .catch(err => {
        logger.error('Error processing email:', err);
        // Check if the error is a rate limit error with a specific response code
        if (err.responseCode) {
          return callback(err);
        }
        callback(new Error('Failed to process email.'));
      });
  },
});

smtpServer.on('error', err => {
  logger.error('SMTP Server Error:', err);
});

smtpServer.listen(smtpPort, () => {
  logger.info(`SMTP server listening on port ${smtpPort}`);
});