camel_core/lifecycle/adapters/

redb_journal.rs

//! Redb-backed runtime event journal.
//!
//! # Schema
//!
//! - Table `events`: `u64 → &[u8]`  (seq → serde_json bytes of `JournalEntry`)
//! - Table `command_ids`: `&str → ()`  (presence = alive)
//!
//! # Sequence numbers
//!
//! No autoincrement in redb. Each `append_batch` derives `next_seq` by reading
//! the last key via `iter().next_back()` and adding 1; defaults to 0 if empty.

use std::path::PathBuf;
use std::sync::Arc;

use async_trait::async_trait;
use redb::{
    Database, Durability, ReadableDatabase, ReadableTable, ReadableTableMetadata, TableDefinition,
};
use serde::{Deserialize, Serialize};

use camel_api::CamelError;

use crate::lifecycle::adapters::runtime_event_record::runtime_event_serde;
use crate::lifecycle::application::ports::RuntimeEventJournalPort;
use crate::lifecycle::domain::{DomainError, RuntimeEvent};

// ── Table definitions ─────────────────────────────────────────────────────────

const EVENTS_TABLE: TableDefinition<u64, &[u8]> = TableDefinition::new("events");
const COMMAND_IDS_TABLE: TableDefinition<&str, ()> = TableDefinition::new("command_ids");

/// Max attempts to acquire the redb file lock on open. Covers the Kubernetes
/// ReadWriteOnce PVC handover window where a terminating pod has not yet
/// released the lock when the replacement pod starts.
const OPEN_LOCK_MAX_ATTEMPTS: u32 = 40;
/// Delay between open attempts (40 × 250ms ≈ 10s total handover budget).
const OPEN_LOCK_RETRY_BACKOFF: std::time::Duration = std::time::Duration::from_millis(250);

/// True when a redb open error is the "another handle holds the lock" case,
/// which is transient during pod handover and worth retrying. Matched on the
/// Display text to stay independent of the redb error-enum shape across versions.
fn is_lock_contention(err: &redb::DatabaseError) -> bool {
    matches!(err, redb::DatabaseError::DatabaseAlreadyOpen) || {
        let msg = err.to_string().to_ascii_lowercase();
        msg.contains("already open") || msg.contains("acquire lock")
    }
}

// ── Public types ──────────────────────────────────────────────────────────────

/// Durability mode for journal writes.
#[derive(Debug, Clone, PartialEq, Default)]
pub enum JournalDurability {
    /// fsync on every commit — protects against kernel crash and power loss (default).
    #[default]
    Immediate,
    /// No fsync — OS decides flush timing. Suitable for dev/test.
    Eventual,
}

/// Options for `RedbRuntimeEventJournal`.
#[derive(Debug, Clone)]
pub struct RedbJournalOptions {
    pub durability: JournalDurability,
    /// Trigger compaction after this many events in the table. Default: 10_000.
    pub compaction_threshold_events: u64,
}

impl Default for RedbJournalOptions {
    fn default() -> Self {
        Self {
            durability: JournalDurability::Immediate,
            compaction_threshold_events: 10_000,
        }
    }
}

/// Internal wire format stored as redb value bytes (serde_json).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct JournalEntry {
    pub seq: u64,
    pub timestamp_ms: i64,
    #[serde(with = "runtime_event_serde")]
    pub event: RuntimeEvent,
}

/// Filter for `RedbRuntimeEventJournal::inspect`.
pub struct JournalInspectFilter {
    pub route_id: Option<String>,
    pub limit: usize,
}

// ── Adapter ───────────────────────────────────────────────────────────────────

/// Redb-backed implementation of `RuntimeEventJournalPort`.
///
/// `Arc<Database>` allows cheap cloning — all clones share the same underlying
/// redb file handle. `redb::Database` is `Send + Sync`.
#[derive(Clone)]
pub struct RedbRuntimeEventJournal {
    db: Arc<Database>,
    options: RedbJournalOptions,
}

impl RedbRuntimeEventJournal {
    /// Open (or create) the redb database at `path`.
    ///
    /// Parent directories are created if they do not exist.
    /// Both tables are initialised on first open.
    /// Uses `tokio::task::spawn_blocking` because `Database::open` is blocking.
    pub async fn new(
        path: impl Into<PathBuf>,
        options: RedbJournalOptions,
    ) -> Result<Self, CamelError> {
        let path = path.into();
        let db = tokio::task::spawn_blocking(move || {
            if let Some(parent) = path.parent() {
                std::fs::create_dir_all(parent).map_err(|e| {
                    CamelError::Io(format!(
                        "failed to create journal directory '{}': {e}",
                        parent.display()
                    ))
                })?;
            }
            // Retry on transient lock contention (k8s RWO PVC pod handover):
            // the old pod may still hold the file lock when the new one opens.
            let mut attempt: u32 = 0;
            let db = loop {
                attempt += 1;
                match Database::create(&path) {
                    Ok(db) => break db,
                    Err(e) if is_lock_contention(&e) && attempt < OPEN_LOCK_MAX_ATTEMPTS => {
                        tracing::warn!(
                            path = %path.display(),
                            attempt,
                            max_attempts = OPEN_LOCK_MAX_ATTEMPTS,
                            "journal file locked by another handle (pod handover?), retrying"
                        );
                        std::thread::sleep(OPEN_LOCK_RETRY_BACKOFF);
                    }
                    Err(e) => {
                        return Err(CamelError::Io(format!(
                            "failed to open journal at '{}': {e}",
                            path.display()
                        )));
                    }
                }
            };
            // Initialise tables so they exist before any reads.
            let tx = db
                .begin_write()
                .map_err(|e| CamelError::Io(format!("redb begin_write: {e}")))?;
            tx.open_table(EVENTS_TABLE)
                .map_err(|e| CamelError::Io(format!("redb open events table: {e}")))?;
            tx.open_table(COMMAND_IDS_TABLE)
                .map_err(|e| CamelError::Io(format!("redb open command_ids table: {e}")))?;
            tx.commit()
                .map_err(|e| CamelError::Io(format!("redb commit init: {e}")))?;
            Ok::<_, CamelError>(db)
        })
        .await
        .map_err(|e| CamelError::Io(format!("spawn_blocking join: {e}")))??;

        Ok(Self {
            db: Arc::new(db),
            options,
        })
    }

