Add read-only mailbox browsing for email addresses

Let users navigate the conversations their workflows have recorded for an
email address. Writes still happen only through nodes, so this surface is
strictly read-only.

- MailboxDO: listThreads, listThreadAttachments, getAttachment read methods
- API: GET endpoints under /:organizationId/emails/:id for threads, thread
  detail, message bodies, and attachment downloads (org + email scoped,
  blobs served with nosniff + CSP sandbox + forced-download headers)
- Web: per-address master-detail inbox page, reached from the Emails list
- Shared mailbox browsing types

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: bchapuis

apps/api/src/durable-objects/mailbox-do.test.ts

@@ -152,6 +152,62 @@ describe("MailboxDO", () => {
     expect(messages).toHaveLength(1);
   });
 
+  it("lists threads scoped to an address", async () => {
+    const other = "email-2";
+    const { threadId: t1 } = await mailbox.ingestInbound(
+      inbound({ subject: "First" })
+    );
+    const { threadId: t2 } = await mailbox.ingestInbound(
+      inbound({ subject: "Second", fromEmail: "other@example.com" })
+    );
+    // A thread on a different address must not leak into the listing.
+    await mailbox.ingestInbound(
+      inbound({ emailId: other, subject: "Elsewhere" })
+    );
+
+    const threads = await mailbox.listThreads(EMAIL_ID, 100, 0);
+    expect(new Set(threads.map((t) => t.id))).toEqual(new Set([t1, t2]));
+
+    const elsewhere = await mailbox.listThreads(other, 100, 0);
+    expect(elsewhere).toHaveLength(1);
+    expect(elsewhere[0].subject).toBe("Elsewhere");
+  });
+
+  it("lists and fetches attachments scoped to a thread", async () => {
+    const messageId = uuidv7();
+    const attachmentId = uuidv7();
+    const r2Key = `${EMAIL_ID}/${messageId}/attachments/0-report.pdf`;
+    const { threadId } = await mailbox.ingestInbound(
+      inbound({
+        messageId,
+        attachments: [
+          {
+            id: attachmentId,
+            filename: "report.pdf",
+            contentType: "application/pdf",
+            sizeBytes: 1234,
+            r2Key,
+            contentId: null,
+          },
+        ],
+      })
+    );
+
+    const rows = await mailbox.listThreadAttachments(threadId);
+    expect(rows).toHaveLength(1);
+    expect(rows[0]).toMatchObject({
+      id: attachmentId,
+      messageId,
+      filename: "report.pdf",
+      sizeBytes: 1234,
+      r2Key,
+    });
+
+    const att = await mailbox.getAttachment(attachmentId);
+    expect(att?.r2Key).toBe(r2Key);
+    expect(await mailbox.getAttachment("missing")).toBeUndefined();
+  });
+
   it("prunes inactive threads and reports their R2 prefixes", async () => {
     const args = inbound();
     const { threadId, messageId } = await mailbox.ingestInbound(args);

apps/api/src/durable-objects/mailbox-do.ts

@@ -92,6 +92,15 @@ export interface MailboxMessageRow {
   createdAt: number;
 }
 
+export interface MailboxAttachmentRow {
+  id: string;
+  messageId: string;
+  filename: string;
+  contentType: string;
+  sizeBytes: number;
+  r2Key: string;
+}
+
 export class MailboxDO extends DurableObject<Bindings> {
   private initialized = false;
 
@@ -364,6 +373,84 @@ export class MailboxDO extends DurableObject<Bindings> {
     return { threadId };
   }
 
+  /**
+   * List threads belonging to one email address, newest activity first. Powers
+   * the read-only mailbox browser; the index on `last_message_at` keeps this
+   * cheap even as a mailbox accumulates conversations.
+   */
+  async listThreads(
+    emailId: string,
+    limit: number,
+    offset: number
+  ): Promise<MailboxThreadRow[]> {
+    this.ensureSchema();
+    const rows = this.ctx.storage.sql
+      .exec(
+        `SELECT id, email_id, subject, from_email, last_message_at, created_at
+         FROM threads WHERE email_id = ?
+         ORDER BY last_message_at DESC LIMIT ? OFFSET ?`,
+        emailId,
+        limit,
+        offset
+      )
+      .toArray() as Record<string, unknown>[];
+    return rows.map((r) => ({
+      id: r.id as string,
+      emailId: r.email_id as string,
+      subject: r.subject as string,
+      fromEmail: r.from_email as string,
+      lastMessageAt: Number(r.last_message_at),
+      createdAt: Number(r.created_at),
+    }));
+  }
+
+  /** Attachment metadata for every message in a thread, for the detail view. */
+  async listThreadAttachments(
+    threadId: string
+  ): Promise<MailboxAttachmentRow[]> {
+    this.ensureSchema();
+    const rows = this.ctx.storage.sql
+      .exec(
+        `SELECT a.id, a.message_id, a.filename, a.content_type, a.size_bytes, a.r2_key
+         FROM attachments a JOIN messages m ON m.id = a.message_id
+         WHERE m.thread_id = ?`,
+        threadId
+      )
+      .toArray() as Record<string, unknown>[];
+    return rows.map((r) => ({
+      id: r.id as string,
+      messageId: r.message_id as string,
+      filename: r.filename as string,
+      contentType: r.content_type as string,
+      sizeBytes: Number(r.size_bytes),
+      r2Key: r.r2_key as string,
+    }));
+  }
+
+  /** Single attachment by id, used to resolve its R2 blob for download. */
+  async getAttachment(
+    attachmentId: string
+  ): Promise<MailboxAttachmentRow | undefined> {
+    this.ensureSchema();
+    const rows = this.ctx.storage.sql
+      .exec(
+        `SELECT id, message_id, filename, content_type, size_bytes, r2_key
+         FROM attachments WHERE id = ? LIMIT 1`,
+        attachmentId
+      )
+      .toArray() as Record<string, unknown>[];
+    if (rows.length === 0) return undefined;
+    const r = rows[0];
+    return {
+      id: r.id as string,
+      messageId: r.message_id as string,
+      filename: r.filename as string,
+      contentType: r.content_type as string,
+      sizeBytes: Number(r.size_bytes),
+      r2Key: r.r2_key as string,
+    };
+  }
+
   async getThread(threadId: string): Promise<MailboxThreadRow | undefined> {
     this.ensureSchema();
     const rows = this.ctx.storage.sql

apps/api/src/routes/emails.ts

@@ -3,7 +3,10 @@ import type {
   CreateEmailResponse,
   DeleteEmailResponse,
   GetEmailResponse,
+  GetMailboxThreadResponse,
   ListEmailsResponse,
+  ListMailboxThreadsResponse,
+  MailboxAttachmentSummary,
   UpdateEmailRequest,
   UpdateEmailResponse,
 } from "@dafthunk/types";
@@ -23,6 +26,7 @@ import {
   updateEmail,
 } from "../db";
 import type { EmailRow } from "../db/schema";
+import { inboxKeys } from "../support-storage";
 import {
   formatEmailAddress,
   generateEmailHandle,
@@ -45,6 +49,21 @@ emailRoutes.use("*", jwtMiddleware);
 
 const nameSchema = z.string().trim().min(1, "Email name is required").max(120);
 
+// Defense-in-depth headers shared by every blob response. Bodies and
+// attachments may carry attacker-influenced content, so they must never render
+// same-origin: nosniff + a CSP sandbox + a forced download disposition together
+// neutralise script execution and top-level navigation. The frontend reads
+// these via `fetch()`, which ignores the headers, so rendering still works.
+const blobSecurityHeaders = (
+  contentType: string,
+  filename: string
+): Record<string, string> => ({
+  "Content-Type": contentType,
+  "X-Content-Type-Options": "nosniff",
+  "Content-Security-Policy": "sandbox",
+  "Content-Disposition": `attachment; filename="${filename.replace(/["\r\n]/g, "")}"`,
+});
+
 const toEmailPayload = (
   email: Pick<EmailRow, "id" | "name" | "handle" | "createdAt" | "updatedAt">,
   domain: string
@@ -231,4 +250,190 @@ emailRoutes.delete("/:id", async (c) => {
   return c.json(response);
 });
 
+// ── Read-only mailbox browsing ──────────────────────────────────────────────
+//
+// These endpoints let a user navigate the conversations recorded for one of
+// their email addresses. They are strictly read-only: messages are written
+// exclusively by workflow nodes through the Mailbox Durable Object. Each
+// endpoint first confirms the address belongs to the caller's organization,
+// then reads from the per-org mailbox (`mailbox:{organizationId}`).
+
+const mailboxStub = (env: ApiContext["Bindings"], organizationId: string) =>
+  env.MAILBOX.get(env.MAILBOX.idFromName(`mailbox:${organizationId}`));
+
+/** List the conversations recorded for an email address (newest first). */
+emailRoutes.get("/:id/threads", async (c) => {
+  const id = c.req.param("id");
+  const organizationId = c.get("organizationId")!;
+  const db = createDatabase(c.env.DB);
+
+  const email = await getEmail(db, id, organizationId);
+  if (!email) {
+    return c.json({ error: "Email not found" }, 404);
+  }
+
+  const threads = await mailboxStub(c.env, organizationId).listThreads(
+    id,
+    200,
+    0
+  );
+
+  const response: ListMailboxThreadsResponse = {
+    threads: threads.map((t) => ({
+      id: t.id,
+      subject: t.subject,
+      fromEmail: t.fromEmail,
+      lastMessageAt: t.lastMessageAt,
+      createdAt: t.createdAt,
+    })),
+  };
+  return c.json(response);
+});
+
+/** A single conversation with its messages and attachment metadata. */
+emailRoutes.get("/:id/threads/:threadId", async (c) => {
+  const id = c.req.param("id");
+  const threadId = c.req.param("threadId");
+  const organizationId = c.get("organizationId")!;
+  const db = createDatabase(c.env.DB);
+
+  // These reads are independent: the ownership check (email exists, thread
+  // belongs to it) gates the *response*, not the fetches, and the DO is already
+  // scoped to the caller's org. Fetch them concurrently and validate after.
+  const stub = mailboxStub(c.env, organizationId);
+  const [email, thread, messages, attachmentRows] = await Promise.all([
+    getEmail(db, id, organizationId),
+    stub.getThread(threadId),
+    stub.listThreadMessages(threadId),
+    stub.listThreadAttachments(threadId),
+  ]);
+
+  if (!email) {
+    return c.json({ error: "Email not found" }, 404);
+  }
+  // Confirm the thread really belongs to this address; the DO is per-org so a
+  // mismatch means the caller addressed the wrong mailbox slot.
+  if (!thread || thread.emailId !== id) {
+    return c.json({ error: "Thread not found" }, 404);
+  }
+
+  const byMessage = new Map<string, MailboxAttachmentSummary[]>();
+  for (const a of attachmentRows) {
+    const list = byMessage.get(a.messageId) ?? [];
+    list.push({
+      id: a.id,
+      filename: a.filename,
+      contentType: a.contentType,
+      sizeBytes: a.sizeBytes,
+    });
+    byMessage.set(a.messageId, list);
+  }
+
+  const response: GetMailboxThreadResponse = {
+    thread: {
+      id: thread.id,
+      subject: thread.subject,
+      fromEmail: thread.fromEmail,
+      lastMessageAt: thread.lastMessageAt,
+      createdAt: thread.createdAt,
+    },
+    messages: messages.map((m) => ({
+      id: m.id,
+      direction: m.direction,
+      fromEmail: m.fromEmail,
+      toEmail: m.toEmail,
+      subject: m.subject,
+      snippet: m.snippet,
+      hasHtml: m.hasHtml,
+      hasText: m.hasText,
+      attachmentCount: m.attachmentCount,
+      createdAt: m.createdAt,
+      attachments: byMessage.get(m.id) ?? [],
+    })),
+  };
+  return c.json(response);
+});
+
+/** Stream a message body part (text or html) from R2. */
+emailRoutes.get(
+  "/:id/messages/:messageId/body",
+  zValidator(
+    "query",
+    z.object({ part: z.enum(["text", "html"]).default("text") })
+  ),
+  async (c) => {
+    const id = c.req.param("id");
+    const messageId = c.req.param("messageId");
+    const organizationId = c.get("organizationId")!;
+    const db = createDatabase(c.env.DB);
+
+    const email = await getEmail(db, id, organizationId);
+    if (!email) {
+      return c.json({ error: "Email not found" }, 404);
+    }
+
+    const { part } = c.req.valid("query");
+    const keys = inboxKeys(id, messageId);
+    const bodyPart =
+      part === "html"
+        ? {
+            key: keys.htmlBody,
+            filename: "body.html",
+            contentType: "text/html; charset=utf-8",
+          }
+        : {
+            key: keys.textBody,
+            filename: "body.txt",
+            contentType: "text/plain; charset=utf-8",
+          };
+
+    // R2 keys are namespaced by emailId, so a body only resolves when the
+    // message genuinely lives under the address the caller just authorized.
+    const obj = await c.env.INBOXES.get(bodyPart.key);
+    if (!obj) {
+      return c.json({ error: "Body part not stored" }, 404);
+    }
+
+    return new Response(obj.body, {
+      status: 200,
+      headers: blobSecurityHeaders(bodyPart.contentType, bodyPart.filename),
+    });
+  }
+);
+
+/** Download a single attachment blob from R2. */
+emailRoutes.get("/:id/attachments/:attachmentId", async (c) => {
+  const id = c.req.param("id");
+  const attachmentId = c.req.param("attachmentId");
+  const organizationId = c.get("organizationId")!;
+  const db = createDatabase(c.env.DB);
+
+  const email = await getEmail(db, id, organizationId);
+  if (!email) {
+    return c.json({ error: "Email not found" }, 404);
+  }
+
+  const att = await mailboxStub(c.env, organizationId).getAttachment(
+    attachmentId
+  );
+  // The R2 key is prefixed with the owning emailId; require it to match the
+  // addressed mailbox so one address can't read another's blobs.
+  if (!att || !att.r2Key.startsWith(`${id}/`)) {
+    return c.json({ error: "Attachment not found" }, 404);
+  }
+
+  const obj = await c.env.INBOXES.get(att.r2Key);
+  if (!obj) {
+    return c.json({ error: "Attachment blob missing" }, 404);
+  }
+
+  return new Response(obj.body, {
+    status: 200,
+    headers: {
+      ...blobSecurityHeaders(att.contentType, att.filename),
+      "Content-Length": String(att.sizeBytes),
+    },
+  });
+});
+
 export default emailRoutes;