hypercall/journal/

engine_journal.rs

//! Durable engine journal for restart-safe command/event logging.
//!
//! This module provides persistent storage for engine commands and their resulting events,
//! enabling:
//! - Restart-safe ACK: API responses only after DB commit
//! - Idempotency: duplicate request_id returns cached response
//! - Auditability: full command + event trace queryable
//! - Future replay: schema supports snapshot + replay recovery

use crate::engine_enums_ext::EventTypeExt;
use crate::observability::command_trace::EngineStateDigest;
use diesel::prelude::*;
use diesel::r2d2::PoolError;
pub use hypercall_db::{JournalCommandSummary, JournalEventRecord, JournalFullRecord};
use hypercall_db_diesel::engine_enums::{CommandType, DbUuid, EventType};
use hypercall_db_diesel::schema::{engine_command_digests, engine_commands, engine_events};
use hypercall_types::EngineMessage;
use std::collections::HashMap;
use std::sync::Arc;

/// Trait for journal writers (enables mock implementations for testing)
pub trait JournalWriter: Send + Sync {
    /// Whether writes through this writer survive process restart.
    fn is_durable(&self) -> bool {
        false
    }

    /// Append a state transition to the journal.
    /// Returns existing record if request_id already exists (idempotency).
    #[allow(clippy::too_many_arguments)]
    fn append_transition(
        &self,
        received_ts_ms: u64,
        command_data: &[u8],
        response_data: Option<&[u8]>,
        order_id: Option<i64>,
        pre_digest: &EngineStateDigest,
        post_digest: &EngineStateDigest,
        duration_ms: u64,
        events: &[EngineMessage],
        request_uuid: DbUuid,
        command_type_enum: Option<CommandType>,
    ) -> Result<JournalAppendResult, EngineJournalError>;

    /// Append a transition while atomically materializing any fill side-effects.
    ///
    /// Writers that do not own durable fill persistence can ignore the side-effects
    /// and fall back to `append_transition`.
    #[allow(clippy::too_many_arguments)]
    fn append_transition_with_fill_side_effects(
        &self,
        received_ts_ms: u64,
        command_data: &[u8],
        response_data: Option<&[u8]>,
        order_id: Option<i64>,
        pre_digest: &EngineStateDigest,
        post_digest: &EngineStateDigest,
        duration_ms: u64,
        events: &[EngineMessage],
        fill_side_effects: &[super::JournalFillSideEffect],
        balance_updates: &[hypercall_types::BalanceUpdate],
        request_uuid: DbUuid,
        command_type_enum: Option<CommandType>,
    ) -> Result<JournalAppendResult, EngineJournalError> {
        let _ = fill_side_effects;
        if !balance_updates.is_empty() {
            return Err(EngineJournalError::InvalidUsage(
                "append_transition_with_fill_side_effects received non-empty balance_updates but this writer does not persist them".to_string(),
            ));
        }
        self.append_transition(
            received_ts_ms,
            command_data,
            response_data,
            order_id,
            pre_digest,
            post_digest,
            duration_ms,
            events,
            request_uuid,
            command_type_enum,
        )
    }

    /// Get a journal record by request_id
    fn get_by_request_id(
        &self,
        request_id: &uuid::Uuid,
    ) -> Result<Option<JournalFullRecord>, EngineJournalError>;

    /// Check if a request_id exists in the journal (lightweight existence check)
    fn request_id_exists(&self, request_id: &uuid::Uuid) -> Result<bool, EngineJournalError>;

    /// Get recent request_ids for idempotency cache warm-up
    fn get_recent_request_ids(
        &self,
        since_hours: i64,
    ) -> Result<std::collections::HashSet<uuid::Uuid>, EngineJournalError>;

    /// Get recent journal command summaries (for monitoring)
    fn get_recent(&self, limit: usize) -> Result<Vec<JournalCommandSummary>, EngineJournalError>;
}

/// Event persistence mode for journal writes.
/// Controls the latency/durability tradeoff.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum EventPersistenceMode {
    /// Persist full event JSON (highest durability, highest latency)
    #[default]
    Full,
    /// Persist only event summaries (type, order_id, trade_id) - lower latency
    Summary,
    /// Don't persist events at all - only command + response + digests (lowest latency)
    None,
}

impl EventPersistenceMode {
    /// Parse from environment variable value
    pub fn from_env_str(s: &str) -> Self {
        match s.to_lowercase().as_str() {
            "full" => EventPersistenceMode::Full,
            "summary" => EventPersistenceMode::Summary,
            "none" => EventPersistenceMode::None,
            _ => {
                tracing::warn!(
                    "Unknown ENGINE_JOURNAL_EVENT_PERSISTENCE value '{}', defaulting to 'full'",
                    s
                );
                EventPersistenceMode::Full
            }
        }
    }

    pub fn from_config(config: &crate::backend_config::JournalRuntimeConfig) -> Self {
        Self::from_env_str(&config.event_persistence)
    }
}

/// Get the set of critical event types that are always persisted in Full mode.
/// These events are needed for the outbox publisher to replay state correctly.
/// Configured via ENGINE_JOURNAL_CRITICAL_EVENT_TYPES env var.
/// Default: "L2Update" (essential for cache state rebuild)
fn get_critical_event_types(configured_values: &[String]) -> std::collections::HashSet<String> {
    let configured: std::collections::HashSet<String> = configured_values
        .iter()
        .map(|value: &String| value.trim().to_string())
        .filter(|value: &String| !value.is_empty())
        .collect();
    if !configured.is_empty() {
        return configured;
    }

    ["L2Update".to_string()].into_iter().collect()
}