    /// Open an existing database at `path` and return entries (newest-first, up to `filter.limit`).
    ///
    /// Uses `Database::open` + `begin_read` — concurrent with a live writer on the same file.
    /// `inspect` is an offline utility: it does NOT require a live `RedbRuntimeEventJournal` instance.
    pub async fn inspect(
        path: impl Into<PathBuf>,
        filter: JournalInspectFilter,
    ) -> Result<Vec<JournalEntry>, CamelError> {
        let path = path.into();
        let limit = filter.limit;
        let route_id = filter.route_id;
        tokio::task::spawn_blocking(move || {
            if !path.exists() {
                return Err(CamelError::Io(format!(
                    "journal file not found: {}",
                    path.display()
                )));
            }
            let db = Database::open(&path)
                .map_err(|e| CamelError::Io(format!("invalid journal file: {e}")))?;
            let tx = db
                .begin_read()
                .map_err(|e| CamelError::Io(format!("redb begin_read: {e}")))?;
            let table = tx
                .open_table(EVENTS_TABLE)
                .map_err(|e| CamelError::Io(format!("redb open events: {e}")))?;

            // Collect in descending order (newest first).
            // Filter by route_id FIRST, then apply limit — ensures we return
            // `limit` matching entries, not `limit` total entries where most may
            // not match the filter.
            let mut entries: Vec<JournalEntry> = Vec::new();
            for result in table
                .iter()
                .map_err(|e| CamelError::Io(format!("redb iter: {e}")))?
                .rev()
            {
                let (_k, v) = result.map_err(|e| CamelError::Io(format!("redb read: {e}")))?;
                let entry: JournalEntry = serde_json::from_slice(v.value())
                    .map_err(|e| CamelError::Io(format!("journal deserialize: {e}")))?;
                if let Some(ref rid) = route_id
                    && entry.event.route_id() != rid.as_str()
                {
                    continue;
                }
                if entries.len() >= limit {
                    break;
                }
                entries.push(entry);
            }
            Ok(entries)
        })
        .await
        .map_err(|e| CamelError::Io(format!("spawn_blocking join: {e}")))?
    }

    // ── Internal helpers ──────────────────────────────────────────────────────

    fn redb_durability(&self) -> Durability {
        match self.options.durability {
            JournalDurability::Immediate => Durability::Immediate,
            JournalDurability::Eventual => Durability::None,
        }
    }

    /// Derive next sequence number from the last key in the events table.
    /// Must be called inside a write transaction with the table already open.
    fn next_seq(table: &redb::Table<u64, &[u8]>) -> Result<u64, CamelError> {
        match table
            .iter()
            .map_err(|e| CamelError::Io(format!("redb iter for seq: {e}")))?
            .next_back()
        {
            Some(Ok((k, _))) => Ok(k.value() + 1),
            Some(Err(e)) => Err(CamelError::Io(format!("redb seq read: {e}"))),
            None => Ok(0),
        }
    }

    /// Count rows in the events table (read transaction).
    fn event_count(&self) -> Result<u64, CamelError> {
        let tx = self
            .db
            .begin_read()
            .map_err(|e| CamelError::Io(format!("redb begin_read: {e}")))?;
        let table = tx
            .open_table(EVENTS_TABLE)
            .map_err(|e| CamelError::Io(format!("redb open events: {e}")))?;
        table
            .len()
            .map_err(|e| CamelError::Io(format!("redb len: {e}")))
    }

