feat(mail): make mobile attachment sends durable

apps/web/.env.example

@@ -2,6 +2,9 @@ NEXT_PUBLIC_BASE_URL=http://localhost:3000
 
 DATABASE_URL="postgresql://postgres:password@localhost:5432/inboxzero?schema=public"
 DIRECT_URL="postgresql://postgres:password@localhost:5432/inboxzero?schema=public"
+# Optional private Vercel Blob store for mobile attachment staging:
+# BLOB_READ_WRITE_TOKEN=
+# BLOB_STORE_ID= # used with Vercel OIDC instead of BLOB_READ_WRITE_TOKEN
 # Docker Compose credentials (defaults shown; POSTGRES_PASSWORD must match DATABASE_URL):
 # POSTGRES_USER=postgres
 # POSTGRES_PASSWORD=password # change this for production

apps/web/app/api/cron/email-send-operation-retention/route.test.ts

@@ -1,16 +1,22 @@
+import { NextRequest } from "next/server";
 import { beforeEach, describe, expect, it, vi } from "vitest";
 
-const { captureExceptionMock, deleteExpiredMock } = vi.hoisted(() => ({
-  captureExceptionMock: vi.fn(),
-  deleteExpiredMock: vi.fn(),
-}));
+const { captureExceptionMock, cleanupAttachmentsMock, deleteExpiredMock } =
+  vi.hoisted(() => ({
+    captureExceptionMock: vi.fn(),
+    cleanupAttachmentsMock: vi.fn(),
+    deleteExpiredMock: vi.fn(),
+  }));
 
 vi.mock("@/env", () => ({
   env: { CRON_SECRET: "cron-secret" },
 }));
 vi.mock("@/utils/email-send-operation-retention", () => ({
   deleteExpiredEmailSendOperations: () => deleteExpiredMock(),
 }));
+vi.mock("@/utils/email/email-attachment-staging", () => ({
+  cleanupEmailAttachmentStages: () => cleanupAttachmentsMock(),
+}));
 vi.mock("@/utils/error", () => ({
   captureException: (...args: unknown[]) => captureExceptionMock(...args),
 }));