/// Check if an event type is critical and should always be persisted in Full mode
fn is_critical_event_type(
    event_type: &str,
    critical_event_types: &std::collections::HashSet<String>,
) -> bool {
    critical_event_types.contains(event_type)
}

/// Error type for engine journal operations
#[derive(Debug)]
pub enum EngineJournalError {
    Connection(String),
    Query(String),
    Serialization(String),
    InvalidUsage(String),
}

impl std::fmt::Display for EngineJournalError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            EngineJournalError::Connection(s) => write!(f, "Database connection error: {}", s),
            EngineJournalError::Query(s) => write!(f, "Query error: {}", s),
            EngineJournalError::Serialization(s) => write!(f, "Serialization error: {}", s),
            EngineJournalError::InvalidUsage(s) => write!(f, "Invalid journal usage: {}", s),
        }
    }
}

impl std::error::Error for EngineJournalError {}

impl From<diesel::result::Error> for EngineJournalError {
    fn from(err: diesel::result::Error) -> Self {
        EngineJournalError::Query(err.to_string())
    }
}

impl From<PoolError> for EngineJournalError {
    fn from(err: PoolError) -> Self {
        EngineJournalError::Connection(err.to_string())
    }
}

impl From<serde_json::Error> for EngineJournalError {
    fn from(err: serde_json::Error) -> Self {
        EngineJournalError::Serialization(err.to_string())
    }
}

fn ensure_no_fill_events(
    events: &[EngineMessage],
    entrypoint: &str,
) -> Result<(), EngineJournalError> {
    if events
        .iter()
        .any(|event| matches!(event, EngineMessage::OrderFilled { .. }))
    {
        return Err(EngineJournalError::InvalidUsage(format!(
            "{} cannot journal OrderFilled events, use append_transition_with_fill_side_effects",
            entrypoint
        )));
    }
    Ok(())
}

/// Result of appending a transition to the journal
#[derive(Debug, Clone)]
pub struct JournalAppendResult {
    /// Engine sequence number (command_id)
    pub command_id: i64,
    /// Response wire bytes (version byte + msgpack) for idempotent return
    pub response_data: Option<Vec<u8>>,
    /// Events that were persisted
    pub events: Vec<EngineMessage>,
    /// Whether this was a new insert (false if idempotent hit)
    pub was_new_insert: bool,
}

/// Durable engine journal writer
pub struct EngineJournalWriter {
    pool: crate::db_handler::DbPool,
    /// Event persistence mode (controls latency/durability tradeoff)
    event_persistence_mode: EventPersistenceMode,
    critical_event_types: std::collections::HashSet<String>,
}

impl EngineJournalWriter {
    /// Create a new journal writer with the given database pool
    pub fn new(pool: crate::db_handler::DbPool) -> Self {
        Self {
            pool,
            event_persistence_mode: EventPersistenceMode::default(),
            critical_event_types: get_critical_event_types(&[]),
        }
    }

    /// Create a new journal writer with explicit event persistence mode
    pub fn new_with_mode(pool: crate::db_handler::DbPool, mode: EventPersistenceMode) -> Self {
        Self {
            pool,
            event_persistence_mode: mode,
            critical_event_types: get_critical_event_types(&[]),
        }
    }

    pub fn from_database_url_with_config(
        database_url: &str,
        pool_max: u32,
        config: &crate::backend_config::JournalRuntimeConfig,
    ) -> Result<Self, EngineJournalError> {
        let auth = crate::db_handler::DbAuthConfig::password(database_url);
        Self::from_auth_config(&auth, pool_max, config)
    }

    pub fn from_auth_config(
        auth: &crate::db_handler::DbAuthConfig,
        pool_max: u32,
        config: &crate::backend_config::JournalRuntimeConfig,
    ) -> Result<Self, EngineJournalError> {
        let pool = hypercall_db_diesel::build_db_pool(auth, pool_max, 30_000, 10_000)
            .map_err(|e| EngineJournalError::Connection(e.to_string()))?;
        Ok(Self {
            pool,
            event_persistence_mode: EventPersistenceMode::from_config(config),
            critical_event_types: get_critical_event_types(&config.critical_event_types),
        })
    }

    /// Get the current event persistence mode
    pub fn event_persistence_mode(&self) -> EventPersistenceMode {
        self.event_persistence_mode
    }

