fix(ci): run migrations in the API deploy, before the code that needs them (Sentry #7645807976)

.github/workflows/deploy-api.yml

@@ -71,6 +71,31 @@ jobs:
       - name: Remove system Pulumi (SST compatibility fix)
         run: sudo rm -f /usr/local/bin/pulumi-language-nodejs
 
+      # Migrate BEFORE deploying, not after. `sst deploy` publishes each
+      # Lambda's new code the moment that function updates — well before the
+      # command returns — so migrating afterwards leaves a window where new code
+      # runs against the old schema. Any read or write touching a column the
+      # migration has not added yet throws, and for the queue workers that means
+      # burned SQS retries and dead-lettered chunks on a live broadcast.
+      #
+      # Until now this step did not exist at all and every schema change relied
+      # on someone running db:migrate by hand: batch_send.paused_at shipped in
+      # the same commit as the broadcast-reaper query that reads it, nobody ran
+      # it, and the reaper threw on every run against production.
+      #
+      # Safe because migrations are expected to be backward compatible: the
+      # still-running old code ignores columns it does not know about. A
+      # migration that REMOVES or renames still has to be split expand/contract
+      # across two releases — no ordering makes that safe on its own.
+      #
+      # Failing this way round is also the better failure: a migration that dies
+      # leaves old code on the old schema, which works. The other order leaves
+      # new code on the old schema, which does not.
+      - name: Run database migrations
+        run: pnpm --filter @wraps/db db:migrate
+        env:
+          DATABASE_URL: ${{ secrets.DATABASE_URL }}
+
       - name: Deploy SST
         run: pnpm exec sst deploy --stage ${{ github.event.inputs.stage || 'production' }}
         env:

apps/api/src/__tests__/batch-sender-claim-db.test.ts

@@ -68,6 +68,9 @@
 /**
  * Run the re-claim UPDATE (failed rows + stale queued rows).
  * Returns the contactIds that were re-claimed.
+ *
+ * This helper matches the FIXED implementation in batch-sender.ts
+ * (with messageId guard for queued rows).
  */
 async function runReclaim(
   orgId: string,
@@ -83,9 +86,17 @@
         eq(messageSend.batchSendId, BATCH_ID),
         inArray(messageSend.contactId, notClaimedIds),
         or(
-          eq(messageSend.status, "failed"),
+          // Only genuinely-unsent failures: a 'failed' row carrying a
+          // messageId was accepted by SES (e.g. a bookkeeping error was
+          // misfiled as a send failure) — re-claiming it would send a
+          // duplicate, and SES has no idempotency token to stop it.
+          and(
+            eq(messageSend.status, "failed"),
+            sql`${messageSend.messageId} IS NULL`
+          ),
           and(
             eq(messageSend.status, "queued"),
+            sql`${messageSend.messageId} IS NULL`,
             sql`${messageSend.claimedAt} < now() - interval '${sql.raw(String(CLAIM_STALE_MINUTES))} minutes'`
           )
         )
@@ -196,7 +207,7 @@
     const awsAccountId = fixture.ids.awsAccount;
 
     // Seed a 'sent' row (simulates a successfully processed chunk)
     await db.insert(messageSend).values({
       organizationId: orgId,
       contactId: CONTACT_A_ID,
       awsAccountId,
@@ -254,7 +265,7 @@
 
     // Re-claim UPDATE picks it up
     const reclaimed = await runReclaim(orgId, [CONTACT_A_ID]);
     expect(reclaimed).toHaveLength(1);
     expect(reclaimed[0].contactId).toBe(CONTACT_A_ID);
 
     // Verify the row was reset
@@ -319,4 +330,53 @@
     const reclaimed2 = await runReclaim(orgId, [CONTACT_A_ID]);
     expect(reclaimed2).toHaveLength(0);
   });
+
+  // ───────────────────────────────────────────────────────────────────────────
+  // 5. Stale queued row with messageId must NOT be re-claimed (duplicate-send guard)
+  // ───────────────────────────────────────────────────────────────────────────
+
+  it("stale queued row with messageId is NOT re-claimed (prevents duplicate sends)", async () => {
+    const orgId = fixture.ids.org;
+    const awsAccountId = fixture.ids.awsAccount;
+
+    const staleTime = new Date(Date.now() - 20 * 60 * 1000); // 20 minutes ago
+
+    // Seed a stale queued row WITH a messageId (simulates bookkeeping failure:
+    // SES accepted the message, but recordAcceptedSend() failed before writing messageId)
+    await db.insert(messageSend).values({
+      organizationId: orgId,
+      contactId: CONTACT_A_ID,
+      awsAccountId,
+      channel: "email",
+      batchSendId: BATCH_ID,
+      sourceType: "batch",
+      recipient: `${TEST_PREFIX}-ca@example.com`,
+      status: "queued",
+      claimedAt: staleTime,
+      messageId: "ses-message-123", // Row has a messageId = SES already accepted it
+    } as typeof messageSend.$inferInsert);
+
+    // Re-claim should NOT pick up this row (messageId guard prevents duplicate send)
+    const reclaimed = await runReclaim(orgId, [CONTACT_A_ID]);
+    expect(reclaimed).toHaveLength(0);
+
+    // Verify the row remains unchanged
+    const [row] = await db
+      .select({
+        status: messageSend.status,
+        messageId: messageSend.messageId,
+        claimedAt: messageSend.claimedAt,
+      })
+      .from(messageSend)
+      .where(
+        and(
+          eq(messageSend.organizationId, orgId),
+          eq(messageSend.batchSendId, BATCH_ID),
+          eq(messageSend.contactId, CONTACT_A_ID)
+        )
+      );
+    expect(row.status).toBe("queued");
+    expect(row.messageId).toBe("ses-message-123");
+    expect(row.claimedAt?.getTime()).toBe(staleTime.getTime());
+  });
 });

