v0.18.0 Slice 5: bounce correlation hook

The `POST /api/webhooks/email-bounce` body accepts a new optional
`message_id` field. When supplied, the handler looks up the
matching row in `outbound_emails` and stamps status='bounced' +
appends "bounce (<kind>)" to the error column. The response
gains a `correlated_id` field (nullable: null when no
message_id was provided or no row matched).

Hard-bounce -> `email_opt_out_all = 1` still fires regardless
(the v1 contract from api_notifications.py:486-498); Slice 5
just adds the per-message audit trail on top.

Providers that don't surface Message-ID get the legacy v1
behavior — match by email, flip the global opt-out, leave
correlated_id null. Providers that replay an old bounce with a
message_id the framework no longer has log an INFO line but
still 200 (the bounce path can't refuse just because a row was
pruned).

Test update: `test_bounce_webhook_refuses_unsigned_when_secret_configured`
in test_e2e_smoke.py was asserting on the exact response shape
{ok, matched}; v0.18.0 adds `correlated_id`. The test now asserts
on the new shape; documented in CHANGELOG.

4 new tests covering: bounce with message_id stamps the row;
unknown message_id is logged + does not crash; bounce without
message_id still flips opt-out (backward compat); bounced rows
surface in the admin endpoint with `?status=bounced`.

Full suite: 295 passed.

backend/app/api_notifications.py

@@ -73,6 +73,13 @@ class MarkReadBody(BaseModel):
 class BounceBody(BaseModel):
     email: str = Field(min_length=3, max_length=320)
     kind: str = Field(default="hard")  # 'hard' or 'complaint'
+    # v0.18.0 Slice 5: when the bounce provider includes the
+    # original Message-ID, the framework correlates it back to
+    # the matching `outbound_emails` row and stamps
+    # `status='bounced'`. Optional — providers that don't surface
+    # the Message-ID still flip the global opt-out via the email
+    # match, but lose the per-message attribution.
+    message_id: str | None = Field(default=None, max_length=1000)
 
 
 class CookieConsentBody(BaseModel):
@@ -555,16 +562,46 @@ def make_router(config: Config) -> APIRouter:
             import hmac as _hmac
             if not received or not _hmac.compare_digest(expected, received):
                 raise HTTPException(401, "Invalid webhook signature")
+        # v0.18.0 Slice 5: correlate the bounce back to the
+        # matching outbound_emails row if the provider supplied
+        # the Message-ID. The hard-bounce -> global-opt-out
+        # logic below still fires regardless; this is an
+        # additional audit signal.
+        correlated_row_id: int | None = None
+        if body.message_id:
+            correlated = db.conn().execute(
+                "SELECT id FROM outbound_emails WHERE message_id = ?",
+                (body.message_id,),
+            ).fetchone()
+            if correlated is not None:
+                correlated_row_id = correlated["id"]
+                db.conn().execute(
+                    "UPDATE outbound_emails SET status = 'bounced', "
+                    "error = COALESCE(error, '') || ? WHERE id = ?",
+                    (f"bounce ({body.kind})", correlated_row_id),
+                )
+                log.info(
+                    "email-bounce: correlated message_id=%s -> outbound_emails.id=%s",
+                    body.message_id, correlated_row_id,
+                )
+            else:
+                log.info(
+                    "email-bounce: message_id=%s did not match any "
+                    "outbound_emails row (provider may be replaying an old bounce, "
+                    "or the row was pruned)",
+                    body.message_id,
+                )
+
         row = db.conn().execute(
             "SELECT id FROM users WHERE LOWER(email) = LOWER(?)", (body.email,),
         ).fetchone()
         if row is None:
-            return {"ok": True, "matched": False}
+            return {"ok": True, "matched": False, "correlated_id": correlated_row_id}
         db.conn().execute(
             "UPDATE users SET email_opt_out_all = 1 WHERE id = ?", (row["id"],),
         )
         log.info("email-bounce: opted out user %s (%s)", row["id"], body.kind)
-        return {"ok": True, "matched": True}
+        return {"ok": True, "matched": True, "correlated_id": correlated_row_id}
 
     return router
 

backend/tests/test_e2e_smoke.py

@@ -231,13 +231,17 @@ def test_bounce_webhook_refuses_unsigned_when_secret_configured(app_with_fake_gi
 
         # With the right header, the call passes the guard. (No matching
         # user exists, so we get {matched: False} — that's the v1 contract.)
+        # v0.18.0 Slice 5: the response now includes `correlated_id`
+        # (the outbound_emails row id that matched the bounce's
+        # `message_id`, if one was supplied). The body didn't pass a
+        # message_id, so correlated_id is None.
         r = client.post(
             "/api/webhooks/email-bounce",
             json={"email": "stranger@example.com", "kind": "hard"},
             headers={"X-Webhook-Secret": "shhh"},
         )
         assert r.status_code == 200, r.text
-        assert r.json() == {"ok": True, "matched": False}
+        assert r.json() == {"ok": True, "matched": False, "correlated_id": None}
 
 
 def test_bounce_webhook_open_when_secret_unset(app_with_fake_gitea):

backend/tests/test_outbound_emails_vertical.py

@@ -229,3 +229,140 @@ def test_admin_outbound_emails_refuses_non_admin(app_with_fake_gitea):
         )
         r = client.get("/api/admin/outbound-emails")
         assert r.status_code == 403