    /// Append a state transition to the journal.
    ///
    /// This is transactional and provides idempotency:
    /// - If request_id already exists, returns the existing record (no new inserts)
    /// - If request_id is new, inserts command, events, and digests atomically
    ///
    /// # Arguments
    /// * `received_ts_ms` - When the command was received
    /// * `command_data` - Wire-format bytes (version byte + msgpack)
    /// * `response_data` - Optional response wire bytes for idempotent returns
    /// * `order_id` - Extracted order_id for indexed column
    /// * `pre_digest` - State digest before applying
    /// * `post_digest` - State digest after applying
    /// * `duration_ms` - Processing duration
    /// * `events` - Events generated by applying the command
    /// * `request_uuid` - Unique request UUID for idempotency
    /// * `command_type_enum` - Command type enum
    #[allow(clippy::too_many_arguments)]
    pub fn append_transition(
        &self,
        received_ts_ms: u64,
        command_data: &[u8],
        response_data: Option<&[u8]>,
        order_id: Option<i64>,
        pre_digest: &EngineStateDigest,
        post_digest: &EngineStateDigest,
        duration_ms: u64,
        events: &[EngineMessage],
        request_uuid: DbUuid,
        command_type_enum: Option<CommandType>,
    ) -> Result<JournalAppendResult, EngineJournalError> {
        ensure_no_fill_events(events, "append_transition()")?;
        self.append_transition_internal(
            received_ts_ms,
            command_data,
            response_data,
            order_id,
            pre_digest,
            post_digest,
            duration_ms,
            events,
            &[],
            &[],
            request_uuid,
            command_type_enum,
        )
    }

    #[allow(clippy::too_many_arguments)]
    fn append_transition_internal(
        &self,
        received_ts_ms: u64,
        command_data: &[u8],
        response_data: Option<&[u8]>,
        order_id: Option<i64>,
        pre_digest: &EngineStateDigest,
        post_digest: &EngineStateDigest,
        duration_ms: u64,
        events: &[EngineMessage],
        fill_side_effects: &[super::JournalFillSideEffect],
        balance_updates: &[hypercall_types::BalanceUpdate],
        request_uuid: DbUuid,
        command_type_enum: Option<CommandType>,
    ) -> Result<JournalAppendResult, EngineJournalError> {
        let mut conn = self.pool.get()?;

        // Check if request_uuid already exists (idempotency check)
        if let Some(existing) = self.get_by_request_id_internal(&mut conn, &request_uuid.0)? {
            // Deserialize stored msgpack bytes back to EngineMessage for idempotent return.
            return Ok(JournalAppendResult {
                command_id: existing.command_id,
                response_data: existing.response_data,
                events: existing
                    .events
                    .into_iter()
                    .map(|e| {
                        EngineMessage::deserialize_from_wire(&e.event_topic, &e.event_data)
                            .expect("Failed to parse stored journal event payload")
                    })
                    .collect(),
                was_new_insert: false,
            });
        }

        // Serialize digests
        // Serialize events to wire-format bytes (version byte + msgpack).
        // All events are serialized in full — the Summary/None modes are deprecated
        // since events now store pre-serialized wire-format bytes.
        use crate::journal::engine_journal_batcher::EventPayload;
        let event_payloads: Vec<EventPayload> = match self.event_persistence_mode {
            EventPersistenceMode::Full => events
                .iter()
                .map(|e| {
                    let event_type_enum = EventType::from_engine_message(e);
                    let event_data = e
                        .serialize_inner()
                        .expect("Failed to serialize EngineMessage");
                    let event_topic = e.topic().to_string();
                    let event_key = e.partition_key();
                    let l2_sequence = match e {
                        EngineMessage::L2Update(l2) => l2.sequence,
                        _ => None,
                    };
                    EventPayload {
                        event_topic,
                        event_key,
                        event_data,
                        l2_sequence,
                        event_type_enum,
                    }
                })
                .collect(),
            EventPersistenceMode::Summary | EventPersistenceMode::None => {
                // In Summary/None modes, only persist critical events (e.g., L2Update)
                events
                    .iter()
                    .filter(|e| {
                        self.event_persistence_mode == EventPersistenceMode::Summary
                            || is_critical_event_type(
                                EventType::from_engine_message(e).as_str(),
                                &self.critical_event_types,
                            )
                    })
                    .map(|e| {
                        let event_type_enum = EventType::from_engine_message(e);
                        let event_data = e
                            .serialize_inner()
                            .expect("Failed to serialize EngineMessage");
                        let event_topic = e.topic().to_string();
                        let event_key = e.partition_key();
                        let l2_sequence = match e {
                            EngineMessage::L2Update(l2) => l2.sequence,
                            _ => None,
                        };
                        EventPayload {
                            event_topic,
                            event_key,
                            event_data,
                            l2_sequence,
                            event_type_enum,
                        }
                    })
                    .collect()
            }
        };

        // Execute transaction with idempotent inserts
        conn.transaction::<_, EngineJournalError, _>(|conn| {
            // Insert command with ON CONFLICT DO NOTHING for idempotency
            // This is faster than separate SELECT + INSERT
            let insert_result: Option<i64> = diesel::insert_into(engine_commands::table)
                .values((
                    engine_commands::received_ts_ms.eq(received_ts_ms as i64),
                    engine_commands::command_data.eq(command_data),
                    engine_commands::response_data.eq(response_data),
                    engine_commands::order_id.eq(order_id),
                    engine_commands::request_uuid.eq(request_uuid),
                    engine_commands::command_type_enum.eq(command_type_enum),
                ))
                .on_conflict(engine_commands::request_uuid)
                .do_nothing()
                .returning(engine_commands::command_id)
                .get_result(conn)
                .optional()?;

            let (command_id, was_new_insert) = match insert_result {
                Some(id) => (id, true),
                None => {
                    // Conflict - fetch existing command_id
                    let existing_id: i64 = engine_commands::table
                        .filter(engine_commands::request_uuid.eq(request_uuid))
                        .select(engine_commands::command_id)
                        .first(conn)?;
                    (existing_id, false)
                }
            };

            // Only insert events and digests if this was a new insert
            if was_new_insert {
                // Batch insert events (much faster than per-event loop)
                if !event_payloads.is_empty() {
                    let event_rows: Vec<_> = event_payloads
                        .iter()
                        .enumerate()
                        .map(|(idx, event)| {
                            (
                                engine_events::command_id.eq(command_id),
                                engine_events::event_idx.eq(idx as i32),
                                engine_events::event_data.eq(&event.event_data),
                                engine_events::event_key.eq(&event.event_key),
                                engine_events::l2_sequence.eq(event.l2_sequence),
                                engine_events::event_type_enum.eq(event.event_type_enum),
                            )
                        })
                        .collect();

                    diesel::insert_into(engine_events::table)
                        .values(&event_rows)
                        .execute(conn)?;
                }

                let fill_side_effects_by_idx: HashMap<i32, &super::JournalFillSideEffect> =
                    fill_side_effects
                        .iter()
                        .map(|side_effect| (side_effect.event_idx, side_effect))
                        .collect();
                for (idx, event) in events.iter().enumerate() {
                    let EngineMessage::OrderFilled { fill, .. } = event else {
                        continue;
                    };

                    let side_effect = fill_side_effects_by_idx.get(&(idx as i32)).unwrap_or_else(|| {
                        panic!(
                            "CRITICAL_FAILURE: missing journal fill side-effect for sync command_id={}, event_idx={}, trade_id={}",
                            command_id, idx, fill.trade_id
                        )
                    });
                    crate::db_handler::DieselEventHandler::persist_fill_with_side_effects_in_tx(
                        conn,
                        fill,
                        side_effect,
                    )
                    .map_err(|e| EngineJournalError::Query(e.to_string()))?;
                }

                for balance_update in balance_updates {
                    hypercall_db_diesel::ledger_ops::apply_pnl_decimal_sync(
                        conn,
                        &balance_update.wallet,
                        balance_update.delta,
                    )
                    .map_err(|e| EngineJournalError::Query(e.to_string()))?;
                }

                // Insert digests (dual-write JSONB + msgpack)
                let pre_digest_data = hypercall_types::serialize_to_wire_bytes(pre_digest);
                let post_digest_data =
                    hypercall_types::serialize_to_wire_bytes(post_digest);
                diesel::insert_into(engine_command_digests::table)
                    .values((
                        engine_command_digests::command_id.eq(command_id),
                        engine_command_digests::duration_ms.eq(duration_ms as i64),
                        engine_command_digests::pre_digest_data.eq(&pre_digest_data),
                        engine_command_digests::post_digest_data.eq(&post_digest_data),
                    ))
                    .on_conflict(engine_command_digests::command_id)
                    .do_nothing()
                    .execute(conn)?;
            }

            Ok(JournalAppendResult {
                command_id,
                response_data: response_data.map(|d| d.to_vec()),
                events: events.to_vec(),
                was_new_insert,
            })
        })
    }