    /// Compact the events table by dropping events that no longer affect replay.
    ///
    /// Two provably replay-equivalent rules run together, per route id:
    /// - **Removal cutoff:** every event with `seq <= last RouteRemoved seq` is
    ///   dropped — a removed route contributes nothing until it is re-registered.
    /// - **Register checkpoint:** every event with `seq < last RouteRegistered
    ///   seq` is dropped. Replaying a RouteRegistered resets the aggregate to a
    ///   fresh `Registered` v0 (see `apply_replayed_event`), so all prior history
    ///   for that route is redundant. This bounds journal growth under
    ///   adopt-on-recovery, where declarative routes re-append their
    ///   `[Registered, StartRequested, Started]` generation on every boot.
    fn compact(&self) -> Result<(), CamelError> {
        let tx = self
            .db
            .begin_write()
            .map_err(|e| CamelError::Io(format!("redb begin_write: {e}")))?;
        {
            let mut table = tx
                .open_table(EVENTS_TABLE)
                .map_err(|e| CamelError::Io(format!("redb open events: {e}")))?;

            // Pass 1: find the last RouteRemoved and last RouteRegistered seq per route.
            let mut last_removed_seq: std::collections::HashMap<String, u64> =
                std::collections::HashMap::new();
            let mut last_registered_seq: std::collections::HashMap<String, u64> =
                std::collections::HashMap::new();
            for result in table
                .iter()
                .map_err(|e| CamelError::Io(format!("redb iter: {e}")))?
            {
                let (k, v) = result.map_err(|e| CamelError::Io(format!("redb read: {e}")))?;
                let seq = k.value();
                let entry: JournalEntry = serde_json::from_slice(v.value())
                    .map_err(|e| CamelError::Io(format!("journal deserialize: {e}")))?;
                match entry.event {
                    RuntimeEvent::RouteRemoved { .. } => {
                        last_removed_seq.insert(entry.event.route_id().to_string(), seq);
                    }
                    RuntimeEvent::RouteRegistered { .. } => {
                        last_registered_seq.insert(entry.event.route_id().to_string(), seq);
                    }
                    _ => {}
                }
            }

            if last_removed_seq.is_empty() && last_registered_seq.is_empty() {
                drop(table);
                tx.commit()
                    .map_err(|e| CamelError::Io(format!("redb commit compact: {e}")))?;
                return Ok(());
            }

            // Pass 2: collect seqs to delete under either rule.
            let mut to_delete: Vec<u64> = Vec::new();
            for result in table
                .iter()
                .map_err(|e| CamelError::Io(format!("redb iter pass2: {e}")))?
            {
                let (k, v) = result.map_err(|e| CamelError::Io(format!("redb read: {e}")))?;
                let seq = k.value();
                let entry: JournalEntry = serde_json::from_slice(v.value())
                    .map_err(|e| CamelError::Io(format!("journal deserialize: {e}")))?;
                let route_id = entry.event.route_id().to_string();
                let removed = last_removed_seq
                    .get(&route_id)
                    .is_some_and(|&cutoff| seq <= cutoff);
                let superseded = last_registered_seq
                    .get(&route_id)
                    .is_some_and(|&checkpoint| seq < checkpoint);
                if removed || superseded {
                    to_delete.push(seq);
                }
            }

            for seq in to_delete {
                table
                    .remove(&seq)
                    .map_err(|e| CamelError::Io(format!("redb remove seq {seq}: {e}")))?;
            }
        }
        tx.commit()
            .map_err(|e| CamelError::Io(format!("redb commit compact: {e}")))?;
        Ok(())
    }
}

// ── RuntimeEvent helper ───────────────────────────────────────────────────────

/// Extension to extract the `route_id` field from any `RuntimeEvent` variant.
trait RuntimeEventExt {
    fn route_id(&self) -> &str;
}

impl RuntimeEventExt for RuntimeEvent {
    fn route_id(&self) -> &str {
        match self {
            RuntimeEvent::RouteRegistered { route_id }
            | RuntimeEvent::RouteStartRequested { route_id }
            | RuntimeEvent::RouteStarted { route_id }
            | RuntimeEvent::RouteFailed { route_id, .. }
            | RuntimeEvent::RouteStopped { route_id }
            | RuntimeEvent::RouteSuspended { route_id }
            | RuntimeEvent::RouteResumed { route_id }
            | RuntimeEvent::RouteReloaded { route_id }
            | RuntimeEvent::RouteRemoved { route_id } => route_id,
        }
    }
}

// ── RuntimeEventJournalPort impl ──────────────────────────────────────────────

#[async_trait]
impl RuntimeEventJournalPort for RedbRuntimeEventJournal {
    async fn append_batch(&self, events: &[RuntimeEvent]) -> Result<(), DomainError> {
        if events.is_empty() {
            return Ok(());
        }
        let db = Arc::clone(&self.db);
        let durability = self.redb_durability();
        let events = events.to_vec();
        let now_ms = chrono::Utc::now().timestamp_millis();

        tokio::task::spawn_blocking(move || {
            // NOTE: `mut` is required — `set_durability` takes `&mut self` in redb v2.
            let mut tx = db
                .begin_write()
                .map_err(|e| CamelError::Io(format!("redb begin_write: {e}")))?;
            tx.set_durability(durability)
                .map_err(|e| CamelError::Io(format!("redb set_durability: {e}")))?;
            {
                let mut table = tx
                    .open_table(EVENTS_TABLE)
                    .map_err(|e| CamelError::Io(format!("redb open events: {e}")))?;
                let start_seq = Self::next_seq(&table)?;
                for (next_seq, event) in (start_seq..).zip(events) {
                    let entry = JournalEntry {
                        seq: next_seq,
                        timestamp_ms: now_ms,
                        event,
                    };
                    let bytes = serde_json::to_vec(&entry)
                        .map_err(|e| CamelError::Io(format!("journal serialize: {e}")))?;
                    table
                        .insert(&next_seq, bytes.as_slice())
                        .map_err(|e| CamelError::Io(format!("redb insert: {e}")))?;
                }
            }
            tx.commit()
                .map_err(|e| CamelError::Io(format!("redb commit: {e}")))?;
            Ok::<_, CamelError>(())
        })
        .await
        .map_err(|e| DomainError::InvalidState(format!("spawn_blocking join: {e}")))?
        .map_err(|e| DomainError::InvalidState(e.to_string()))?;

        // Trigger compaction if threshold exceeded. Non-fatal if it fails.
        // Both event_count() and compact() do blocking redb I/O — run in spawn_blocking.
        let journal_clone = self.clone();
        let threshold = self.options.compaction_threshold_events;
        tokio::task::spawn_blocking(move || match journal_clone.event_count() {
            Ok(count) if count >= threshold => {
                if let Err(e) = journal_clone.compact() {
                    tracing::warn!("journal compaction failed (non-fatal): {e}");
                }
            }
            Ok(_) => {}
            Err(e) => {
                tracing::warn!("journal event count check failed (non-fatal): {e}");
            }
        })
        .await
        .ok(); // Non-fatal: if spawn_blocking panics, we ignore it

        Ok(())
    }

