Build an Inbound Email Processor

// src/email/inbound.ts — Inbound email processing with parsing, routing, and automation
import { pool } from "../db";
import { Redis } from "ioredis";
import { randomBytes, createHash } from "node:crypto";

const redis = new Redis(process.env.REDIS_URL!);

interface InboundEmail {
  id: string;
  messageId: string;         // RFC 822 Message-ID
  inReplyTo: string | null;  // threading
  references: string[];      // full thread chain
  from: { email: string; name: string };
  to: Array<{ email: string; name: string }>;
  cc: Array<{ email: string; name: string }>;
  subject: string;
  textBody: string;
  htmlBody: string;
  strippedText: string;      // text without quoted replies
  attachments: Attachment[];
  headers: Record<string, string>;
  spamScore: number;
  receivedAt: string;
}

interface Attachment {
  filename: string;
  contentType: string;
  size: number;
  url: string;               // stored URL after upload
  contentId?: string;        // for inline images
}

interface RoutingRule {
  id: string;
  name: string;
  conditions: Array<{
    field: "from" | "to" | "subject" | "body" | "has_attachment";
    operator: "contains" | "equals" | "matches" | "is_true";
    value: string;
  }>;
  actions: Array<{
    type: "assign_team" | "add_tag" | "set_priority" | "auto_reply" | "forward" | "webhook";
    value: string;
  }>;
  priority: number;
}

// Process inbound email webhook (from SendGrid, Mailgun, Postmark, etc.)
export async function processInbound(rawPayload: any): Promise<{
  email: InboundEmail;
  ticketId: string | null;
  actions: string[];
}> {
  // Parse email from webhook payload
  const email = parseWebhookPayload(rawPayload);

  // Spam check
  if (email.spamScore > 5) {
    await pool.query(
      "INSERT INTO email_spam (message_id, from_email, subject, spam_score, received_at) VALUES ($1, $2, $3, $4, NOW())",
      [email.messageId, email.from.email, email.subject, email.spamScore]
    );
    return { email, ticketId: null, actions: ["spam_filtered"] };
  }

  // Store email
  await pool.query(
    `INSERT INTO inbound_emails (id, message_id, in_reply_to, from_email, from_name, subject, text_body, stripped_text, attachments, received_at)
     VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, NOW())`,
    [email.id, email.messageId, email.inReplyTo, email.from.email,
     email.from.name, email.subject, email.textBody, email.strippedText,
     JSON.stringify(email.attachments)]
  );

  // Thread detection
  let ticketId = await findExistingThread(email);
  const actions: string[] = [];

  if (ticketId) {
    // Add to existing ticket thread
    await pool.query(
      "INSERT INTO ticket_messages (ticket_id, email_id, content, from_email, created_at) VALUES ($1, $2, $3, $4, NOW())",
      [ticketId, email.id, email.strippedText, email.from.email]
    );
    actions.push(`added_to_ticket:${ticketId}`);
  } else {
    // Create new ticket
    ticketId = `ticket-${randomBytes(6).toString("hex")}`;
    await pool.query(
      `INSERT INTO tickets (id, subject, customer_email, customer_name, status, priority, created_at)
       VALUES ($1, $2, $3, $4, 'open', 'normal', NOW())`,
      [ticketId, email.subject, email.from.email, email.from.name]
    );
    await pool.query(
      "INSERT INTO ticket_messages (ticket_id, email_id, content, from_email, created_at) VALUES ($1, $2, $3, $4, NOW())",
      [ticketId, email.id, email.strippedText, email.from.email]
    );
    actions.push(`created_ticket:${ticketId}`);
  }

  // Apply routing rules
  const routingActions = await applyRoutingRules(email, ticketId);
  actions.push(...routingActions);

  // Process attachments
  if (email.attachments.length > 0) {
    for (const attachment of email.attachments) {
      await pool.query(
        "INSERT INTO ticket_attachments (ticket_id, filename, content_type, size, url, created_at) VALUES ($1, $2, $3, $4, $5, NOW())",
        [ticketId, attachment.filename, attachment.contentType, attachment.size, attachment.url]
      );
    }
    actions.push(`attachments:${email.attachments.length}`);
  }

  return { email, ticketId, actions };
}