    /// Get a journal record by request_id
    pub fn get_by_request_id(
        &self,
        request_id: &uuid::Uuid,
    ) -> Result<Option<JournalFullRecord>, EngineJournalError> {
        let mut conn = self.pool.get()?;
        self.get_by_request_id_internal(&mut conn, request_id)
    }

    fn get_by_request_id_internal(
        &self,
        conn: &mut PgConnection,
        request_id: &uuid::Uuid,
    ) -> Result<Option<JournalFullRecord>, EngineJournalError> {
        // Query command
        let command: Option<(
            i64,
            DbUuid,
            i64,
            Option<CommandType>,
            Option<Vec<u8>>,
            Option<Vec<u8>>,
            chrono::DateTime<chrono::Utc>,
        )> = engine_commands::table
            .filter(engine_commands::request_uuid.eq(DbUuid(*request_id)))
            .select((
                engine_commands::command_id,
                engine_commands::request_uuid,
                engine_commands::received_ts_ms,
                engine_commands::command_type_enum,
                engine_commands::command_data,
                engine_commands::response_data,
                engine_commands::created_at,
            ))
            .first(conn)
            .optional()?;

        let Some((
            command_id,
            req_id,
            received_ts_ms,
            command_type,
            command_data,
            response_data,
            created_at,
        )) = command
        else {
            return Ok(None);
        };

        // Query events
        let events: Vec<(i64, i32, EventType, Vec<u8>, Option<String>, Option<i64>)> =
            engine_events::table
                .filter(engine_events::command_id.eq(command_id))
                .order(engine_events::event_idx.asc())
                .select((
                    engine_events::event_id,
                    engine_events::event_idx,
                    engine_events::event_type_enum,
                    engine_events::event_data,
                    engine_events::event_key,
                    engine_events::l2_sequence,
                ))
                .load(conn)?;

        // Query digests
        let digest: Option<(Vec<u8>, Vec<u8>, i64)> = engine_command_digests::table
            .filter(engine_command_digests::command_id.eq(command_id))
            .select((
                engine_command_digests::pre_digest_data,
                engine_command_digests::post_digest_data,
                engine_command_digests::duration_ms,
            ))
            .first(conn)
            .optional()?;

        let (pre_digest, post_digest, duration_ms) = digest
            .map(|(pre, post, dur)| {
                let pre =
                    rmp_serde::from_slice::<serde_json::Value>(&pre[1..]).unwrap_or_else(|e| {
                        panic!(
                            "Failed to decode pre_digest_data msgpack (command_id={}): {}",
                            command_id, e
                        )
                    });
                let post =
                    rmp_serde::from_slice::<serde_json::Value>(&post[1..]).unwrap_or_else(|e| {
                        panic!(
                            "Failed to decode post_digest_data msgpack (command_id={}): {}",
                            command_id, e
                        )
                    });
                (Some(pre), Some(post), Some(dur))
            })
            .unwrap_or((None, None, None));

        Ok(Some(JournalFullRecord {
            command_id,
            request_id: req_id.to_string(),
            received_ts_ms,
            command_type: command_type
                .map(|ct| ct.as_str().to_string())
                .unwrap_or_default(),
            command_data: command_data.expect("command_data is NULL — run backfill first"),
            response_data,
            pre_digest,
            post_digest,
            duration_ms,
            events: events
                .into_iter()
                .map(
                    |(event_id, event_idx, event_type_enum, event_data, event_key, l2_sequence)| {
                        JournalEventRecord {
                            event_id,
                            event_idx,
                            event_type: event_type_enum.as_str().to_string(),
                            event_data,
                            event_topic: event_type_enum.topic().to_string(),
                            event_key,
                            l2_sequence,
                        }
                    },
                )
                .collect(),
            created_at,
        }))
    }