    async fn load_all(&self) -> Result<Vec<RuntimeEvent>, DomainError> {
        let db = Arc::clone(&self.db);
        tokio::task::spawn_blocking(move || {
            let tx = db
                .begin_read()
                .map_err(|e| CamelError::Io(format!("redb begin_read: {e}")))?;
            let table = tx
                .open_table(EVENTS_TABLE)
                .map_err(|e| CamelError::Io(format!("redb open events: {e}")))?;
            let mut events = Vec::new();
            for result in table
                .iter()
                .map_err(|e| CamelError::Io(format!("redb iter: {e}")))?
            {
                let (_k, v) = result.map_err(|e| CamelError::Io(format!("redb read: {e}")))?;
                let entry: JournalEntry = serde_json::from_slice(v.value())
                    .map_err(|e| CamelError::Io(format!("journal deserialize: {e}")))?;
                events.push(entry.event);
            }
            Ok(events)
        })
        .await
        .map_err(|e| DomainError::InvalidState(format!("spawn_blocking join: {e}")))?
        .map_err(|e: CamelError| DomainError::InvalidState(e.to_string()))
    }

    async fn append_command_id(&self, command_id: &str) -> Result<(), DomainError> {
        let db = Arc::clone(&self.db);
        let durability = self.redb_durability();
        let id = command_id.to_string();
        tokio::task::spawn_blocking(move || {
            // NOTE: `mut` required — `set_durability` takes `&mut self` in redb v2.
            let mut tx = db
                .begin_write()
                .map_err(|e| CamelError::Io(format!("redb begin_write: {e}")))?;
            tx.set_durability(durability)
                .map_err(|e| CamelError::Io(format!("redb set_durability: {e}")))?;
            {
                let mut table = tx
                    .open_table(COMMAND_IDS_TABLE)
                    .map_err(|e| CamelError::Io(format!("redb open command_ids: {e}")))?;
                table
                    .insert(id.as_str(), ())
                    .map_err(|e| CamelError::Io(format!("redb insert command_id: {e}")))?;
            }
            tx.commit()
                .map_err(|e| CamelError::Io(format!("redb commit: {e}")))?;
            Ok::<_, CamelError>(())
        })
        .await
        .map_err(|e| DomainError::InvalidState(format!("spawn_blocking join: {e}")))?
        .map_err(|e| DomainError::InvalidState(e.to_string()))
    }

    async fn remove_command_id(&self, command_id: &str) -> Result<(), DomainError> {
        let db = Arc::clone(&self.db);
        let durability = self.redb_durability();
        let id = command_id.to_string();
        tokio::task::spawn_blocking(move || {
            // NOTE: `mut` required — `set_durability` takes `&mut self` in redb v2.
            let mut tx = db
                .begin_write()
                .map_err(|e| CamelError::Io(format!("redb begin_write: {e}")))?;
            tx.set_durability(durability)
                .map_err(|e| CamelError::Io(format!("redb set_durability: {e}")))?;
            {
                let mut table = tx
                    .open_table(COMMAND_IDS_TABLE)
                    .map_err(|e| CamelError::Io(format!("redb open command_ids: {e}")))?;
                table
                    .remove(id.as_str())
                    .map_err(|e| CamelError::Io(format!("redb remove command_id: {e}")))?;
            }
            tx.commit()
                .map_err(|e| CamelError::Io(format!("redb commit: {e}")))?;
            Ok::<_, CamelError>(())
        })
        .await
        .map_err(|e| DomainError::InvalidState(format!("spawn_blocking join: {e}")))?
        .map_err(|e| DomainError::InvalidState(e.to_string()))
    }

    async fn load_command_ids(&self) -> Result<Vec<String>, DomainError> {
        let db = Arc::clone(&self.db);
        tokio::task::spawn_blocking(move || {
            let tx = db
                .begin_read()
                .map_err(|e| CamelError::Io(format!("redb begin_read: {e}")))?;
            let table = tx
                .open_table(COMMAND_IDS_TABLE)
                .map_err(|e| CamelError::Io(format!("redb open command_ids: {e}")))?;
            let mut ids = Vec::new();
            for result in table
                .iter()
                .map_err(|e| CamelError::Io(format!("redb iter: {e}")))?
            {
                let (k, _) = result.map_err(|e| CamelError::Io(format!("redb read: {e}")))?;
                ids.push(k.value().to_string());
            }
            Ok(ids)
        })
        .await
        .map_err(|e| DomainError::InvalidState(format!("spawn_blocking join: {e}")))?
        .map_err(|e: CamelError| DomainError::InvalidState(e.to_string()))
    }
}

