feat(jobs): transactional outbox — opt-in atomic event dispatch

config/config.json

@@ -198,6 +198,9 @@
     "dlq_metric_refresh_sec": "${JOBS_DLQ_METRIC_REFRESH_SEC:-10}",
     "queue_metric_refresh_sec": "${JOBS_QUEUE_METRIC_REFRESH_SEC:-10}"
   },
+  "outbox": {
+    "drain_interval_sec": "${OUTBOX_DRAIN_INTERVAL_SEC:-0}"
+  },
   "content": {
     "enabled": "${CONTENT_ENABLED:-false}"
   },

config/config.sample.json

@@ -198,6 +198,9 @@
     "dlq_metric_refresh_sec": "${JOBS_DLQ_METRIC_REFRESH_SEC:-10}",
     "queue_metric_refresh_sec": "${JOBS_QUEUE_METRIC_REFRESH_SEC:-10}"
   },
+  "outbox": {
+    "drain_interval_sec": "${OUTBOX_DRAIN_INTERVAL_SEC:-0}"
+  },
   "content": {
     "enabled": "${CONTENT_ENABLED:-false}"
   },

docs/CONFIG.md

@@ -211,6 +211,7 @@ For URL components: `REDIS_HOST`, `REDIS_PORT`.
 | `JOBS_VISIBILITY_TIMEOUT_SEC` | `jobs.visibility_timeout_sec` | int | `0` | Processing lease: a job whose worker dies is re-queued after this many seconds. `0` disables leases (legacy behaviour) |
 | `JOBS_DLQ_METRIC_REFRESH_SEC` | `jobs.dlq_metric_refresh_sec` | int | `10` | Exports `jobs_dlq_depth{type="..."}` plus an aggregate `type="_total"` |
 | `JOBS_QUEUE_METRIC_REFRESH_SEC` | `jobs.queue_metric_refresh_sec` | int | `10` | Same bookkeeping for the waiting queue: `jobs_queue_depth{type="..."}` plus `type="_total"` |
+| `OUTBOX_DRAIN_INTERVAL_SEC` | `outbox.drain_interval_sec` | int | `0` | Transactional outbox (`src/jobs/Outbox.hpp`): how often the API pod relays `outbox` table rows to the job queue. `0` (default) disables draining — the pattern is opt-in; rows written via `Outbox::enqueue` sit in Postgres until a deploy enables this. Needs `jobs.enabled=true`. |
 | `DB_REPLICA_LAG_METRIC_REFRESH_SEC` | `database.replica_lag_metric_refresh_sec` | int | `15` | Refresh interval for the `db_replica_lag_seconds` gauge. Only registered when read replicas are configured (primary has no replay timestamp). |
 
 ## Content

docs/CONVENTIONS.md

@@ -164,6 +164,14 @@ These don't fall out of reading the code; they were learned the hard way.
 17. **Multi-table test cleanup goes through `TestHelpers::wipe_app_data()`.** It is the one place that knows the FK order (users CASCADE before extra roles; posts/audit_log/used_tokens ride one TRUNCATE) and re-seeds migration 001's two roles. Don't hand-roll TRUNCATE/DELETE sequences in fixtures.
 18. **Never accumulate with a self-referencing upsert** (`INSERT ... ON CONFLICT (k) DO UPDATE SET x = t.x + EXCLUDED.x`) through `Database::execute_write`. In a downstream fork every second-or-later write computed as though the existing row's value were 0 — invisible for positive deltas, fatal for negative ones (`CHECK` violations on refunds). CI evidence proved the conflicting row WAS detected and visible earlier in the same transaction, yet the self-referencing SET still read it as absent; the root cause was never found (forensics: site fork, commit b676430). Safe idiom: `INSERT ... ON CONFLICT DO NOTHING` (materialize the row so it can be locked) → `SELECT ... FOR UPDATE` → compute the new total in C++ → plain `UPDATE`.
 19. **After-response side effects go through the 4-arg `with_repo_errors` overload** (email dispatch, enqueue, webhook fire): `after_fn` runs once the guarded block completed, OUTSIDE the catch ladder, so a throwing side effect can never fire `callback` a second time. Stash results into a `std::optional` captured by reference if `after_fn` needs them (learned downstream: a receipt-email dispatch inside the guarded lambda double-fired the callback).
