How would you design a distributed email service?

// Outbound mail delivery on a fake clock: SMTP reply classes (2xx done, 4xx retry, 5xx bounce),
// exponential backoff, a 5-day queue lifetime, a suppression list, and a DKIM body hash.
const crypto = require("crypto");

// Scripted remote servers: the reply for attempt 1, 2, 3, ... (the last one repeats)
const remote = {
  "alice@example.org": ["250 2.0.0 OK"],
  "bob@busy.example": ["451 4.7.1 greylisted, try later", "451 4.7.1 greylisted, try later", "250 2.0.0 OK"],
  "carol@gone.example": ["550 5.1.1 user unknown"],
  "dave@down.example": ["421 4.3.2 service not available"],
};
const LIFETIME = 5 * 24 * 60; // minutes; Postfix default maximal_queue_lifetime is 5d
const backoff = (n) => Math.min(5 * 2 ** (n - 1), 240); // 5, 10, 20, ... capped at 4 hours
const suppressed = new Set();
const triesBy = new Map();
const fmtT = (m) => "t+" + Math.floor(m / 1440) + "d" + String(Math.floor((m % 1440) / 60)).padStart(2, "0") + "h" + String(m % 60).padStart(2, "0") + "m";

function deliver(queue) {
  const log = [];
  let attempts = 0;
  while (queue.length) {
    queue.sort((a, b) => a.due - b.due || (a.rcpt < b.rcpt ? -1 : 1));
    const job = queue.shift();
    if (suppressed.has(job.rcpt)) { log.push(fmtT(job.due) + " " + job.rcpt + ": skipped, on suppression list"); continue; }
    job.tries++;
    attempts++;
    triesBy.set(job.rcpt, job.tries);
    const replies = remote[job.rcpt];
    const reply = replies[Math.min(job.tries, replies.length) - 1];
    const cls = reply[0];
    if (cls === "2") log.push(fmtT(job.due) + " " + job.rcpt + ": " + reply + " after " + job.tries + " attempt(s)");
    else if (cls === "5") { suppressed.add(job.rcpt); log.push(fmtT(job.due) + " " + job.rcpt + ": " + reply + " -> hard bounce, suppress, notify sender"); }
    else {
      const next = job.due + backoff(job.tries);
      if (next - job.queued > LIFETIME) log.push(fmtT(job.due) + " " + job.rcpt + ": " + reply + " -> gave up after " + job.tries + " attempts, bounce to sender");
      else { if (job.tries <= 3) log.push(fmtT(job.due) + " " + job.rcpt + ": " + reply + " -> retry at " + fmtT(next)); job.due = next; queue.push(job); }
    }
  }
  return [log, attempts];
}

console.log("== 1. Send one message to four recipients ==");
let [log, attempts] = deliver(Object.keys(remote).map((rcpt) => ({ rcpt, due: 0, queued: 0, tries: 0 })));
for (const l of log) console.log("  " + l);
console.log("  total SMTP attempts: " + attempts + " (" + [...triesBy.entries()].map(([r, n]) => r.split("@")[0] + " " + n).join(", ") + ")");

console.log("\n== 2. Next day, a newsletter to the same list ==");
[log, attempts] = deliver(["alice@example.org", "carol@gone.example"].map((rcpt) => ({ rcpt, due: 1440, queued: 1440, tries: 0 })));
for (const l of log) console.log("  " + l);
console.log("  suppression list: " + [...suppressed].sort().join(", "));

console.log("\n== 3. DKIM body hash (bh=) survives relays, not edits ==");
const bh = (body) => crypto.createHash("sha256").update(body.replace(/\r?\n/g, "\r\n")).digest("base64");
const body = "Hi Alice,\nYour order has shipped.\n";
const signed = bh(body);
console.log("  signed bh  = " + signed);
console.log("  relay keeps body:        bh " + (bh(body) === signed ? "matches -> dkim=pass" : "differs -> dkim=fail"));
const footer = body + "-- Sent via list-server\n";
console.log("  list appends a footer:   bh " + (bh(footer) === signed ? "matches -> dkim=pass" : "differs -> dkim=fail"));

// Mailbox storage: capacity arithmetic, metadata sharded by user, threading by Message-ID /
// In-Reply-To / References headers (vs. by subject), and unread counters per folder.
console.log("== 1. Capacity (assumptions) ==");
const users = 1e9, sentPerDay = 10, recvPerDay = 40, msgKB = 50, attachPct = 20, attachKB = 500;
console.log("send QPS = 1e9 x 10 / 86400 = " + Math.round((users * sentPerDay) / 86400));
console.log("new mail per year = " + (users * recvPerDay * msgKB * 365) / 1e12 + " PB of bodies + " + (users * recvPerDay * attachPct * attachKB * 365) / 100 / 1e12 + " PB of attachments");