// ── Unit tests ────────────────────────────────────────────────────────────────

#[cfg(test)]
mod tests {
    use super::*;
    use camel_component_api::test_support::timeout_wait;
    use tempfile::tempdir;

    async fn new_journal(dir: &tempfile::TempDir) -> RedbRuntimeEventJournal {
        RedbRuntimeEventJournal::new(dir.path().join("test.db"), RedbJournalOptions::default())
            .await
            .unwrap()
    }

    #[tokio::test]
    async fn redb_journal_roundtrip() {
        let dir = tempdir().unwrap();
        let journal = new_journal(&dir).await;

        let events = vec![
            RuntimeEvent::RouteRegistered {
                route_id: "r1".to_string(),
            },
            RuntimeEvent::RouteStarted {
                route_id: "r1".to_string(),
            },
        ];
        journal.append_batch(&events).await.unwrap();

        let loaded = journal.load_all().await.unwrap();
        assert_eq!(loaded, events);
    }

    #[tokio::test]
    async fn redb_journal_command_id_lifecycle() {
        let dir = tempdir().unwrap();
        let journal = new_journal(&dir).await;

        journal.append_command_id("c1").await.unwrap();
        journal.append_command_id("c2").await.unwrap();
        journal.remove_command_id("c1").await.unwrap();

        let ids = journal.load_command_ids().await.unwrap();
        assert_eq!(ids, vec!["c2".to_string()]);
    }

    #[tokio::test]
    async fn redb_journal_compaction_removes_completed_routes() {
        let dir = tempdir().unwrap();
        // Threshold of 1 triggers compaction on every append.
        let journal = RedbRuntimeEventJournal::new(
            dir.path().join("compact.db"),
            RedbJournalOptions {
                durability: JournalDurability::Eventual,
                compaction_threshold_events: 1,
            },
        )
        .await
        .unwrap();

        // Removed route — full lifecycle.
        journal
            .append_batch(&[RuntimeEvent::RouteRegistered {
                route_id: "old".to_string(),
            }])
            .await
            .unwrap();
        journal
            .append_batch(&[RuntimeEvent::RouteRemoved {
                route_id: "old".to_string(),
            }])
            .await
            .unwrap();

        // Active route — no RouteRemoved.
        journal
            .append_batch(&[RuntimeEvent::RouteRegistered {
                route_id: "live".to_string(),
            }])
            .await
            .unwrap();

        let loaded = journal.load_all().await.unwrap();
        assert!(
            !loaded.iter().any(
                |e| matches!(e, RuntimeEvent::RouteRegistered { route_id } if route_id == "old")
            ),
            "old route events must be compacted"
        );
        assert!(
            loaded.iter().any(
                |e| matches!(e, RuntimeEvent::RouteRegistered { route_id } if route_id == "live")
            ),
            "live route events must survive compaction"
        );
    }

    #[tokio::test]
    async fn redb_journal_compaction_preserves_reregistered_route() {
        let dir = tempdir().unwrap();
        let journal = RedbRuntimeEventJournal::new(
            dir.path().join("rereg.db"),
            RedbJournalOptions {
                durability: JournalDurability::Eventual,
                compaction_threshold_events: 1,
            },
        )
        .await
        .unwrap();

        journal
            .append_batch(&[RuntimeEvent::RouteRegistered {
                route_id: "rereg".to_string(),
            }])
            .await
            .unwrap();
        journal
            .append_batch(&[RuntimeEvent::RouteRemoved {
                route_id: "rereg".to_string(),
            }])
            .await
            .unwrap();
        journal
            .append_batch(&[RuntimeEvent::RouteRegistered {
                route_id: "rereg".to_string(),
            }])
            .await
            .unwrap();

        let loaded = journal.load_all().await.unwrap();
        let rereg_count = loaded
            .iter()
            .filter(
                |e| matches!(e, RuntimeEvent::RouteRegistered { route_id } if route_id == "rereg"),
            )
            .count();
        assert_eq!(
            rereg_count, 1,
            "re-registered route must have exactly one event after compaction"
        );
    }

    #[tokio::test]
    async fn redb_journal_durability_eventual() {
        let dir = tempdir().unwrap();
        let journal = RedbRuntimeEventJournal::new(
            dir.path().join("eventual.db"),
            RedbJournalOptions {
                durability: JournalDurability::Eventual,
                compaction_threshold_events: 10_000,
            },
        )
        .await
        .unwrap();

        journal
            .append_batch(&[RuntimeEvent::RouteRegistered {
                route_id: "ev".to_string(),
            }])
            .await
            .unwrap();
        let loaded = journal.load_all().await.unwrap();
        assert_eq!(loaded.len(), 1);
    }