    /// Get recent journal command summaries
    pub fn get_recent(
        &self,
        limit: usize,
    ) -> Result<Vec<JournalCommandSummary>, EngineJournalError> {
        let mut conn = self.pool.get()?;

        // Query recent commands with event counts
        let commands: Vec<(
            i64,
            DbUuid,
            i64,
            Option<CommandType>,
            Option<Vec<u8>>,
            chrono::DateTime<chrono::Utc>,
        )> = engine_commands::table
            .order(engine_commands::created_at.desc())
            .limit(limit as i64)
            .select((
                engine_commands::command_id,
                engine_commands::request_uuid,
                engine_commands::received_ts_ms,
                engine_commands::command_type_enum,
                engine_commands::response_data,
                engine_commands::created_at,
            ))
            .load(&mut conn)?;

        let mut summaries = Vec::with_capacity(commands.len());
        for (command_id, request_id, received_ts_ms, command_type, response_data, created_at) in
            commands
        {
            // Count events
            let event_count: i64 = engine_events::table
                .filter(engine_events::command_id.eq(command_id))
                .count()
                .get_result(&mut conn)?;

            // Get first 5 event types as a sample
            let event_types_sample: Vec<EventType> = engine_events::table
                .filter(engine_events::command_id.eq(command_id))
                .order(engine_events::event_idx.asc())
                .limit(5)
                .select(engine_events::event_type_enum)
                .load(&mut conn)?;

            // Get digests
            let digest: Option<(Vec<u8>, Vec<u8>, i64)> = engine_command_digests::table
                .filter(engine_command_digests::command_id.eq(command_id))
                .select((
                    engine_command_digests::pre_digest_data,
                    engine_command_digests::post_digest_data,
                    engine_command_digests::duration_ms,
                ))
                .first(&mut conn)
                .optional()?;

            let (pre_digest, post_digest, duration_ms) = digest
                .map(|(pre, post, dur)| {
                    let pre =
                        rmp_serde::from_slice::<serde_json::Value>(&pre[1..]).unwrap_or_else(|e| {
                            panic!(
                                "Failed to decode pre_digest_data msgpack (command_id={}): {}",
                                command_id, e
                            )
                        });
                    let post = rmp_serde::from_slice::<serde_json::Value>(&post[1..])
                        .unwrap_or_else(|e| {
                            panic!(
                                "Failed to decode post_digest_data msgpack (command_id={}): {}",
                                command_id, e
                            )
                        });
                    (Some(pre), Some(post), Some(dur))
                })
                .unwrap_or((None, None, None));

            summaries.push(JournalCommandSummary {
                command_id,
                request_id: request_id.to_string(),
                received_ts_ms,
                command_type: command_type
                    .map(|ct| ct.as_str().to_string())
                    .unwrap_or_default(),
                response_data,
                event_count,
                event_types_sample: event_types_sample
                    .into_iter()
                    .map(|e| e.as_str().to_string())
                    .collect(),
                pre_digest,
                post_digest,
                duration_ms,
                created_at,
            });
        }

        Ok(summaries)
    }

    /// Get recent request_ids for idempotency cache warm-up.
    /// Returns request_ids from commands received in the last `since_hours` hours.
    /// This is used at startup to populate the in-memory HashSet for fast idempotency checks.
    pub fn get_recent_request_ids(
        &self,
        since_hours: i64,
    ) -> Result<std::collections::HashSet<uuid::Uuid>, EngineJournalError> {
        let mut conn = self.pool.get()?;

        // Calculate cutoff timestamp
        let cutoff = chrono::Utc::now() - chrono::Duration::hours(since_hours);

        let request_ids: Vec<DbUuid> = engine_commands::table
            .filter(engine_commands::created_at.ge(cutoff))
            .select(engine_commands::request_uuid)
            .load(&mut conn)?;

        let count = request_ids.len();
        tracing::info!(
            "Loaded {} request_ids from journal for idempotency cache (since {} hours ago)",
            count,
            since_hours
        );

        Ok(request_ids.into_iter().map(|id| id.0).collect())
    }