+20. **Events that must not be lost go through the transactional outbox** (`src/jobs/Outbox.hpp`, migration 010, opt-in via `OUTBOX_DRAIN_INTERVAL_SEC`). The after-response hook above is best-effort BY DESIGN: a process dying between the DB commit and `Jobs::submit` loses the event, and a failed submit is only logged. Fine for a re-requestable confirm-email; not fine for money paths, where "wallet debited but the receipt/webhook never fired" is a truth defect someone repairs by hand. There, write the event in the SAME transaction as the ledger write — commit makes both durable, rollback erases both — and the periodic drain relays it to the job queue (at-least-once; handlers must tolerate a duplicate, which Jobs' retry/lease paths already require). A billing-fork receipt dispatch is three lines inside the existing wallet transaction:
+    ```cpp
+    // inside Database::execute_transaction([&](auto& txn) { ... ledger write ... 
+    if (result.credited)  // same dedupe gate BillingEmails::receipt documents
+        Jobs::Outbox::enqueue(txn, Email::SendEmail::kJobType,
+                              {{"to", user.email}, {"subject", subject}, {"text", text}});
+    ```
+    The template's own email flows deliberately stay on the after-response path (bit-for-bit unchanged behaviour) — the outbox is the upgrade a fork opts specific call sites into, not a global rewire.
 
 ### Frontend gotchas
 
@@ -193,5 +201,6 @@ The scaffolding scripts are the entry points:
 | Check spec ↔ code drift | `./scripts/check-openapi-drift.sh` |
 | Validate Helm render | `make helm-validate` |
 | Worked CRUD example | `docs/EXAMPLES.md` |
+| Must-not-lose event dispatch (outbox vs after-response) | `src/jobs/Outbox.hpp` (doc comment is the decision guide; gotcha 20 above has the worked example) |
 | Money / ledger / payment-provider reference | `src/billing/Wallet.hpp` (billing module: append-only ledger, idempotent capture/refund/spend — spend is reference-keyed via migration 009's partial unique index, no HTTP endpoint (charging is fork domain), integer-only money; best-effort emails in `src/email/BillingEmails.hpp`) |
 | ADRs / architecture decisions | `docs/adr/` |

docs/module-deps.txt

@@ -59,6 +59,7 @@ email -> repositories
 email -> security
 email -> utils
 jobs -> cache
+jobs -> database
 jobs -> email
 jobs -> observability
 jobs -> utils

migrations/010_outbox.sql

@@ -0,0 +1,38 @@
+-- Migration 010: transactional outbox — opt-in atomic event dispatch.
+--
+-- Applied in numeric order on boot. The MigrationRunner wraps this file in ONE
+-- transaction under an advisory lock — do NOT add BEGIN/COMMIT. Idempotent DDL.
+--
+-- The after-response hook (HandlerSupport.hpp 4-arg with_repo_errors) fires a
+-- side effect AFTER the DB commit: a process that dies between the commit and
+-- Jobs::submit loses the event forever. For events that must not be lost
+-- (money mail, webhooks about paid state), the row is written HERE, inside the
+-- SAME transaction as the domain write, and a periodic drain task relays it to
+-- the Redis job queue afterwards (src/jobs/Outbox.hpp). Commit → the event is
+-- durable; rollback → the event never existed. Delivery is at-least-once.
+--
+-- Columns:
+--   kind        the Jobs type Outbox::drain() will submit ("email.send",
+--               "webhook.deliver", any registered handler type)
+--   payload     the exact Jobs::submit payload, opaque to the outbox
+--   claimed_at  NULL = ready for drain; non-NULL = a drainer claimed it
+--               (re-claimable after Outbox::kStaleClaimSec so a drainer that
+--               died between claim and submit can't strand the row forever)
+--   attempts    failed submit attempts so far (grows without bound — retry
+--               forever is correct for "must not be lost"; alert on it)
+--   last_error  what the most recent failed submit threw
+CREATE TABLE IF NOT EXISTS outbox (
+    id          UUID        PRIMARY KEY DEFAULT gen_random_uuid(),
+    kind        TEXT        NOT NULL,
+    payload     JSONB       NOT NULL DEFAULT '{}'::jsonb,
+    created_at  TIMESTAMPTZ NOT NULL DEFAULT now(),
+    claimed_at  TIMESTAMPTZ,
+    attempts    INTEGER     NOT NULL DEFAULT 0,
+    last_error  TEXT        NOT NULL DEFAULT ''
+);
+
+-- Drain order is FIFO over the unclaimed backlog; the partial index keeps the
+-- claim query cheap no matter how many claimed-in-flight rows exist.
+CREATE INDEX IF NOT EXISTS idx_outbox_unclaimed
+    ON outbox (created_at)
+    WHERE claimed_at IS NULL;

src/core/Core.cpp

@@ -30,6 +30,7 @@
 #include "database/Migrations.hpp"
 #include "email/Mailer.hpp"
 #include "jobs/Jobs.hpp"
+#include "jobs/Outbox.hpp"
 #include "messaging/Messaging.hpp"
 #include "observability/Observability.hpp"
 #include "security/Auth.hpp"
@@ -599,6 +600,31 @@ void Application::init_jobs_(Config::AppConfig& cfg) {
     Jobs::get().set_visibility_timeout(visibility);
     register_dlq_metric_(cfg);
     register_queue_depth_metric_(cfg);
+    register_outbox_drain_(cfg);
+}
+
+void Application::register_outbox_drain_(Config::AppConfig& cfg) {
+    // Tasks is only up in server mode (worker_main runs its own loop), and the
+    // drain needs both stores: rows come from Postgres, jobs land in Redis.
+    if (!Tasks::is_initialized() || !Database::is_initialized() || !Jobs::is_initialized())
+        return;
+    const int interval_sec = cfg.get<int>("outbox.drain_interval_sec", "OUTBOX_DRAIN_INTERVAL_SEC", 0);
+    if (interval_sec <= 0)
+        return;  // opt-in: the outbox table sits inert until a deploy enables draining
+    spdlog::info("Transactional outbox drain enabled: every {}s", interval_sec);
+    Tasks::schedule_recurring("outbox_drain", std::chrono::seconds(interval_sec), [] {
+        if (!Database::is_initialized() || !Jobs::is_initialized())
+            return;
+        try {
+            auto stats = Jobs::Outbox::drain();
+            if (stats.failed > 0)
+                spdlog::warn("outbox drain: {} row(s) failed to submit (will retry)", stats.failed);
+        } catch (const std::exception& e) {
+            // DB hiccup — rows stay put, the next tick retries. Losing a tick
+            // never loses an event; that's the whole point of the table.
+            spdlog::warn("outbox drain failed: {}", e.what());
+        }
+    });
 }
 
 void Application::register_health_check(std::string name, HealthFn probe, bool critical) {

src/core/Core.hpp

@@ -152,6 +152,12 @@ class Application {
     // so without a reaper the table + its index grow monotonically.
     static void register_token_reaper_();
 
+    // Schedules the transactional-outbox drain (Jobs::Outbox::drain) every
+    // outbox.drain_interval_sec seconds. Opt-in: the default 0 schedules
+    // nothing, so the outbox table sits inert unless a deploy turns it on.
+    // Server-mode only (Tasks isn't initialized in the worker).
+    static void register_outbox_drain_(Config::AppConfig& cfg);
+
     // Registers the subsystem probes the template ships with. Services
     // that add their own modules can call Core::get().register_health_check
     // at any point after Core::initialize() returns.

src/jobs/Outbox.cpp

@@ -0,0 +1,105 @@
+/**
+ * @file Outbox.cpp
+ * @brief Body for src/jobs/Outbox.hpp — compiled once into app_core (ADR 0003
+ *        as amended 2026-08-22). The only jobs TU that includes the Database
+ *        layer (edge declared in docs/module-deps.txt).
+ */
+
+#include "jobs/Outbox.hpp"
+
+#include <exception>
+#include <string>
+#include <vector>
+
+#include <spdlog/spdlog.h>
+
+#include "database/Database.hpp"
+#include "jobs/Jobs.hpp"
+
+namespace Jobs {
+namespace Outbox {
+
+namespace {
+
+struct PendingRow {
+    std::string id;
+    std::string kind;
+    std::string payload;  // jsonb as text; parsed at submit time
+    int attempts = 0;
+};
+
+}  // namespace
+
+DrainStats drain(int batch, int reclaim_sec) {
+    DrainStats stats;
+    if (batch <= 0)
+        return stats;
+
+    // 1. Claim a batch in a SHORT transaction — no Redis I/O while holding row
+    //    locks. SKIP LOCKED makes concurrent drainers partition the backlog
+    //    instead of blocking or double-claiming; the committed claimed_at then
+    //    hides the rows from later passes until they're finalized (or the
+    //    claim goes stale — reclaim_sec — because this process died here).
+    auto rows = Database::get().execute_write([&](auto& txn) {
+        std::vector<PendingRow> out;
+        auto r = txn.exec_params(
+            "UPDATE outbox SET claimed_at = now() "
+            "WHERE id IN (SELECT id FROM outbox "
+            "             WHERE claimed_at IS NULL "
+            "                OR claimed_at < now() - make_interval(secs => $2::int) "
+            "             ORDER BY created_at "
+            "             LIMIT $1 "
+            "             FOR UPDATE SKIP LOCKED) "
+            "RETURNING id, kind, payload, attempts",
+            batch,
+            reclaim_sec);
+        out.reserve(r.size());
+        for (const auto& row : r) {
+            out.push_back({row["id"].template as<std::string>(),
+                           row["kind"].template as<std::string>(),
+                           row["payload"].template as<std::string>(),
+                           row["attempts"].template as<int>()});
+        }
+        return out;
+    });
+    if (rows.empty())
+        return stats;
+
+    // 2. Relay each row to the job queue. A per-row failure (Redis down, Jobs
+    //    not initialized, unparseable payload) is recorded, never thrown —
+    //    one poisoned row must not stall the rest of the batch.
+    std::vector<std::string> submitted_ids;
+    std::vector<std::pair<std::string, std::string>> failed;  // id -> error
+    for (const auto& row : rows) {
+        try {
+            Jobs::get().submit(row.kind, json::parse(row.payload));
+            submitted_ids.push_back(row.id);
+        } catch (const std::exception& e) {
+            spdlog::warn(
+                "outbox: submit of {} (kind={}, attempt {}) failed: {}", row.id, row.kind, row.attempts + 1, e.what());
+            failed.emplace_back(row.id, e.what());
+        }
+    }
+
+    // 3. Finalize: drop the relayed rows, release the failed ones for the next
+    //    pass. If THIS transaction is lost (process death, DB outage), the
+    //    claims go stale and every row — including already-submitted ones — is
+    //    redelivered after reclaim_sec: at-least-once, as documented.
+    Database::get().execute_write([&](auto& txn) {
+        for (const auto& id : submitted_ids)
+            txn.exec_params("DELETE FROM outbox WHERE id = $1", id);
+        for (const auto& [id, err] : failed)
+            txn.exec_params(
+                "UPDATE outbox SET claimed_at = NULL, attempts = attempts + 1, last_error = $2 WHERE id = $1", id, err);
+        return 0;
+    });
+
+    stats.submitted = static_cast<long>(submitted_ids.size());
+    stats.failed = static_cast<long>(failed.size());
+    if (stats.submitted > 0 || stats.failed > 0)
+        spdlog::debug("outbox: drained batch — {} submitted, {} failed", stats.submitted, stats.failed);
+    return stats;
+}
+
+}  // namespace Outbox
+}  // namespace Jobs

src/jobs/Outbox.hpp

@@ -0,0 +1,93 @@
+/**
+ * @file Outbox.hpp
+ * @brief Transactional outbox: atomic "domain write + event" dispatch to the
+ *        Jobs queue. Opt-in — nothing drains until `outbox.drain_interval_sec`
+ *        (OUTBOX_DRAIN_INTERVAL_SEC) is set > 0.
+ *
+ * ## When to use the outbox vs the after-response hook
+ *
+ * The default dispatch discipline is the after-response hook — the 4-arg
+ * `Api::with_repo_errors` overload (api/HandlerSupport.hpp): commit the DB
+ * write, send the response, then fire the side effect (Jobs::submit, email,
+ * webhook). That is best-effort BY DESIGN: a process that dies between the DB
+ * commit and the submit loses the event, and a failed submit is only logged.
+ * For a confirm-email link the user can re-request, that trade-off is right —
+ * don't add an outbox row, a table and a drain hop for events whose loss is
+ * an inconvenience.
+ *
+ * Use the outbox when losing the event corrupts truth someone relies on:
+ *   - money paths: a receipt/refund notice for a ledger write that DID happen
+ *     (src/email/BillingEmails.hpp discipline — the wallet write is durable,
+ *     so the notice about it must eventually go out too);
+ *   - webhooks that downstream systems reconcile against ("payment captured");
+ *   - any event where "DB committed but nobody was told" needs a human to
+ *     notice and repair by hand.
+ *
+ * Mechanics: `enqueue(txn, kind, payload)` INSERTs into the `outbox` table
+ * INSIDE the caller's open transaction — commit makes the event durable
+ * atomically with the domain write, rollback erases both. A periodic task
+ * (`Outbox::drain`, scheduled by Core when the interval is > 0) claims
+ * unclaimed rows (`FOR UPDATE SKIP LOCKED` — concurrent drainers never double-
+ * claim) and relays each to `Jobs::submit(kind, payload)`: the row is DELETEd
+ * on success, or released (attempts+1, last_error, claimed_at back to NULL)
+ * on failure so the next drain retries it. Delivery is therefore
+ * AT-LEAST-ONCE: a drainer dying between submit and delete redelivers after
+ * kStaleClaimSec — job handlers for outbox kinds must tolerate a duplicate
+ * (the same contract Jobs' retry/visibility-timeout paths already impose).
+ *
+ * Existing flows are NOT routed through here — the after-response paths keep
+ * their exact behaviour; the outbox is the opt-in upgrade for the call sites
+ * that need it. Worked example: docs/CONVENTIONS.md gotcha 20.
+ */
+
+#pragma once
+
+#include <string>
+
+#include "jobs/Job.hpp"
+
+namespace Jobs {
+namespace Outbox {
+
+/// A claim older than this is considered abandoned (the drainer died between
+/// claiming and finalizing) and becomes drainable again — the self-healing
+/// half of the at-least-once contract.
+inline constexpr int kStaleClaimSec = 300;
+
+/**
+ * @brief Record an event in the SAME transaction as the caller's domain
+ *        write. @p txn is the `auto& txn` every repository lambda receives
+ *        from Database::execute_write / execute_transaction
+ *        (Database::detail::TracingTxn — only exec/exec_params are used).
+ *        @p kind must be a job type a worker handles (jobs/BuiltinHandlers.cpp
+ *        or a fork-registered handler); @p payload is passed to Jobs::submit
+ *        verbatim. Throws on SQL failure — which rolls the caller's
+ *        transaction back, exactly the atomicity the pattern promises.
+ */
+template <typename Txn>
+void enqueue(Txn& txn, const std::string& kind, const json& payload) {
+    txn.exec_params("INSERT INTO outbox (kind, payload) VALUES ($1, $2::jsonb)", kind, payload.dump());
+}
+
+/// What one drain pass did — returned for tests/observability; the periodic
+/// task just logs it.
+struct DrainStats {
+    long submitted = 0;  ///< rows relayed to Jobs::submit and deleted
+    long failed = 0;     ///< rows whose submit threw (attempts bumped, retried next pass)
+};
+
+/**
+ * @brief Relay up to @p batch pending outbox rows to the Jobs queue.
+ *        Claim (UPDATE ... WHERE claimed_at IS NULL ... FOR UPDATE SKIP
+ *        LOCKED RETURNING) → Jobs::submit per row → one finalize transaction
+ *        (DELETE the submitted, release the failed). Safe to call from
+ *        several processes at once. Throws only when the claim/finalize
+ *        transactions themselves fail (DB down) — per-row submit failures are
+ *        recorded on the row, never thrown.
+ * @param reclaim_sec claims older than this are treated as abandoned and
+ *        re-claimed (defaults to kStaleClaimSec).
+ */
+DrainStats drain(int batch = 100, int reclaim_sec = kStaleClaimSec);
+
+}  // namespace Outbox
+}  // namespace Jobs