    #[tokio::test]
    async fn redb_journal_clone_shares_db() {
        let dir = tempdir().unwrap();
        let j1 = new_journal(&dir).await;
        let j2 = j1.clone();

        j1.append_batch(&[RuntimeEvent::RouteRegistered {
            route_id: "shared".to_string(),
        }])
        .await
        .unwrap();

        // j2 must see j1's write since they share the same Arc<Database>.
        let loaded = j2.load_all().await.unwrap();
        assert_eq!(loaded.len(), 1);
    }

    #[tokio::test]
    async fn redb_journal_append_empty_batch_is_noop() {
        let dir = tempdir().unwrap();
        let journal = new_journal(&dir).await;

        journal.append_batch(&[]).await.unwrap();
        let loaded = journal.load_all().await.unwrap();
        assert!(loaded.is_empty());
    }

    #[tokio::test]
    async fn redb_journal_sequence_numbers_across_batches() {
        let dir = tempdir().unwrap();
        let journal = new_journal(&dir).await;

        journal
            .append_batch(&[
                RuntimeEvent::RouteRegistered {
                    route_id: "r1".to_string(),
                },
                RuntimeEvent::RouteStarted {
                    route_id: "r1".to_string(),
                },
            ])
            .await
            .unwrap();

        journal
            .append_batch(&[RuntimeEvent::RouteStopped {
                route_id: "r1".to_string(),
            }])
            .await
            .unwrap();

        let loaded = journal.load_all().await.unwrap();
        assert_eq!(loaded.len(), 3);

        // Drop journal to release redb lock before inspect.
        drop(journal);

        // Verify sequence numbers are monotonic.
        let entries = RedbRuntimeEventJournal::inspect(
            dir.path().join("test.db"),
            JournalInspectFilter {
                route_id: None,
                limit: 10,
            },
        )
        .await
        .unwrap();
        let seqs: Vec<u64> = entries.iter().map(|e| e.seq).collect();
        // inspect returns newest-first.
        assert_eq!(seqs, vec![2, 1, 0]);
    }

    #[tokio::test]
    async fn redb_journal_load_all_empty() {
        let dir = tempdir().unwrap();
        let journal = new_journal(&dir).await;
        let loaded = journal.load_all().await.unwrap();
        assert!(loaded.is_empty());
    }

    #[tokio::test]
    async fn redb_journal_inspect_file_not_found() {
        let dir = tempdir().unwrap();
        let result = RedbRuntimeEventJournal::inspect(
            dir.path().join("nonexistent.db"),
            JournalInspectFilter {
                route_id: None,
                limit: 10,
            },
        )
        .await;
        assert!(result.is_err());
        let err = result.unwrap_err().to_string();
        assert!(err.contains("journal file not found"));
    }

    #[tokio::test]
    async fn redb_journal_inspect_with_route_id_filter() {
        let dir = tempdir().unwrap();
        let journal = new_journal(&dir).await;

        journal
            .append_batch(&[
                RuntimeEvent::RouteRegistered {
                    route_id: "alpha".to_string(),
                },
                RuntimeEvent::RouteRegistered {
                    route_id: "beta".to_string(),
                },
                RuntimeEvent::RouteStarted {
                    route_id: "alpha".to_string(),
                },
            ])
            .await
            .unwrap();

        drop(journal);

        let entries = RedbRuntimeEventJournal::inspect(
            dir.path().join("test.db"),
            JournalInspectFilter {
                route_id: Some("alpha".to_string()),
                limit: 10,
            },
        )
        .await
        .unwrap();

        assert_eq!(entries.len(), 2);
        assert!(entries.iter().all(|e| {
            matches!(&e.event, RuntimeEvent::RouteRegistered { route_id } | RuntimeEvent::RouteStarted { route_id } if route_id == "alpha")
        }));
    }

    #[tokio::test]
    async fn redb_journal_inspect_limit_enforcement() {
        let dir = tempdir().unwrap();
        let journal = new_journal(&dir).await;

        for i in 0..5 {
            journal
                .append_batch(&[RuntimeEvent::RouteRegistered {
                    route_id: format!("r{i}"),
                }])
                .await
                .unwrap();
        }

        drop(journal);

        let entries = RedbRuntimeEventJournal::inspect(
            dir.path().join("test.db"),
            JournalInspectFilter {
                route_id: None,
                limit: 2,
            },
        )
        .await
        .unwrap();

        assert_eq!(entries.len(), 2);
        // Newest first: r4, r3
        assert!(
            matches!(&entries[0].event, RuntimeEvent::RouteRegistered { route_id } if route_id == "r4")
        );
        assert!(
            matches!(&entries[1].event, RuntimeEvent::RouteRegistered { route_id } if route_id == "r3")
        );
    }

    #[tokio::test]
    async fn redb_journal_inspect_limit_with_filter_returns_matching_count() {
        let dir = tempdir().unwrap();
        let journal = new_journal(&dir).await;

        // Interleave alpha and beta events.
        for i in 0..4 {
            let rid = if i % 2 == 0 { "alpha" } else { "beta" };
            journal
                .append_batch(&[RuntimeEvent::RouteRegistered {
                    route_id: rid.to_string(),
                }])
                .await
                .unwrap();
        }

        drop(journal);

        // limit=1 but 2 alpha events exist — should return exactly 1 alpha.
        let entries = RedbRuntimeEventJournal::inspect(
            dir.path().join("test.db"),
            JournalInspectFilter {
                route_id: Some("alpha".to_string()),
                limit: 1,
            },
        )
        .await
        .unwrap();

        assert_eq!(entries.len(), 1);
        assert!(
            matches!(&entries[0].event, RuntimeEvent::RouteRegistered { route_id } if route_id == "alpha")
        );
    }