// Thread detection using Message-ID chain
async function findExistingThread(email: InboundEmail): Promise<string | null> {
  // Check In-Reply-To header
  if (email.inReplyTo) {
    const { rows: [existing] } = await pool.query(
      `SELECT t.id FROM tickets t
       JOIN ticket_messages tm ON t.id = tm.ticket_id
       JOIN inbound_emails ie ON tm.email_id = ie.id
       WHERE ie.message_id = $1`,
      [email.inReplyTo]
    );
    if (existing) return existing.id;
  }

  // Check References header
  for (const ref of email.references) {
    const { rows: [existing] } = await pool.query(
      `SELECT t.id FROM tickets t
       JOIN ticket_messages tm ON t.id = tm.ticket_id
       JOIN inbound_emails ie ON tm.email_id = ie.id
       WHERE ie.message_id = $1`,
      [ref]
    );
    if (existing) return existing.id;
  }

  // Fallback: match by subject + sender within 7 days
  const normalizedSubject = email.subject.replace(/^(Re|Fwd|Fw):\s*/gi, "").trim();
  const { rows: [match] } = await pool.query(
    `SELECT id FROM tickets WHERE customer_email = $1 AND subject = $2 AND created_at > NOW() - INTERVAL '7 days' AND status != 'closed'`,
    [email.from.email, normalizedSubject]
  );
  return match?.id || null;
}

async function applyRoutingRules(email: InboundEmail, ticketId: string): Promise<string[]> {
  const { rows: rules } = await pool.query(
    "SELECT * FROM routing_rules WHERE enabled = true ORDER BY priority DESC"
  );

  const actions: string[] = [];

  for (const rule of rules) {
    const conditions = JSON.parse(rule.conditions);
    const ruleActions = JSON.parse(rule.actions);

    const matches = conditions.every((cond: any) => {
      const fieldValue = getFieldValue(email, cond.field);
      switch (cond.operator) {
        case "contains": return fieldValue.toLowerCase().includes(cond.value.toLowerCase());
        case "equals": return fieldValue.toLowerCase() === cond.value.toLowerCase();
        case "matches": return new RegExp(cond.value, "i").test(fieldValue);
        case "is_true": return !!fieldValue;
        default: return false;
      }
    });

    if (matches) {
      for (const action of ruleActions) {
        switch (action.type) {
          case "assign_team":
            await pool.query("UPDATE tickets SET assigned_team = $2 WHERE id = $1", [ticketId, action.value]);
            actions.push(`assigned:${action.value}`);
            break;
          case "set_priority":
            await pool.query("UPDATE tickets SET priority = $2 WHERE id = $1", [ticketId, action.value]);
            actions.push(`priority:${action.value}`);
            break;
          case "add_tag":
            await pool.query("UPDATE tickets SET tags = array_append(tags, $2) WHERE id = $1", [ticketId, action.value]);
            break;
          case "auto_reply":
            await redis.rpush("email:outbound", JSON.stringify({
              to: email.from.email, subject: `Re: ${email.subject}`,
              body: action.value, inReplyTo: email.messageId,
            }));
            actions.push("auto_replied");
            break;
        }
      }
      break;  // first matching rule wins
    }
  }

  return actions;
}

function getFieldValue(email: InboundEmail, field: string): string {
  switch (field) {
    case "from": return email.from.email;
    case "to": return email.to.map((t) => t.email).join(", ");
    case "subject": return email.subject;
    case "body": return email.strippedText;
    case "has_attachment": return email.attachments.length > 0 ? "true" : "";
    default: return "";
  }
}

function parseWebhookPayload(payload: any): InboundEmail {
  return {
    id: `email-${randomBytes(6).toString("hex")}`,
    messageId: payload.headers?.["Message-ID"] || payload["Message-ID"] || `${randomBytes(8).toString("hex")}@parsed`,
    inReplyTo: payload.headers?.["In-Reply-To"] || payload["In-Reply-To"] || null,
    references: (payload.headers?.["References"] || "").split(/\s+/).filter(Boolean),
    from: { email: payload.from || payload.sender || "", name: payload.from_name || "" },
    to: [{ email: payload.to || "", name: "" }],
    cc: [],
    subject: payload.subject || "(no subject)",
    textBody: payload.text || payload["body-plain"] || "",
    htmlBody: payload.html || payload["body-html"] || "",
    strippedText: payload["stripped-text"] || stripQuotedReplies(payload.text || ""),
    attachments: (payload.attachments || []).map((a: any) => ({
      filename: a.filename || a.name, contentType: a.contentType || a.type,
      size: a.size || 0, url: a.url || "",
    })),
    headers: payload.headers || {},
    spamScore: parseFloat(payload["spam-score"] || payload.spamScore || "0"),
    receivedAt: new Date().toISOString(),
  };
}

function stripQuotedReplies(text: string): string {
  const lines = text.split("\n");
  const stripped: string[] = [];
  for (const line of lines) {
    if (line.match(/^>/) || line.match(/^On .* wrote:$/)) break;
    if (line.match(/^-{3,}\s*Original Message/i)) break;
    stripped.push(line);
  }
  return stripped.join("\n").trim();
}