    /// Check if a request_id exists in the journal (lightweight existence check).
    /// Returns true if the request_id has been processed before.
    pub fn request_id_exists(&self, request_id: &uuid::Uuid) -> Result<bool, EngineJournalError> {
        let mut conn = self.pool.get()?;

        let count: i64 = engine_commands::table
            .filter(engine_commands::request_uuid.eq(DbUuid(*request_id)))
            .count()
            .get_result(&mut conn)?;

        Ok(count > 0)
    }
}

impl JournalWriter for EngineJournalWriter {
    fn is_durable(&self) -> bool {
        true
    }

    fn append_transition(
        &self,
        received_ts_ms: u64,
        command_data: &[u8],
        response_data: Option<&[u8]>,
        order_id: Option<i64>,
        pre_digest: &EngineStateDigest,
        post_digest: &EngineStateDigest,
        duration_ms: u64,
        events: &[EngineMessage],
        request_uuid: DbUuid,
        command_type_enum: Option<CommandType>,
    ) -> Result<JournalAppendResult, EngineJournalError> {
        EngineJournalWriter::append_transition(
            self,
            received_ts_ms,
            command_data,
            response_data,
            order_id,
            pre_digest,
            post_digest,
            duration_ms,
            events,
            request_uuid,
            command_type_enum,
        )
    }

    fn append_transition_with_fill_side_effects(
        &self,
        received_ts_ms: u64,
        command_data: &[u8],
        response_data: Option<&[u8]>,
        order_id: Option<i64>,
        pre_digest: &EngineStateDigest,
        post_digest: &EngineStateDigest,
        duration_ms: u64,
        events: &[EngineMessage],
        fill_side_effects: &[super::JournalFillSideEffect],
        balance_updates: &[hypercall_types::BalanceUpdate],
        request_uuid: DbUuid,
        command_type_enum: Option<CommandType>,
    ) -> Result<JournalAppendResult, EngineJournalError> {
        self.append_transition_internal(
            received_ts_ms,
            command_data,
            response_data,
            order_id,
            pre_digest,
            post_digest,
            duration_ms,
            events,
            fill_side_effects,
            balance_updates,
            request_uuid,
            command_type_enum,
        )
    }

    fn get_by_request_id(
        &self,
        request_id: &uuid::Uuid,
    ) -> Result<Option<JournalFullRecord>, EngineJournalError> {
        EngineJournalWriter::get_by_request_id(self, request_id)
    }

    fn request_id_exists(&self, request_id: &uuid::Uuid) -> Result<bool, EngineJournalError> {
        EngineJournalWriter::request_id_exists(self, request_id)
    }

    fn get_recent_request_ids(
        &self,
        since_hours: i64,
    ) -> Result<std::collections::HashSet<uuid::Uuid>, EngineJournalError> {
        EngineJournalWriter::get_recent_request_ids(self, since_hours)
    }

    fn get_recent(&self, limit: usize) -> Result<Vec<JournalCommandSummary>, EngineJournalError> {
        EngineJournalWriter::get_recent(self, limit)
    }
}

/// In-memory journal writer for testing.
/// Exercises real idempotency logic without requiring a database.
pub struct InMemoryJournalWriter {
    entries: std::sync::Mutex<HashMap<uuid::Uuid, JournalFullRecord>>,
    next_command_id: std::sync::atomic::AtomicI64,
    durable: bool,
}

impl InMemoryJournalWriter {
    /// Create a new in-memory journal writer
    pub fn new() -> Self {
        Self::with_durability(false)
    }

    fn with_durability(durable: bool) -> Self {
        Self {
            entries: std::sync::Mutex::new(HashMap::new()),
            next_command_id: std::sync::atomic::AtomicI64::new(1),
            durable,
        }
    }

    #[cfg(test)]
    pub(crate) fn durable_for_tests() -> Self {
        Self::with_durability(true)
    }

    /// Get all stored entries (for test verification)
    pub fn get_all_entries(&self) -> Vec<JournalFullRecord> {
        let entries = self.entries.lock().unwrap();
        entries.values().cloned().collect()
    }

    /// Clear all entries (for test isolation)
    pub fn clear(&self) {
        let mut entries = self.entries.lock().unwrap();
        entries.clear();
    }

    /// Get number of entries stored
    pub fn len(&self) -> usize {
        let entries = self.entries.lock().unwrap();
        entries.len()
    }

    /// Check if empty
    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }
}

#[cfg(test)]
pub(crate) fn request_uuid_for_test(uuid: uuid::Uuid) -> DbUuid {
    DbUuid(uuid)
}

impl Default for InMemoryJournalWriter {
    fn default() -> Self {
        Self::new()
    }
}

impl JournalWriter for InMemoryJournalWriter {
    fn is_durable(&self) -> bool {
        self.durable
    }