    #[test]
    fn redb_journal_durability_default_is_immediate() {
        assert_eq!(JournalDurability::default(), JournalDurability::Immediate);
    }

    #[test]
    fn redb_journal_options_default() {
        let opts = RedbJournalOptions::default();
        assert_eq!(opts.durability, JournalDurability::Immediate);
        assert_eq!(opts.compaction_threshold_events, 10_000);
    }

    #[test]
    fn redb_journal_entry_serialization_roundtrip() {
        let entry = JournalEntry {
            seq: 42,
            timestamp_ms: 1_700_000_000_000,
            event: RuntimeEvent::RouteFailed {
                route_id: "fail-route".to_string(),
                error: "boom".to_string(),
            },
        };

        let bytes = serde_json::to_vec(&entry).unwrap();
        let decoded: JournalEntry = serde_json::from_slice(&bytes).unwrap();
        assert_eq!(decoded.seq, 42);
        assert_eq!(decoded.timestamp_ms, 1_700_000_000_000);
        assert_eq!(decoded.event, entry.event);
    }

    #[test]
    fn redb_journal_runtime_event_ext_all_variants() {
        let events = [
            RuntimeEvent::RouteRegistered {
                route_id: "a".into(),
            },
            RuntimeEvent::RouteStartRequested {
                route_id: "b".into(),
            },
            RuntimeEvent::RouteStarted {
                route_id: "c".into(),
            },
            RuntimeEvent::RouteFailed {
                route_id: "d".into(),
                error: "err".into(),
            },
            RuntimeEvent::RouteStopped {
                route_id: "e".into(),
            },
            RuntimeEvent::RouteSuspended {
                route_id: "f".into(),
            },
            RuntimeEvent::RouteResumed {
                route_id: "g".into(),
            },
            RuntimeEvent::RouteReloaded {
                route_id: "h".into(),
            },
            RuntimeEvent::RouteRemoved {
                route_id: "i".into(),
            },
        ];
        let expected = ["a", "b", "c", "d", "e", "f", "g", "h", "i"];
        for (event, expected_id) in events.iter().zip(expected.iter()) {
            assert_eq!(event.route_id(), *expected_id);
        }
    }

    #[tokio::test]
    async fn redb_journal_compaction_no_removed_routes_early_return() {
        let dir = tempdir().unwrap();
        let journal = RedbRuntimeEventJournal::new(
            dir.path().join("no_remove.db"),
            RedbJournalOptions {
                durability: JournalDurability::Eventual,
                compaction_threshold_events: 1,
            },
        )
        .await
        .unwrap();

        // Only registered and started — no RouteRemoved.
        journal
            .append_batch(&[RuntimeEvent::RouteRegistered {
                route_id: "active".to_string(),
            }])
            .await
            .unwrap();
        journal
            .append_batch(&[RuntimeEvent::RouteStarted {
                route_id: "active".to_string(),
            }])
            .await
            .unwrap();

        let loaded = journal.load_all().await.unwrap();
        assert_eq!(loaded.len(), 2);
    }

    #[tokio::test]
    async fn redb_journal_command_ids_multiple_and_remove_nonexistent() {
        let dir = tempdir().unwrap();
        let journal = new_journal(&dir).await;

        journal.append_command_id("cmd1").await.unwrap();
        journal.append_command_id("cmd2").await.unwrap();
        journal.append_command_id("cmd3").await.unwrap();

        // Remove a non-existent command — should not error.
        journal.remove_command_id("nonexistent").await.unwrap();

        let ids = journal.load_command_ids().await.unwrap();
        assert_eq!(ids.len(), 3);
        assert!(ids.contains(&"cmd1".to_string()));
        assert!(ids.contains(&"cmd2".to_string()));
        assert!(ids.contains(&"cmd3".to_string()));
    }

    #[tokio::test]
    async fn redb_journal_multiple_routes_compaction() {
        let dir = tempdir().unwrap();
        let journal = RedbRuntimeEventJournal::new(
            dir.path().join("multi_compact.db"),
            RedbJournalOptions {
                durability: JournalDurability::Eventual,
                compaction_threshold_events: 1,
            },
        )
        .await
        .unwrap();

        // Two removed routes, one active.
        journal
            .append_batch(&[RuntimeEvent::RouteRegistered {
                route_id: "removed1".to_string(),
            }])
            .await
            .unwrap();
        journal
            .append_batch(&[RuntimeEvent::RouteRemoved {
                route_id: "removed1".to_string(),
            }])
            .await
            .unwrap();
        journal
            .append_batch(&[RuntimeEvent::RouteRegistered {
                route_id: "removed2".to_string(),
            }])
            .await
            .unwrap();
        journal
            .append_batch(&[RuntimeEvent::RouteRemoved {
                route_id: "removed2".to_string(),
            }])
            .await
            .unwrap();
        journal
            .append_batch(&[RuntimeEvent::RouteRegistered {
                route_id: "kept".to_string(),
            }])
            .await
            .unwrap();

        let loaded = journal.load_all().await.unwrap();
        assert!(
            !loaded.iter().any(|e| matches!(e, RuntimeEvent::RouteRegistered { route_id } if route_id == "removed1" || route_id == "removed2")),
            "removed routes must be compacted"
        );
        assert!(
            loaded.iter().any(
                |e| matches!(e, RuntimeEvent::RouteRegistered { route_id } if route_id == "kept")
            ),
            "kept route must survive"
        );
    }