apps/api/src/workers/batch-sender.ts

@@ -1313,6 +1313,7 @@ async function processJob(
             ),
             and(
               eq(messageSend.status, "queued"),
+              isNull(messageSend.messageId),
               sql`${messageSend.claimedAt} < now() - interval '${sql.raw(String(CLAIM_STALE_MINUTES))} minutes'`
             )
           )

baseline/deploy-api-workflow.test.ts

@@ -0,0 +1,60 @@
+import { readFileSync } from "node:fs";
+import { describe, expect, it } from "vitest";
+
+// The hosted API deploy had no migration step at all, so every schema change
+// reached production only if someone remembered to run `db:migrate` by hand.
+// batch_send.paused_at is what happened when nobody did: the column and the
+// broadcast-reaper query that reads it shipped in the same commit, the Lambda
+// went live, and the reaper threw `column batch_send.paused_at does not exist`
+// on every run until the migration was applied out of band.
+//
+// scripts/selfhost/upgrade.ts already learned this and migrates before
+// deploying; this file holds the same guarantee for the hosted pipeline.
+const workflowSource = readFileSync(
+  new URL("../.github/workflows/deploy-api.yml", import.meta.url),
+  "utf-8"
+);
+
+// Anchored to the `run:` commands, not bare mentions: both strings also appear
+// in the prose explaining why the order matters, and a comment sitting above
+// the migrate step would otherwise register as the deploy happening first.
+const migrateIndex = workflowSource.search(/^\s+run: .*db:migrate/m);
+const deployIndex = workflowSource.search(/^\s+run: .*sst deploy/m);
+
+// Scoped to the migrate step alone: DATABASE_URL is also mapped into the deploy
+// step, which would satisfy a file-wide match while leaving migrate with none.
+const MIGRATE_STEP =
+  /- name: Run database migrations[\s\S]*?(?=\n {6}- name: )/;
+const DATABASE_URL_FROM_SECRETS = /^\s+DATABASE_URL: \$\{\{ secrets\./m;
+
+describe(".github/workflows/deploy-api.yml", () => {
+  it("runs database migrations as part of the deploy", () => {
+    expect(migrateIndex).toBeGreaterThan(-1);
+  });
+
+  it("runs migrations BEFORE deploying new code", () => {
+    // `sst deploy` publishes each Lambda's new code as that function updates,
+    // well before the command returns, so migrating afterwards still leaves a
+    // window where new code runs against the old schema. Order is the whole
+    // guarantee, so assert the order rather than only that both steps exist.
+    //
+    // This order also fails the better way round: a migration that dies leaves
+    // the old code on the old schema, which works. The other order leaves new
+    // code on the old schema, which does not.
+    // Both bounds asserted: a missing migrate step leaves migrateIndex at -1,
+    // which would sit "before" the deploy and pass this on a vacuous truth.
+    expect(migrateIndex).toBeGreaterThan(-1);
+    expect(deployIndex).toBeGreaterThan(-1);
+    expect(migrateIndex).toBeLessThan(deployIndex);
+  });
+
+  it("gives the migration step a database to connect to", () => {
+    // drizzle-kit reads DATABASE_URL from the environment: its dotenv call
+    // targets apps/web/.env.local, which does not exist on a CI runner, and
+    // dotenv does not override an already-set variable anyway. A migrate step
+    // without this mapping connects to an empty URL and fails the deploy.
+    const migrateStep = workflowSource.match(MIGRATE_STEP)?.[0];
+    expect(migrateStep).toBeDefined();
+    expect(migrateStep).toMatch(DATABASE_URL_FROM_SECRETS);
+  });
+});