email: group messages into conversations (Gmail-style threading)

Hybrid grouping via a new thread_id column: new mail threads exactly on References/In-Reply-To (id is sha1(Message-Id), so a referenced id hashes to the ancestor's own id); already-synced mail is backfilled with a normalized-subject + counterpart key. Folder views collapse to one row per thread with a count badge; the reader shows the thread as a collapsible stack. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-24 08:21:40 +00:00
parent 18d69616d2
commit 58373c9d59
6 changed files with 321 additions and 103 deletions
@@ -1,4 +1,5 @@
 import { Database } from 'bun:sqlite';
+import { createHash } from 'node:crypto';
 import { join } from 'node:path';
 import { chmodSync } from 'node:fs';
 import { DATA_PATH } from '@@/data-path';
@@ -22,7 +23,8 @@ CREATE TABLE IF NOT EXISTS emails (
  attachment_count INTEGER DEFAULT 0,
  read INTEGER DEFAULT 0,
  deleted INTEGER DEFAULT 0,
-  labels TEXT
+  labels TEXT,
+  thread_id TEXT
 );

 CREATE TABLE IF NOT EXISTS attachments (
@@ -48,6 +50,7 @@ CREATE INDEX IF NOT EXISTS idx_emails_from_address ON emails(from_address);
 CREATE INDEX IF NOT EXISTS idx_emails_integration ON emails(integration);
 CREATE INDEX IF NOT EXISTS idx_emails_email_account ON emails(email_account);
 CREATE INDEX IF NOT EXISTS idx_emails_labels ON emails(labels);
+CREATE INDEX IF NOT EXISTS idx_emails_thread ON emails(thread_id);
 `;

 /** Convert label IDs to lowercase comma-separated string for storage */
@@ -74,6 +77,59 @@ function extractDomain(address: string): string {
  return at >= 0 ? address.slice(at + 1) : '';
 }

+// ── Conversation threading ──
+// `id` is sha1(Message-Id) (see resync.messageIdToStableId), so hashing a referenced Message-Id the
+// same way yields the *id of that referenced email*. That makes header-based threading trivial:
+// a reply's thread_id is the hash of its root Message-Id, which equals the root email's own id.
+
+const hashMsgId = (msgId: string): string => createHash('sha1').update(msgId).digest('hex').slice(0, 16);
+
+/** Ordered Message-Ids this email references (References first, root→leaf; else In-Reply-To). */
+function extractReferenceIds(raw: string): string[] {
+  const refs = extractFullHeader(raw, 'References') || extractFullHeader(raw, 'In-Reply-To');
+  return Array.from(refs.matchAll(/<([^>]+)>/g), (m) => m[1]!.trim()).filter(Boolean);
+}
+
+/** Compute a header-based thread_id for a freshly ingested email (falls back to its own id = new thread). */
+function computeThreadId(db: Database, id: string, raw: string): string {
+  const refIds = extractReferenceIds(raw);
+  if (refIds.length === 0) return id; // no ancestors → this email is a thread root
+  const hashed = refIds.map(hashMsgId);
+  // Adopt an ancestor's thread if we already have one stored (robust to In-Reply-To-only clients).
+  const placeholders = hashed.map(() => '?').join(',');
+  const found = db
+    .query(`SELECT thread_id FROM emails WHERE id IN (${placeholders}) AND thread_id IS NOT NULL LIMIT 1`)
+    .get(...hashed) as { thread_id: string } | null;
+  return found?.thread_id ?? hashMsgId(refIds[0]!);
+}
+
+const RE_PREFIX = /^\s*((re|fwd?|aw|wg|sv|vs|res|antw)\s*(\[\d+\])?\s*:\s*)+/i;
+
+/** Normalize a subject for fallback grouping: strip reply/forward prefixes, fold whitespace, lowercase. */
+function normalizeSubject(subject: string): string {
+  return subject.replace(RE_PREFIX, '').replace(/\s+/g, ' ').trim().toLowerCase();
+}
+
+const firstAddress = (value: unknown): string => {
+  if (typeof value !== 'string') return '';
+  const first = value.split(',')[0] ?? '';
+  return (first.match(/<([^>]+)>/)?.[1] ?? first).trim().toLowerCase();
+};
+
+/**
+ * Subject-based thread key for mail synced before header capture (no References available).
+ * Groups by normalized subject + the counterpart address, so recurring 1:1 conversations collapse
+ * while unrelated same-subject mail from different people stays apart. Trivial subjects stay ungrouped.
+ */
+function fallbackThreadId(row: { id: string; subject: unknown; from_address: unknown; to_address: unknown; email_account: unknown }): string {
+  const norm = normalizeSubject(typeof row.subject === 'string' ? row.subject : '');
+  if (!norm) return row.id;
+  const me = typeof row.email_account === 'string' ? row.email_account.toLowerCase() : '';
+  const from = typeof row.from_address === 'string' ? row.from_address.toLowerCase() : '';
+  const counterpart = from && from !== me ? from : firstAddress(row.to_address) || from;
+  return `s:${norm}|${counterpart}`;
+}
+
 export function openEmailDb(email: string): Database {
  const dbPath = join(DATA_PATH, email, 'emails.db');
  const db = new Database(dbPath, { create: true });
@@ -108,6 +164,13 @@ function migrate(db: Database): void {
  if (!colNames.has('labels')) {
    db.exec('ALTER TABLE emails ADD COLUMN labels TEXT');
  }
+  if (!colNames.has('thread_id')) {
+    db.exec('ALTER TABLE emails ADD COLUMN thread_id TEXT');
+  }
+
+  // Backfill thread_id for any rows missing it. Header data isn't kept for already-synced mail, so
+  // these use the subject-based fallback. New mail gets an exact header-based thread_id at insert.
+  backfillThreadIds(db);

  // Ensure sync_meta table exists (for DBs created before it was added to SCHEMA_TABLES)
  db.exec('CREATE TABLE IF NOT EXISTS sync_meta (key TEXT PRIMARY KEY, value TEXT)');
@@ -120,6 +183,23 @@ function migrate(db: Database): void {
  }
 }

+function backfillThreadIds(db: Database): void {
+  const rows = db
+    .query('SELECT id, subject, from_address, to_address, email_account FROM emails WHERE thread_id IS NULL')
+    .all() as Array<{ id: string; subject: unknown; from_address: unknown; to_address: unknown; email_account: unknown }>;
+  if (rows.length === 0) return;
+
+  const update = db.prepare('UPDATE emails SET thread_id = ? WHERE id = ?');
+  db.exec('BEGIN');
+  try {
+    for (const row of rows) update.run(fallbackThreadId(row), row.id);
+    db.exec('COMMIT');
+  } catch (err) {
+    db.exec('ROLLBACK');
+    throw err;
+  }
+}
+
 // ── Full-text search (FTS5) ──
 // A standalone FTS5 index kept in sync with `emails` on every upsert. unicode61 + diacritic folding
 // gives accent-insensitive matching; per-term prefix queries make it feel incremental.
@@ -267,11 +347,12 @@ type ParsedEmail = {
  text?: string;
  attachments: Array<{ filename: string; size: number; contentType: string; content: string }>;
  labels?: string[];
+  threadId?: string;
 };

 const upsertEmailStmt = `
-  INSERT OR REPLACE INTO emails (id, integration, email_account, from_name, from_address, from_domain, to_address, cc, subject, date, snippet, html, text_body, attachment_count, labels)
-  VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+  INSERT OR REPLACE INTO emails (id, integration, email_account, from_name, from_address, from_domain, to_address, cc, subject, date, snippet, html, text_body, attachment_count, labels, thread_id)
+  VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
 `;

 const deleteAttachmentsStmt = 'DELETE FROM attachments WHERE email_id = ?';
@@ -298,6 +379,7 @@ export function upsertEmail(db: Database, email: ParsedEmail): void {
      email.text ?? null,
      email.attachments.length,
      labelsToString(email.labels),
+      email.threadId ?? email.id,
    ]);

    db.run(deleteAttachmentsStmt, [email.id]);
@@ -338,9 +420,10 @@ export function upsertFromRawEml({ db, id, raw, integration, emailAccount, label
  const domain = extractDomain(address);

  const { html, text } = extractBody(raw);
+  const threadId = computeThreadId(db, id, raw);

  db.run(upsertEmailStmt, [
-    id, integration, emailAccount, name, address, domain, to, cc, subject, date, snippet, html, text, attachments.length, labelsToString(labels),
+    id, integration, emailAccount, name, address, domain, to, cc, subject, date, snippet, html, text, attachments.length, labelsToString(labels), threadId,
  ]);

  if (attachments.length > 0) {
@@ -372,6 +455,8 @@ export function rowToSummary(row: Record<string, unknown>): EmailSummary {
    ...(row.attachment_count ? { attachmentCount: row.attachment_count as number } : {}),
    ...(row.read ? { read: true } : {}),
    ...(labels ? { labels } : {}),
+    ...(row.thread_count && (row.thread_count as number) > 1 ? { threadCount: row.thread_count as number } : {}),
+    ...(row.thread_unread ? { threadUnread: row.thread_unread as number } : {}),
  };
 }

@@ -173,10 +173,24 @@ emailRouter.get('/messages', async (ctx) => {

  const db = openEmailDb(email);
  try {
+    // One row per conversation: the latest message in each thread within this folder, plus the
+    // thread's message count and how many are unread. COALESCE guards any un-backfilled rows.
    const rows = db
-      .query(`SELECT * FROM emails WHERE ${folderWhere} ORDER BY date DESC LIMIT ? OFFSET ?`)
+      .query(
+        `SELECT * FROM (
+           SELECT e.*,
+             COUNT(*) OVER (PARTITION BY COALESCE(thread_id, id)) AS thread_count,
+             SUM(CASE WHEN read = 0 THEN 1 ELSE 0 END) OVER (PARTITION BY COALESCE(thread_id, id)) AS thread_unread,
+             ROW_NUMBER() OVER (PARTITION BY COALESCE(thread_id, id) ORDER BY date DESC, id DESC) AS rn
+           FROM emails e
+           WHERE ${folderWhere}
+         ) WHERE rn = 1
+         ORDER BY date DESC LIMIT ? OFFSET ?`,
+      )
      .all(limit, offset) as Record<string, unknown>[];
-    const countRow = db.query(`SELECT COUNT(*) as total FROM emails WHERE ${folderWhere}`).get() as { total: number };
+    const countRow = db
+      .query(`SELECT COUNT(DISTINCT COALESCE(thread_id, id)) as total FROM emails WHERE ${folderWhere}`)
+      .get() as { total: number };
    const messages = rows.map(rowToSummary);
    return ctx.json({ messages, total: countRow.total });
  } finally {
@@ -184,6 +198,32 @@ emailRouter.get('/messages', async (ctx) => {
  }
 });

+function buildMessage(db: ReturnType<typeof openEmailDb>, row: Record<string, unknown>): EmailMessage {
+  const attachmentRows = db.query('SELECT * FROM attachments WHERE email_id = ? ORDER BY idx').all(row.id as string) as Array<
+    Record<string, unknown>
+  >;
+  const from = row.from_name ? `${row.from_name} <${row.from_address}>` : (row.from_address as string);
+  return {
+    id: row.id as string,
+    from,
+    to: row.to_address as string,
+    cc: (row.cc as string) ?? undefined,
+    subject: row.subject as string,
+    date: row.date as string,
+    snippet: row.snippet as string,
+    html: (row.html as string) ?? undefined,
+    text: (row.text_body as string) ?? undefined,
+    attachments: attachmentRows.map((a) => ({
+      filename: a.filename as string,
+      size: a.size as number,
+      contentType: a.content_type as string,
+    })),
+    ...(row.attachment_count ? { attachmentCount: row.attachment_count as number } : {}),
+    ...(row.read ? { read: true } : {}),
+    ...(row.labels ? { labels: (row.labels as string).split(',') } : {}),
+  };
+}
+
 emailRouter.get('/messages/:id', async (ctx) => {
  const email = ctx.get('user').email;
  const id = ctx.req.param('id');
@@ -192,34 +232,47 @@ emailRouter.get('/messages/:id', async (ctx) => {
  try {
    const row = db.query('SELECT * FROM emails WHERE id = ? AND deleted = 0').get(id) as Record<string, unknown> | null;
    if (!row) return ctx.text('Not found', 404);
+    return ctx.json(buildMessage(db, row));
+  } finally {
+    db.close();
+  }
+});

-    const attachmentRows = db.query('SELECT * FROM attachments WHERE email_id = ? ORDER BY idx').all(id) as Array<
-      Record<string, unknown>
-    >;
+// GET /thread/:id — the full conversation containing message :id, oldest message first.
+emailRouter.get('/thread/:id', (ctx) => {
+  const email = ctx.get('user').email;
+  const id = ctx.req.param('id');

-    const from = row.from_name ? `${row.from_name} <${row.from_address}>` : (row.from_address as string);
+  const db = openEmailDb(email);
+  try {
+    const head = db.query('SELECT thread_id, subject FROM emails WHERE id = ? AND deleted = 0').get(id) as
+      | { thread_id: string | null; subject: string }
+      | null;
+    if (!head) return ctx.text('Not found', 404);

-    const message: EmailMessage = {
-      id: row.id as string,
-      from,
-      to: row.to_address as string,
-      cc: (row.cc as string) ?? undefined,
-      subject: row.subject as string,
-      date: row.date as string,
-      snippet: row.snippet as string,
-      html: (row.html as string) ?? undefined,
-      text: (row.text_body as string) ?? undefined,
-      attachments: attachmentRows.map((a) => ({
-        filename: a.filename as string,
-        size: a.size as number,
-        contentType: a.content_type as string,
-      })),
-      ...(row.attachment_count ? { attachmentCount: row.attachment_count as number } : {}),
-      ...(row.read ? { read: true } : {}),
-      ...(row.labels ? { labels: (row.labels as string).split(',') } : {}),
-    };
+    const threadKey = head.thread_id ?? id;
+    const rows = db
+      .query('SELECT * FROM emails WHERE COALESCE(thread_id, id) = ? AND deleted = 0 ORDER BY date ASC')
+      .all(threadKey) as Record<string, unknown>[];
+    const messages = rows.map((row) => buildMessage(db, row));
+    return ctx.json({ id, subject: head.subject, messages });
+  } finally {
+    db.close();
+  }
+});

-    return ctx.json(message);
+// PATCH /thread/:id/read — mark every message in the conversation as read.
+emailRouter.patch('/thread/:id/read', (ctx) => {
+  const email = ctx.get('user').email;
+  const id = ctx.req.param('id');
+
+  const db = openEmailDb(email);
+  try {
+    const head = db.query('SELECT thread_id FROM emails WHERE id = ?').get(id) as { thread_id: string | null } | null;
+    if (!head) return ctx.text('Not found', 404);
+    const threadKey = head.thread_id ?? id;
+    db.run('UPDATE emails SET read = 1 WHERE COALESCE(thread_id, id) = ? AND read = 0', [threadKey]);
+    return ctx.json({ ok: true });
  } finally {
    db.close();
  }