    #[tokio::test]
    async fn compaction_collapses_redundant_generations_at_last_registered() {
        // adopt-on-recovery re-appends [Registered, StartRequested, Started]
        // for the same declarative route on every boot. Because replay of a
        // RouteRegistered resets the aggregate to v0, every event before the
        // route's LAST RouteRegistered is redundant and must be compacted so
        // the durable journal does not grow without bound across restarts.
        let dir = tempdir().unwrap();
        let journal = RedbRuntimeEventJournal::new(
            dir.path().join("generations.db"),
            RedbJournalOptions {
                durability: JournalDurability::Eventual,
                // Keep compaction from firing until the whole history is written,
                // so the test asserts the compaction rule, not append timing.
                compaction_threshold_events: 1_000,
            },
        )
        .await
        .unwrap();

        // Three boot generations for the same route.
        for _ in 0..3 {
            journal
                .append_batch(&[
                    RuntimeEvent::RouteRegistered {
                        route_id: "declarative".to_string(),
                    },
                    RuntimeEvent::RouteStartRequested {
                        route_id: "declarative".to_string(),
                    },
                    RuntimeEvent::RouteStarted {
                        route_id: "declarative".to_string(),
                    },
                ])
                .await
                .unwrap();
        }
        assert_eq!(journal.load_all().await.unwrap().len(), 9);

        journal.compact().unwrap();

        // Only the last generation survives, and it still replays to Started.
        let loaded = journal.load_all().await.unwrap();
        assert_eq!(
            loaded.len(),
            3,
            "history before the last RouteRegistered must be compacted, got {loaded:?}"
        );
        assert!(matches!(
            loaded.first(),
            Some(RuntimeEvent::RouteRegistered { route_id }) if route_id == "declarative"
        ));
        assert!(matches!(
            loaded.last(),
            Some(RuntimeEvent::RouteStarted { route_id }) if route_id == "declarative"
        ));
    }

    #[tokio::test]
    async fn compaction_keeps_single_generation_intact() {
        // Regression guard: a route with a single generation (one
        // RouteRegistered) must keep its full post-register history — the
        // checkpoint rule only drops events strictly before the LAST register.
        let dir = tempdir().unwrap();
        let journal = RedbRuntimeEventJournal::new(
            dir.path().join("single_gen.db"),
            RedbJournalOptions {
                durability: JournalDurability::Eventual,
                compaction_threshold_events: 1_000,
            },
        )
        .await
        .unwrap();

        journal
            .append_batch(&[
                RuntimeEvent::RouteRegistered {
                    route_id: "r".to_string(),
                },
                RuntimeEvent::RouteStartRequested {
                    route_id: "r".to_string(),
                },
                RuntimeEvent::RouteStarted {
                    route_id: "r".to_string(),
                },
                RuntimeEvent::RouteStopped {
                    route_id: "r".to_string(),
                },
            ])
            .await
            .unwrap();

        journal.compact().unwrap();

        assert_eq!(journal.load_all().await.unwrap().len(), 4);
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn open_retries_until_lock_released() {
        // Simulates the k8s RWO PVC handover: a second open of the same journal
        // file is blocked by the first handle's lock, then succeeds once that
        // handle is dropped — instead of crashing with "Database already open".
        let dir = tempdir().unwrap();
        let path = dir.path().join("locked.db");

        let first = RedbRuntimeEventJournal::new(path.clone(), RedbJournalOptions::default())
            .await
            .unwrap();

        let path2 = path.clone();
        let opener = tokio::spawn(async move {
            RedbRuntimeEventJournal::new(path2, RedbJournalOptions::default()).await
        });

        // Hold the lock across at least one retry backoff, then release it.
        tokio::time::sleep(std::time::Duration::from_millis(400)).await;
        drop(first);

        let second = timeout_wait(
            "redb open retry task join",
            std::time::Duration::from_secs(30),
            opener,
        )
        .await
        .unwrap();
        assert!(
            second.is_ok(),
            "second open must succeed after the lock is released, got err: {:?}",
            second.err().map(|e| e.to_string())
        );
    }

    #[tokio::test]
    async fn open_fails_fast_on_non_lock_error() {
        // A genuine bad path (parent is a file, not a directory) is not lock
        // contention and must surface immediately, not spin the retry loop.
        let dir = tempdir().unwrap();
        let file_as_parent = dir.path().join("iamafile");
        std::fs::write(&file_as_parent, b"x").unwrap();
        let bogus = file_as_parent.join("journal.db");

        let start = std::time::Instant::now();
        let result = RedbRuntimeEventJournal::new(bogus, RedbJournalOptions::default()).await;
        assert!(result.is_err());
        assert!(
            start.elapsed() < OPEN_LOCK_RETRY_BACKOFF,
            "non-lock error must fail fast without retrying"
        );
    }
}