feat(outbox): archive old outbox events to cold storage

Cargo.toml

@@ -17,7 +17,7 @@ cala-ledger = { path = "cala-ledger", version = "0.20.2-dev" }
 cel = "0.14.1"
 es-entity = "0.11.11"
 job = { version = "0.6.36", features = ["es-entity"] }
-obix = { version = "0.5.0", default-features = false }
+obix = { git = "https://github.com/GaloyMoney/obix.git", branch = "feat-event-archive", default-features = false } # git dependency until GaloyMoney/obix#105 is released
 
 anyhow = "1.0.99"
 cached = { version = "2.0", features = ["async"] }

cala-ledger/migrations/20251204130226_cala_obix_setup.sql

@@ -8,6 +8,21 @@ CREATE TABLE cala_persistent_outbox_events (
   seen_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
 );
 
+-- Archive manifest: one row per exported JSONL chunk of pruned
+-- persistent outbox events. Chunks are contiguous — the next chunk starts
+-- at max_sequence + 1 of the previous one.
+-- Any grouping label (e.g. a calendar date) is encoded in a chunk's
+-- path; grouping semantics belong to the deployment, not to obix.
+CREATE TABLE cala_persistent_outbox_archive_chunks (
+  path TEXT PRIMARY KEY,
+  min_sequence BIGINT NOT NULL,
+  max_sequence BIGINT NOT NULL,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+CREATE INDEX cala_idx_persistent_outbox_archive_chunks_max_sequence
+  ON cala_persistent_outbox_archive_chunks (max_sequence);
+
 -- Ephemeral outbox events
 CREATE TABLE cala_ephemeral_outbox_events (
   event_type VARCHAR NOT NULL UNIQUE,

cala-ledger/src/ledger/config.rs

@@ -1,6 +1,8 @@
 use derive_builder::Builder;
 use es_entity::clock::{Clock, ClockHandle};
 
+use crate::outbox::OutboxArchiveConfig;
+
 #[derive(Builder, Clone, Debug)]
 #[builder(build_fn(validate = "Self::validate"))]
 pub struct CalaLedgerConfig {
@@ -14,6 +16,12 @@ pub struct CalaLedgerConfig {
     pub(super) pool: Option<sqlx::PgPool>,
     #[builder(setter(into), default = "Clock::handle().clone()")]
     pub(super) clock: ClockHandle,
+    /// Cold-storage archiving of old outbox events. When set, settled
+    /// history is swept out of postgres by the archiver job (see
+    /// [`CalaLedger::register_outbox_archiver`](crate::ledger::CalaLedger::register_outbox_archiver))
+    /// and pre-watermark reads fall back to the archive.
+    #[builder(setter(strip_option), default)]
+    pub(super) outbox_archive: Option<OutboxArchiveConfig>,
 }
 
 impl CalaLedgerConfig {

cala-ledger/src/ledger/mod.rs

@@ -63,7 +63,8 @@ impl CalaLedger {
         }
 
         let clock = config.clock;
-        let publisher = OutboxPublisher::init(&pool, &clock).await?;
+        let publisher =
+            OutboxPublisher::init(&pool, &clock, config.outbox_archive.as_ref()).await?;
         let accounts = Accounts::new(&pool, &publisher, &clock);
         let journals = Journals::new(&pool, &publisher, &clock);
         let tx_templates = TxTemplates::new(&pool, &publisher, &clock);
@@ -223,6 +224,24 @@ impl CalaLedger {
         self.publisher.inner()
     }
 
+    /// Register the job sweeping settled spans of outbox history to the
+    /// configured archive storage (one span per run, rescheduling until
+    /// caught up). Requires `outbox_archive` to have been set at
+    /// [`init`](Self::init) — fails with `obix::ArchiveError::NotConfigured`
+    /// otherwise.
+    pub async fn register_outbox_archiver(
+        &self,
+        jobs: &mut job::Jobs,
+    ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+        self.publisher
+            .inner()
+            .register_event_archiver(
+                jobs,
+                obix::OutboxArchiverJobConfig::new(job::JobType::new("cala.outbox.archiver")),
+            )
+            .await
+    }
+
     pub fn register_outbox_listener(
         &self,
         start_after: Option<obix::EventSequence>,

cala-ledger/src/outbox/config.rs

@@ -0,0 +1,69 @@
+use es_entity::clock::ClockHandle;
+
+use std::sync::Arc;
+
+use super::CalaMailboxTables;
+
+pub const DEFAULT_OUTBOX_ARCHIVE_RETENTION_DAYS: u32 = 3;
+pub const DEFAULT_OUTBOX_ARCHIVE_PATH_PREFIX: &str = "outbox-archive/cala/";
+
+/// Cold-storage archiving of old outbox events (obix archive). Archived
+/// history is swept to object storage and pruned from postgres; reads of
+/// pre-watermark events transparently fall back to the archive.
+///
+/// The object-storage backend is supplied by the consumer via
+/// [`obix::EventArchiveStorage`] (GCS, S3, local filesystem, ...);
+/// [`obix::InMemoryArchiveStorage`] works for tests.
+#[derive(Clone)]
+pub struct OutboxArchiveConfig {
+    /// The object-storage backend chunks are written to / read from.
+    pub storage: Arc<dyn obix::EventArchiveStorage>,
+    /// Days of history kept in postgres; older, fully-elapsed days are
+    /// swept to storage one day per archiver run.
+    pub retention_days: u32,
+    /// Prepended to every chunk path, e.g. `"outbox-archive/cala/"`.
+    pub path_prefix: String,
+}
+
+impl OutboxArchiveConfig {
+    pub fn new(storage: Arc<dyn obix::EventArchiveStorage>) -> Self {
+        Self {
+            storage,
+            retention_days: DEFAULT_OUTBOX_ARCHIVE_RETENTION_DAYS,
+            path_prefix: DEFAULT_OUTBOX_ARCHIVE_PATH_PREFIX.to_string(),
+        }
+    }
+
+    pub fn with_retention_days(mut self, retention_days: u32) -> Self {
+        self.retention_days = retention_days;
+        self
+    }
+
+    pub fn with_path_prefix(mut self, prefix: impl Into<String>) -> Self {
+        self.path_prefix = prefix.into();
+        self
+    }
+
+    pub(super) fn build(&self, pool: &sqlx::PgPool, clock: &ClockHandle) -> obix::ArchiveConfig {
+        obix::ArchiveConfig::new(
+            self.storage.clone(),
+            Arc::new(obix::DailyRetentionBoundary::<CalaMailboxTables>::new(
+                pool,
+                chrono::Duration::days(i64::from(self.retention_days)),
+                clock.clone(),
+            )),
+        )
+        .with_path_prefix(self.path_prefix.clone())
+        .with_clock(clock.clone())
+    }
+}
+
+impl std::fmt::Debug for OutboxArchiveConfig {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("OutboxArchiveConfig")
+            .field("storage", &"<dyn EventArchiveStorage>")
+            .field("retention_days", &self.retention_days)
+            .field("path_prefix", &self.path_prefix)
+            .finish()
+    }
+}

cala-ledger/src/outbox/mod.rs

@@ -1,9 +1,11 @@
+mod config;
 mod publisher;
 
 mod event {
     pub use cala_types::outbox::*;
 }
 
+pub use config::OutboxArchiveConfig;
 pub use event::*;
 pub use publisher::OutboxPublisher;
 

cala-ledger/src/outbox/publisher.rs

@@ -1,19 +1,24 @@
 use cala_types::outbox::OutboxEventPayload;
 use es_entity::clock::ClockHandle;
 
-use super::ObixOutbox;
+use super::{ObixOutbox, OutboxArchiveConfig};
 
 #[derive(Debug, Clone)]
 pub struct OutboxPublisher {
     inner: ObixOutbox,
 }
 
 impl OutboxPublisher {
-    pub async fn init(pool: &sqlx::PgPool, clock: &ClockHandle) -> Result<Self, sqlx::Error> {
+    pub async fn init(
+        pool: &sqlx::PgPool,
+        clock: &ClockHandle,
+        archive: Option<&OutboxArchiveConfig>,
+    ) -> Result<Self, sqlx::Error> {
         let config = obix::MailboxConfig::builder()
             .clock(clock.clone())
             .event_buffer_size(50_000)
             .event_cache_size(10_000)
+            .archive(archive.map(|a| a.build(pool, clock)))
             .build()
             .expect("MailboxConfig");
         let outbox = ObixOutbox::init(pool, config).await?;