    fn append_transition(
        &self,
        received_ts_ms: u64,
        command_data: &[u8],
        response_data: Option<&[u8]>,
        _order_id: Option<i64>,
        pre_digest: &EngineStateDigest,
        post_digest: &EngineStateDigest,
        duration_ms: u64,
        events: &[EngineMessage],
        request_uuid: DbUuid,
        command_type_enum: Option<CommandType>,
    ) -> Result<JournalAppendResult, EngineJournalError> {
        let mut entries = self.entries.lock().unwrap();

        // Idempotency check - return existing record if request_uuid exists
        if let Some(existing) = entries.get(&request_uuid.0) {
            return Ok(JournalAppendResult {
                command_id: existing.command_id,
                response_data: existing.response_data.clone(),
                events: existing
                    .events
                    .iter()
                    .map(|e| {
                        EngineMessage::deserialize_from_wire(&e.event_topic, &e.event_data)
                            .expect("Failed to parse in-memory journal event payload")
                    })
                    .collect(),
                was_new_insert: false,
            });
        }

        // Create new entry
        let command_id = self
            .next_command_id
            .fetch_add(1, std::sync::atomic::Ordering::SeqCst);

        let event_records: Vec<JournalEventRecord> = events
            .iter()
            .enumerate()
            .map(|(idx, e)| {
                let event_type = EventType::from_engine_message(e).as_str().to_string();
                let event_data = e
                    .serialize_inner()
                    .expect("Failed to serialize EngineMessage");
                let event_topic = e.topic().to_string();
                let event_key = e.partition_key();
                let l2_sequence = match e {
                    EngineMessage::L2Update(l2) => l2.sequence,
                    _ => None,
                };
                JournalEventRecord {
                    event_id: (command_id * 1000) + idx as i64, // Synthetic event_id
                    event_idx: idx as i32,
                    event_type,
                    event_data,
                    event_topic,
                    event_key,
                    l2_sequence,
                }
            })
            .collect();

        let record = JournalFullRecord {
            command_id,
            request_id: request_uuid.0.to_string(),
            received_ts_ms: received_ts_ms as i64,
            command_type: command_type_enum
                .map(|ct| ct.as_str().to_string())
                .unwrap_or_default(),
            command_data: command_data.to_vec(),
            response_data: response_data.map(|d| d.to_vec()),
            pre_digest: serde_json::to_value(pre_digest).ok(),
            post_digest: serde_json::to_value(post_digest).ok(),
            duration_ms: Some(duration_ms as i64),
            events: event_records,
            created_at: chrono::Utc::now(),
        };

        entries.insert(request_uuid.0, record);

        Ok(JournalAppendResult {
            command_id,
            response_data: response_data.map(|d| d.to_vec()),
            events: events.to_vec(),
            was_new_insert: true,
        })
    }

    #[allow(clippy::too_many_arguments)]
    fn append_transition_with_fill_side_effects(
        &self,
        received_ts_ms: u64,
        command_data: &[u8],
        response_data: Option<&[u8]>,
        order_id: Option<i64>,
        pre_digest: &EngineStateDigest,
        post_digest: &EngineStateDigest,
        duration_ms: u64,
        events: &[EngineMessage],
        _fill_side_effects: &[super::JournalFillSideEffect],
        _balance_updates: &[hypercall_types::BalanceUpdate],
        request_uuid: DbUuid,
        command_type_enum: Option<CommandType>,
    ) -> Result<JournalAppendResult, EngineJournalError> {
        self.append_transition(
            received_ts_ms,
            command_data,
            response_data,
            order_id,
            pre_digest,
            post_digest,
            duration_ms,
            events,
            request_uuid,
            command_type_enum,
        )
    }

    fn get_by_request_id(
        &self,
        request_id: &uuid::Uuid,
    ) -> Result<Option<JournalFullRecord>, EngineJournalError> {
        let entries = self.entries.lock().unwrap();
        Ok(entries.get(request_id).cloned())
    }

    fn request_id_exists(&self, request_id: &uuid::Uuid) -> Result<bool, EngineJournalError> {
        let entries = self.entries.lock().unwrap();
        Ok(entries.contains_key(request_id))
    }

    fn get_recent_request_ids(
        &self,
        _since_hours: i64,
    ) -> Result<std::collections::HashSet<uuid::Uuid>, EngineJournalError> {
        let entries = self.entries.lock().unwrap();
        Ok(entries.keys().cloned().collect())
    }

    fn get_recent(&self, limit: usize) -> Result<Vec<JournalCommandSummary>, EngineJournalError> {
        let entries = self.entries.lock().unwrap();
        let mut summaries: Vec<JournalCommandSummary> = entries
            .values()
            .map(|record| JournalCommandSummary {
                command_id: record.command_id,
                request_id: record.request_id.clone(),
                received_ts_ms: record.received_ts_ms,
                command_type: record.command_type.clone(),
                response_data: record.response_data.clone(),
                event_count: record.events.len() as i64,
                event_types_sample: record
                    .events
                    .iter()
                    .take(5)
                    .map(|e| e.event_type.clone())
                    .collect(),
                pre_digest: record.pre_digest.clone(),
                post_digest: record.post_digest.clone(),
                duration_ms: record.duration_ms,
                created_at: record.created_at,
            })
            .collect();

        // Sort by created_at descending and take limit
        summaries.sort_by_key(|s| std::cmp::Reverse(s.created_at));
        summaries.truncate(limit);
        Ok(summaries)
    }
}

/// Thread-safe wrapper for journal writers (trait object for mock support)
pub type SharedEngineJournalWriter = Arc<dyn JournalWriter>;

#[cfg(test)]
mod tests {
    use super::*;