@@ -27,14 +33,19 @@ import { GET, POST } from "./route";
 describe("email send operation retention cron route", () => {
   beforeEach(() => {
     vi.clearAllMocks();
+    cleanupAttachmentsMock.mockResolvedValue({
+      deletedBlobs: 2,
+      deletedTombstones: 1,
+    });
     deleteExpiredMock.mockResolvedValue(3);
   });
 
   it("rejects requests without the cron bearer token", async () => {
     const response = await GET(
-      new Request(
+      new NextRequest(
         "http://localhost:3000/api/cron/email-send-operation-retention",
       ),
+      { params: Promise.resolve({}) },
     );
 
     expect(response.status).toBe(401);
@@ -44,26 +55,32 @@ describe("email send operation retention cron route", () => {
 
   it("deletes expired operations for an authorized GET request", async () => {
     const response = await GET(
-      new Request(
+      new NextRequest(
         "http://localhost:3000/api/cron/email-send-operation-retention",
         { headers: { authorization: "Bearer cron-secret" } },
       ),
+      { params: Promise.resolve({}) },
     );
 
     expect(response.status).toBe(200);
-    await expect(response.json()).resolves.toEqual({ deleted: 3 });
+    await expect(response.json()).resolves.toEqual({
+      attachments: { deletedBlobs: 2, deletedTombstones: 1 },
+      deleted: 3,
+    });
+    expect(cleanupAttachmentsMock).toHaveBeenCalledOnce();
     expect(deleteExpiredMock).toHaveBeenCalledOnce();
   });
 
   it("rejects POST requests without the cron secret", async () => {
     const response = await POST(
-      new Request(
+      new NextRequest(
         "http://localhost:3000/api/cron/email-send-operation-retention",
         {
           method: "POST",
           body: JSON.stringify({ CRON_SECRET: "wrong-secret" }),
         },
       ),
+      { params: Promise.resolve({}) },
     );
 
     expect(response.status).toBe(401);
@@ -73,17 +90,22 @@ describe("email send operation retention cron route", () => {
 
   it("deletes expired operations for an authorized POST request", async () => {
     const response = await POST(
-      new Request(
+      new NextRequest(
         "http://localhost:3000/api/cron/email-send-operation-retention",
         {
           method: "POST",
           body: JSON.stringify({ CRON_SECRET: "cron-secret" }),
         },
       ),
+      { params: Promise.resolve({}) },
     );
 
     expect(response.status).toBe(200);
-    await expect(response.json()).resolves.toEqual({ deleted: 3 });
+    await expect(response.json()).resolves.toEqual({
+      attachments: { deletedBlobs: 2, deletedTombstones: 1 },
+      deleted: 3,
+    });
+    expect(cleanupAttachmentsMock).toHaveBeenCalledOnce();
     expect(deleteExpiredMock).toHaveBeenCalledOnce();
   });
 });

apps/web/app/api/cron/email-send-operation-retention/route.ts

@@ -1,6 +1,7 @@
 import { NextResponse } from "next/server";
 import { hasCronSecret, hasPostCronSecret } from "@/utils/cron";
 import { deleteExpiredEmailSendOperations } from "@/utils/email-send-operation-retention";
+import { cleanupEmailAttachmentStages } from "@/utils/email/email-attachment-staging";
 import { captureException } from "@/utils/error";
 import { type RequestWithLogger, withError } from "@/utils/middleware";
 
@@ -39,9 +40,11 @@ export const POST = withError(
 );
 
 async function runRetention(request: RequestWithLogger) {
+  const attachments = await cleanupEmailAttachmentStages();
   const deleted = await deleteExpiredEmailSendOperations();
   request.logger.info("Deleted expired email send operations", {
+    attachments,
     count: deleted,
   });
-  return NextResponse.json({ deleted });
+  return NextResponse.json({ attachments, deleted });
 }

apps/web/app/api/messages/send-attachments/complete/route.test.ts

@@ -0,0 +1,94 @@
+import { NextRequest } from "next/server";
+import { beforeEach, describe, expect, it, vi } from "vitest";
+import { POST } from "./route";
+
+const completeEmailAttachments = vi.hoisted(() => vi.fn());
+const stagingErrors = vi.hoisted(() => ({
+  Consumed: class EmailAttachmentStageConsumedError extends Error {},
+  Incomplete: class EmailAttachmentStageIncompleteError extends Error {},
+  Invalid: class EmailAttachmentStageInvalidError extends Error {},
+}));
+
+vi.mock("@/utils/email/email-attachment-staging", () => ({
+  completeEmailAttachments,
+  EmailAttachmentStageConsumedError: stagingErrors.Consumed,
+  EmailAttachmentStageIncompleteError: stagingErrors.Incomplete,
+  EmailAttachmentStageInvalidError: stagingErrors.Invalid,
+}));
+vi.mock("@/utils/middleware", () => ({
+  withEmailAccount:
+    (_name: string, handler: (request: MockedRequest) => Promise<Response>) =>
+    (request: NextRequest) =>
+      handler(
+        Object.assign(request, {
+          auth: { emailAccountId: "account-1" },
+        }) as MockedRequest,
+      ),
+}));
+
+type MockedRequest = NextRequest & { auth: { emailAccountId: string } };
+
+describe("POST /api/messages/send-attachments/complete", () => {
+  beforeEach(() => vi.clearAllMocks());
+
+  it("verifies opaque IDs inside the authenticated account scope", async () => {
+    completeEmailAttachments.mockResolvedValue({
+      attachments: [
+        { id: "attachment-1", stageId: "stage-1", status: "ready" },
+      ],
+    });
+    const input = {
+      mutationId: "41ec6d2b-d0e8-4f75-924a-f6f4e5bab4cf",
+      attachments: [{ id: "attachment-1", stageId: "stage-1" }],
+    };
+
+    const response = await POST(
+      new NextRequest(
+        "http://localhost/api/messages/send-attachments/complete",
+        {
+          method: "POST",
+          body: JSON.stringify(input),
+          headers: { "content-type": "application/json" },
+        },
+      ),
+      { params: Promise.resolve({}) },
+    );
+
+    await expect(response.json()).resolves.toEqual({
+      attachments: [
+        { id: "attachment-1", stageId: "stage-1", status: "ready" },
+      ],
+    });
+    expect(completeEmailAttachments).toHaveBeenCalledWith({
+      emailAccountId: "account-1",
+      input,
+    });
+  });
+
+  it.each([
+    [new stagingErrors.Incomplete("Upload incomplete"), 409],
+    [new stagingErrors.Consumed("Upload expired"), 410],
+    [new stagingErrors.Invalid("Upload invalid"), 410],
+  ])("maps staging lifecycle failures to a retryable status", async (error, status) => {
+    completeEmailAttachments.mockRejectedValue(error);
+
+    const response = await post();
+
+    expect(response.status).toBe(status);
+    await expect(response.json()).resolves.toEqual({ error: error.message });
+  });
+});
+
+function post() {
+  return POST(
+    new NextRequest("http://localhost/api/messages/send-attachments/complete", {
+      method: "POST",
+      body: JSON.stringify({
+        mutationId: "41ec6d2b-d0e8-4f75-924a-f6f4e5bab4cf",
+        attachments: [{ id: "attachment-1", stageId: "stage-1" }],
+      }),
+      headers: { "content-type": "application/json" },
+    }),
+    { params: Promise.resolve({}) },
+  );
+}

apps/web/app/api/messages/send-attachments/complete/route.ts

@@ -0,0 +1,34 @@
+import { NextResponse } from "next/server";
+import {
+  completeEmailAttachments,
+  EmailAttachmentStageConsumedError,
+  EmailAttachmentStageIncompleteError,
+  EmailAttachmentStageInvalidError,
+} from "@/utils/email/email-attachment-staging";
+import { completeEmailAttachmentsBody } from "@/utils/email/email-attachment-staging.validation";
+import { withEmailAccount } from "@/utils/middleware";
+
+export const POST = withEmailAccount(
+  "messages/send-attachments/complete",
+  async (request) => {
+    const input = completeEmailAttachmentsBody.parse(await request.json());
+    try {
+      const result = await completeEmailAttachments({
+        emailAccountId: request.auth.emailAccountId,
+        input,
+      });
+      return NextResponse.json(result);
+    } catch (error) {
+      if (error instanceof EmailAttachmentStageIncompleteError) {
+        return NextResponse.json({ error: error.message }, { status: 409 });
+      }
+      if (
+        error instanceof EmailAttachmentStageInvalidError ||
+        error instanceof EmailAttachmentStageConsumedError
+      ) {
+        return NextResponse.json({ error: error.message }, { status: 410 });
+      }
+      throw error;
+    }
+  },
+);

apps/web/app/api/messages/send-attachments/stage/route.test.ts

@@ -0,0 +1,78 @@
+import { NextRequest } from "next/server";
+import { beforeEach, describe, expect, it, vi } from "vitest";
+import { POST } from "./route";
+
+const staging = vi.hoisted(() => ({
+  stageEmailAttachments: vi.fn(),
+}));
+
+vi.mock("@/utils/email/email-attachment-staging", () => ({
+  ...staging,
+  EmailAttachmentStageConflictError: class EmailAttachmentStageConflictError extends Error {},
+  EmailAttachmentStageUnavailableError: class EmailAttachmentStageUnavailableError extends Error {},
+}));
+vi.mock("@/utils/middleware", () => ({
+  withEmailAccount:
+    (_name: string, handler: (request: MockedRequest) => Promise<Response>) =>
+    (request: NextRequest) =>
+      handler(
+        Object.assign(request, {
+          auth: { emailAccountId: "account-1" },
+        }) as MockedRequest,
+      ),
+}));
+
+type MockedRequest = NextRequest & { auth: { emailAccountId: string } };
+
+describe("POST /api/messages/send-attachments/stage", () => {
+  beforeEach(() => vi.clearAllMocks());
+
+  it("passes only the authenticated account context to staging", async () => {
+    staging.stageEmailAttachments.mockResolvedValue({ mode: "direct" });
+
+    const response = await post(stageBody());
+
+    await expect(response.json()).resolves.toEqual({ mode: "direct" });
+    expect(staging.stageEmailAttachments).toHaveBeenCalledWith({
+      emailAccountId: "account-1",
+      input: stageBody(),
+    });
+  });
+
+  it("rejects malformed aggregate metadata before creating intents", async () => {
+    const body = stageBody();
+    body.attachments[0].size = 20 * 1024 * 1024;
+
+    await expect(post(body)).rejects.toThrow(
+      "Attachments must be 10 MB or smaller.",
+    );
+    expect(staging.stageEmailAttachments).not.toHaveBeenCalled();
+  });
+});
+
+function post(body: unknown) {
+  return POST(
+    new NextRequest("http://localhost/api/messages/send-attachments/stage", {
+      method: "POST",
+      body: JSON.stringify(body),
+      headers: { "content-type": "application/json" },
+    }),
+    { params: Promise.resolve({}) },
+  );
+}
+
+function stageBody() {
+  return {
+    mutationId: "41ec6d2b-d0e8-4f75-924a-f6f4e5bab4cf",
+    queuedAt: 1_788_000_000_000,
+    attachments: [
+      {
+        id: "attachment-1",
+        filename: "notes.txt",
+        mimeType: "text/plain",
+        size: 5,
+        disposition: "attachment" as const,
+      },
+    ],
+  };
+}

apps/web/app/api/messages/send-attachments/stage/route.ts

@@ -0,0 +1,30 @@
+import { NextResponse } from "next/server";
+import {
+  EmailAttachmentStageConflictError,
+  EmailAttachmentStageUnavailableError,
+  stageEmailAttachments,
+} from "@/utils/email/email-attachment-staging";
+import { stageEmailAttachmentsBody } from "@/utils/email/email-attachment-staging.validation";
+import { withEmailAccount } from "@/utils/middleware";
+
+export const POST = withEmailAccount(
+  "messages/send-attachments/stage",
+  async (request) => {
+    const input = stageEmailAttachmentsBody.parse(await request.json());
+    try {
+      const result = await stageEmailAttachments({
+        emailAccountId: request.auth.emailAccountId,
+        input,
+      });
+      return NextResponse.json(result);
+    } catch (error) {
+      if (error instanceof EmailAttachmentStageUnavailableError) {
+        return NextResponse.json({ error: error.message }, { status: 503 });
+      }
+      if (error instanceof EmailAttachmentStageConflictError) {
+        return NextResponse.json({ error: error.message }, { status: 409 });
+      }
+      throw error;
+    }
+  },
+);