const fnv = (s) => { let h = 0x811c9dc5; for (const ch of s) { h ^= ch.charCodeAt(0); h = Math.imul(h, 0x01000193) >>> 0; } return h; };
const SHARDS = 16;
console.log("\n== 2. Every row of a mailbox lives on one shard (partition key = user_id) ==");
for (const u of ["u:alice", "u:bob", "u:carol"]) console.log("  " + u + " -> shard " + (fnv(u) % SHARDS));

// Alice's inbox, in arrival order. The reply from Dan arrives before the message it answers.
const inbox = [
  { id: "<a1@x>", t: 1, from: "bob", subject: "Trip plan", inReplyTo: null, refs: [] },
  { id: "<c3@x>", t: 2, from: "dan", subject: "Re: Trip plan", inReplyTo: "<b2@x>", refs: ["<a1@x>", "<b2@x>"] },
  { id: "<b2@x>", t: 3, from: "carol", subject: "Re: Trip plan", inReplyTo: "<a1@x>", refs: ["<a1@x>"] },
  { id: "<h1@y>", t: 4, from: "eve", subject: "Hello", inReplyTo: null, refs: [] },
  { id: "<h2@z>", t: 5, from: "frank", subject: "Hello", inReplyTo: null, refs: [] },
  { id: "<d4@x>", t: 6, from: "bob", subject: "Re: Trip plan", inReplyTo: "<c3@x>", refs: ["<a1@x>", "<b2@x>", "<c3@x>"] },
];

console.log("\n== 3. Threading ==");
const irt = new Map(); // In-Reply-To only: a reply that arrives before its parent starts a new thread
const byIrt = new Map();
for (const m of inbox) {
  const tid = m.inReplyTo && irt.has(m.inReplyTo) ? irt.get(m.inReplyTo) : m.id;
  irt.set(m.id, tid);
  if (!byIrt.has(tid)) byIrt.set(tid, []);
  byIrt.get(tid).push(m.from);
}
for (const [tid, who] of byIrt) console.log("  in-reply-to " + tid.padEnd(7) + " " + who.join(", "));
const threadOf = new Map(); // message-id -> thread id
const byHeaders = new Map();
for (const m of inbox) {
  const known = [m.inReplyTo, ...m.refs].filter((r) => r && threadOf.has(r)).map((r) => threadOf.get(r));
  const root = m.refs.length ? m.refs[0] : m.inReplyTo; // References starts with the thread root
  const tid = known.length ? known[0] : root || m.id;
  threadOf.set(m.id, tid);
  for (const r of m.refs) if (!threadOf.has(r)) threadOf.set(r, tid);
  if (!byHeaders.has(tid)) byHeaders.set(tid, []);
  byHeaders.get(tid).push(m.from);
}
for (const [tid, who] of byHeaders) console.log("  references  " + tid.padEnd(7) + " " + who.join(", "));
const bySubject = new Map();
for (const m of inbox) {
  const s = m.subject.replace(/^(re|fwd?):\s*/i, "");
  if (!bySubject.has(s)) bySubject.set(s, []);
  bySubject.get(s).push(m.from);
}
for (const [s, who] of bySubject) console.log("  by subject  " + s.padEnd(10) + " " + who.join(", ") + (s === "Hello" ? "   <- two strangers merged" : ""));

console.log("\n== 4. Folder listing and unread counters ==");
const folder = { name: "INBOX", unread: 0, msgs: [] }; // counter row updated in the same transaction as the message rows
for (const m of inbox) { folder.msgs.push({ ...m, read: false }); folder.unread++; }
const markRead = (id) => { const m = folder.msgs.find((x) => x.id === id); if (!m.read) { m.read = true; folder.unread--; } };
markRead("<a1@x>");
markRead("<a1@x>"); // a second client sends the same flag change: no double decrement
markRead("<h1@y>");
const page = [...folder.msgs].sort((a, b) => b.t - a.t).slice(0, 3);
console.log("  INBOX unread=" + folder.unread + " (recount " + folder.msgs.filter((m) => !m.read).length + "), newest 3: " + page.map((m) => m.from + (m.read ? "" : "*")).join(", "));