    #[cfg(feature = "test-utils")]
    async fn setup_test_pool() -> (
        hypercall_db_diesel::DbPool,
        Option<testcontainers::ContainerAsync<testcontainers_modules::postgres::Postgres>>,
        crate::test_contracts::TestnetContractEnvGuard,
    ) {
        use crate::test_contracts::test_database_url_with_database;
        use testcontainers::runners::AsyncRunner;
        use testcontainers::ImageExt;
        use testcontainers_modules::postgres::Postgres;

        let env_guard = crate::test_contracts::TestnetContractEnvGuard::apply();

        let (database_url, container) = if let Ok(url) = std::env::var("TEST_DATABASE_URL") {
            let db_name = format!("test_sync_journal_{}", uuid::Uuid::new_v4().simple());
            let admin_pool = sqlx::postgres::PgPoolOptions::new()
                .max_connections(1)
                .connect(&url)
                .await
                .expect("connect to TEST_DATABASE_URL");
            sqlx::query(&format!(r#"CREATE DATABASE "{}""#, db_name))
                .execute(&admin_pool)
                .await
                .expect("create isolated test database");
            admin_pool.close().await;
            let db_url = test_database_url_with_database(&url, &db_name);
            (db_url, None)
        } else {
            let container = Postgres::default()
                .with_db_name("test_sync_journal")
                .with_user("test_user")
                .with_password("test_password")
                .with_tag("16-alpine")
                .start()
                .await
                .expect("start postgres container");

            let port = container
                .get_host_port_ipv4(5432)
                .await
                .expect("get postgres port");
            let url = format!(
                "postgresql://test_user:test_password@127.0.0.1:{}/test_sync_journal",
                port
            );
            tokio::time::sleep(tokio::time::Duration::from_secs(2)).await;

            (url, Some(container))
        };

        let _handler = crate::db_handler::DieselEventHandler::new(&database_url)
            .expect("create diesel handler and run migrations");
        let auth = hypercall_db_diesel::DbAuthConfig::password(&database_url);
        let pool = hypercall_db_diesel::build_db_pool(&auth, 5, 30_000, 10_000)
            .expect("build diesel pool");

        (pool, container, env_guard)
    }

    #[test]
    fn test_event_type_topic_derive() {
        // Just verify the function doesn't panic
        use hypercall_types::OrderActionMessage;
        use hypercall_types::{OrderInfo, Side, WalletAddress};
        use rust_decimal_macros::dec;
        use std::str::FromStr;

        let wallet = WalletAddress::from_str("0x1234567890123456789012345678901234567890").unwrap();
        let msg = OrderActionMessage {
            timestamp: 1000,
            info: OrderInfo {
                symbol: "TEST".to_string(),
                price: dec!(1.0),
                size: dec!(1.0),
                side: Side::Buy,
                tif: hypercall_types::TimeInForce::GTC,
                client_id: None,
                order_id: None,
                is_perp: false,
                underlying: None,
                reduce_only: None,
                nonce: None,
                signature: None,
                mmp_enabled: false,
                builder_code_address: None,
            },
            action: hypercall_types::OrderAction::CreateOrder,
            wallet,
            api_wallet_address: None,
            mmp_triggered: false,
            request_id: None,
        };

        let event = EngineMessage::OrderAction(msg);
        assert_eq!(
            EventType::from_engine_message(&event).as_str(),
            "OrderAction"
        );
    }

    #[cfg(feature = "test-utils")]
    #[tokio::test]
    #[serial_test::serial(testnet_env)]
    async fn sync_writer_materializes_balance_updates_once() {
        use diesel::prelude::*;
        use rust_decimal_macros::dec;

        let (pool, _container, _env_guard) = setup_test_pool().await;
        let writer = EngineJournalWriter::new(pool.clone());
        let request_uuid = DbUuid(uuid::Uuid::new_v4());
        let wallet = hypercall_types::wallet_address::test_wallet(11);
        let update = hypercall_types::BalanceUpdate {
            balance_update_seq: 1,
            wallet,
            delta: dec!(42),
            balance_after: dec!(42),
            reason: hypercall_types::BalanceUpdateReason::Deposit,
            reference_id: Some("sync-writer-test".to_string()),
            source_command_id: None,
            timestamp_ms: 1234,
        };

        let first = writer
            .append_transition_with_fill_side_effects(
                1234,
                &[1, 2, 3],
                None,
                None,
                &EngineStateDigest::default(),
                &EngineStateDigest::default(),
                1,
                &[],
                &[],
                std::slice::from_ref(&update),
                request_uuid,
                Some(CommandType::DepositUpdate),
            )
            .expect("sync writer should persist balance updates");
        assert!(first.was_new_insert);

        let duplicate = writer
            .append_transition_with_fill_side_effects(
                1234,
                &[1, 2, 3],
                None,
                None,
                &EngineStateDigest::default(),
                &EngineStateDigest::default(),
                1,
                &[],
                &[],
                std::slice::from_ref(&update),
                request_uuid,
                Some(CommandType::DepositUpdate),
            )
            .expect("duplicate request should be idempotent");
        assert!(!duplicate.was_new_insert);

        let mut conn = pool.get().expect("db connection");
        let balance: rust_decimal::Decimal = hypercall_db_diesel::schema::account_balances::table
            .filter(hypercall_db_diesel::schema::account_balances::account_address.eq(wallet))
            .select(hypercall_db_diesel::schema::account_balances::balance)
            .first(&mut conn)
            .expect("projection balance");
        assert_eq!(balance, dec!(42));
    }
}