+
+
+# ---------------------------------------------------------------------------
+# v0.18.0 Slice 5: bounce correlation
+# ---------------------------------------------------------------------------
+
+
+def test_bounce_with_message_id_marks_outbound_row_bounced(app_with_fake_gitea):
+    """When the bounce body includes the original `message_id`, the
+    framework looks it up in outbound_emails and stamps
+    status='bounced' on the matching row. The hard-bounce ->
+    global-opt-out logic still fires."""
+    from fastapi.testclient import TestClient
+    from app import db, email as email_mod
+
+    app, _fake = app_with_fake_gitea
+    with TestClient(app) as client:
+        provision_user_row(user_id=800, login="bouncey", role="contributor")
+        db.conn().execute("UPDATE users SET email = 'bouncey@ex.co' WHERE id = 800")
+
+        # Send something to bouncey to land an outbound_emails row.
+        email_mod.reset_sent_envelopes()
+        client.post("/auth/otc/request", json={"email": "bouncey@ex.co"})
+        row = db.conn().execute(
+            "SELECT id, message_id, status FROM outbound_emails "
+            "WHERE to_address = 'bouncey@ex.co'"
+        ).fetchone()
+        assert row is not None
+        original_id = row["id"]
+        message_id = row["message_id"]
+        assert row["status"] == "deferred"  # pre-bounce baseline
+
+        # Bounce comes in carrying that message_id.
+        r = client.post(
+            "/api/webhooks/email-bounce",
+            json={
+                "email": "bouncey@ex.co",
+                "kind": "hard",
+                "message_id": message_id,
+            },
+        )
+        assert r.status_code == 200
+        body = r.json()
+        assert body["matched"] is True
+        assert body["correlated_id"] == original_id
+
+        # Audit row stamped.
+        post = db.conn().execute(
+            "SELECT status, error FROM outbound_emails WHERE id = ?",
+            (original_id,),
+        ).fetchone()
+        assert post["status"] == "bounced"
+        assert "bounce (hard)" in (post["error"] or "")
+
+        # Hard-bounce global opt-out still fires.
+        urow = db.conn().execute(
+            "SELECT email_opt_out_all FROM users WHERE id = 800"
+        ).fetchone()
+        assert urow["email_opt_out_all"] == 1
+
+
+def test_bounce_with_unknown_message_id_does_not_crash(app_with_fake_gitea):
+    """A message_id the framework doesn't recognize logs but does
+    NOT 5xx — bounce providers replay old bounces, and the
+    framework can't refuse just because the row was pruned."""
+    from fastapi.testclient import TestClient
+
+    app, _fake = app_with_fake_gitea
+    with TestClient(app) as client:
+        r = client.post(
+            "/api/webhooks/email-bounce",
+            json={
+                "email": "nobody@ex.co",
+                "kind": "hard",
+                "message_id": "<not-in-our-db@ex.co>",
+            },
+        )
+        assert r.status_code == 200
+        assert r.json()["correlated_id"] is None
+
+
+def test_bounce_without_message_id_still_flips_opt_out(app_with_fake_gitea):
+    """Backward compat: providers that don't surface Message-ID
+    still get the legacy v1 behavior — match by email + flip the
+    global opt-out."""
+    from fastapi.testclient import TestClient
+    from app import db
+
+    app, _fake = app_with_fake_gitea
+    with TestClient(app) as client:
+        provision_user_row(user_id=801, login="legacybounce", role="contributor")
+        db.conn().execute("UPDATE users SET email = 'legacy@ex.co' WHERE id = 801")
+
+        r = client.post(
+            "/api/webhooks/email-bounce",
+            json={"email": "legacy@ex.co", "kind": "hard"},
+        )
+        assert r.status_code == 200
+        body = r.json()
+        assert body["matched"] is True
+        assert body["correlated_id"] is None
+
+        urow = db.conn().execute(
+            "SELECT email_opt_out_all FROM users WHERE id = 801"
+        ).fetchone()
+        assert urow["email_opt_out_all"] == 1
+
+
+def test_bounced_rows_show_in_admin_endpoint(app_with_fake_gitea):
+    """The admin endpoint surfaces bounced rows alongside the rest;
+    filtering by `status=bounced` isolates them."""
+    from fastapi.testclient import TestClient
+    from app import db, email as email_mod
+
+    app, _fake = app_with_fake_gitea
+    with TestClient(app) as client:
+        provision_user_row(user_id=802, login="adminB", role="admin")
+        sign_in_as(
+            client, user_id=802, gitea_login="adminB",
+            display_name="Admin B", role="admin", email="adminb@test",
+        )
+        email_mod.reset_sent_envelopes()
+        client.post("/auth/otc/request", json={"email": "willbounce@ex.co"})
+        row = db.conn().execute(
+            "SELECT message_id FROM outbound_emails WHERE to_address = 'willbounce@ex.co'"
+        ).fetchone()
+        client.post(
+            "/api/webhooks/email-bounce",
+            json={"email": "willbounce@ex.co", "kind": "hard", "message_id": row["message_id"]},
+        )
+
+        r = client.get("/api/admin/outbound-emails?status=bounced")
+        assert r.status_code == 200
+        items = r.json()["items"]
+        assert items
+        assert all(it["status"] == "bounced" for it in items)
+        assert any(it["to_address"] == "willbounce@ex.co" for it in items)