hypercall_db_diesel/

archiver.rs

//! Diesel implementation of ArchiverReader and ArchiverWriter traits.
//!
//! Moves all SQL from `tools/db-archiver/src/db.rs` into the shared
//! persistence layer so the archiver binary becomes a thin orchestrator.

use anyhow::{Context, Result};
use chrono::{DateTime, Duration, Utc};
use diesel::prelude::*;
use diesel::sql_query;
use diesel::sql_types::{BigInt, Nullable};
use tracing::{info, warn};

use crate::database_handler::DatabaseHandler;
use crate::engine_enums::{CommandType, DbUuid, EventType};
use crate::models::{OrderActionRecord, OrderUpdateRecord};
use crate::schema::{
    engine_command_digests, engine_commands, engine_events, notifications, order_actions,
    order_updates,
};
use hypercall_db::{
    ArchivalBoundary, ArchivedCommand, ArchivedEvent, ArchivedOrderAction, ArchivedOrderUpdate,
    ArchiverReader, ArchiverWriter,
};

// ---------------------------------------------------------------------------
// Private Diesel row types (same pattern as analytics.rs)
// ---------------------------------------------------------------------------

/// Typed struct matching engine_events columns.
#[derive(Queryable, Selectable, Debug)]
#[diesel(table_name = engine_events)]
struct EventRow {
    pub event_id: i64,
    pub command_id: i64,
    pub event_idx: i32,
    pub event_data: Vec<u8>,
    pub event_key: Option<String>,
    pub l2_sequence: Option<i64>,
    pub created_at: DateTime<Utc>,
    pub published_at: Option<DateTime<Utc>>,
    pub publish_attempts: i32,
    pub last_publish_error: Option<String>,
    pub last_publish_attempt_at: Option<DateTime<Utc>>,
    pub event_type_enum: EventType,
}

impl From<EventRow> for ArchivedEvent {
    fn from(r: EventRow) -> Self {
        Self {
            event_id: r.event_id,
            command_id: r.command_id,
            event_idx: r.event_idx,
            event_data: r.event_data,
            event_key: r.event_key,
            l2_sequence: r.l2_sequence,
            created_at: r.created_at,
            published_at: r.published_at,
            publish_attempts: r.publish_attempts,
            last_publish_error: r.last_publish_error,
            last_publish_attempt_at: r.last_publish_attempt_at,
            event_type: r.event_type_enum.as_str().to_string(),
        }
    }
}

/// Typed struct matching engine_commands columns.
#[derive(Queryable, Selectable, QueryableByName, Debug)]
#[diesel(table_name = engine_commands)]
struct CommandRow {
    pub command_id: i64,
    pub received_ts_ms: i64,
    pub command_data: Option<Vec<u8>>,
    pub response_data: Option<Vec<u8>>,
    pub order_id: Option<i64>,
    pub request_uuid: DbUuid,
    pub command_type_enum: Option<CommandType>,
    pub created_at: DateTime<Utc>,
}

impl From<CommandRow> for ArchivedCommand {
    fn from(r: CommandRow) -> Self {
        Self {
            command_id: r.command_id,
            received_ts_ms: r.received_ts_ms,
            command_data: r.command_data,
            response_data: r.response_data,
            order_id: r.order_id,
            request_uuid: r.request_uuid.to_string(),
            command_type: r.command_type_enum.map(|ct| ct.as_str().to_string()),
            created_at: r.created_at,
        }
    }
}

impl From<OrderActionRecord> for ArchivedOrderAction {
    fn from(r: OrderActionRecord) -> Self {
        Self {
            id: r.id,
            timestamp: r.timestamp,
            wallet: r.wallet.0.as_slice().to_vec(),
            action: r.action,
            symbol: r.symbol,
            price: r.price,
            size: r.size,
            side: r.side,
            tif: r.tif,
            client_id: r.client_id,
            created_at: r.created_at,
        }
    }
}

impl From<OrderUpdateRecord> for ArchivedOrderUpdate {
    fn from(r: OrderUpdateRecord) -> Self {
        Self {
            id: r.id,
            timestamp: r.timestamp,
            order_id: r.order_id,
            status: r.status,
            reason: r.reason,
            filled_size: r.filled_size,
            symbol: r.symbol,
            price: r.price,
            size: r.size,
            side: r.side,
            tif: r.tif,
            client_id: r.client_id,
            created_at: r.created_at,
        }
    }
}

// ---------------------------------------------------------------------------
// ArchiverReader
// ---------------------------------------------------------------------------

impl ArchiverReader for DatabaseHandler {
    fn find_safe_boundary(&self, min_age_hours: i64) -> Result<ArchivalBoundary> {
        let mut conn = self.pool().get().context("failed to get connection")?;
        let age_cutoff = Utc::now() - Duration::hours(min_age_hours);

        #[derive(diesel::QueryableByName, Debug)]
        struct BoundaryRow {
            #[diesel(sql_type = BigInt)]
            last_command_id: i64,
        }

        let snapshot_boundary: i64 = sql_query(
            "SELECT COALESCE(\
                 (SELECT last_command_id FROM engine_snapshot_boundary WHERE id = 1), \
                 0\
             ) AS last_command_id",
        )
        .get_result::<BoundaryRow>(&mut conn)
        .map(|r| r.last_command_id)
        .unwrap_or(0);

        if snapshot_boundary <= 0 {
            warn!(
                age_cutoff = %age_cutoff,
                "engine journal archival disabled: no snapshot boundary persisted yet"
            );
            return Ok(ArchivalBoundary {
                max_command_id: 0,
                age_cutoff,
            });
        }

        info!(
            snapshot_boundary,
            age_cutoff = %age_cutoff,
            "archival boundary: engine_commands up to snapshot last_command_id"
        );

        Ok(ArchivalBoundary {
            max_command_id: snapshot_boundary,
            age_cutoff,
        })
    }

    fn fetch_events_batch(
        &self,
        max_command_id: i64,
        age_cutoff: &DateTime<Utc>,
        batch_size: i64,
        after_event_id: Option<i64>,
    ) -> Result<Vec<ArchivedEvent>> {
        let mut conn = self.pool().get().context("failed to get connection")?;

        let mut query = engine_events::table
            .into_boxed()
            .filter(engine_events::command_id.le(max_command_id))
            .filter(engine_events::created_at.lt(age_cutoff));
        if let Some(after_event_id) = after_event_id {
            query = query.filter(engine_events::event_id.gt(after_event_id));
        }

        let rows = query
            .order(engine_events::event_id.asc())
            .limit(batch_size)
            .select(EventRow::as_select())
            .load::<EventRow>(&mut conn)
            .context("failed to fetch engine_events batch")?;

        Ok(rows.into_iter().map(ArchivedEvent::from).collect())
    }

    fn fetch_orphan_commands_batch(
        &self,
        max_command_id: i64,
        age_cutoff: &DateTime<Utc>,
        batch_size: i64,
        after_command_id: Option<i64>,
    ) -> Result<Vec<ArchivedCommand>> {
        let mut conn = self.pool().get().context("failed to get connection")?;

        let rows = sql_query(
            r#"
            SELECT ec.command_id, ec.received_ts_ms, ec.command_data, ec.response_data,
                   ec.order_id, ec.request_uuid, ec.command_type_enum, ec.created_at
            FROM engine_commands ec
            WHERE NOT EXISTS (SELECT 1 FROM engine_events ee WHERE ee.command_id = ec.command_id)
              AND ec.command_id <= $1
              AND ec.created_at < $2
              AND ($3 IS NULL OR ec.command_id > $3)
            ORDER BY ec.command_id ASC
            LIMIT $4
            "#,
        )
        .bind::<BigInt, _>(max_command_id)
        .bind::<diesel::sql_types::Timestamptz, _>(age_cutoff)
        .bind::<Nullable<BigInt>, _>(after_command_id)
        .bind::<BigInt, _>(batch_size)
        .load::<CommandRow>(&mut conn)
        .context("failed to fetch orphan engine_commands batch")?;

        Ok(rows.into_iter().map(ArchivedCommand::from).collect())
    }

    fn fetch_order_actions_batch(
        &self,
        age_cutoff: &DateTime<Utc>,
        batch_size: i64,
        after_id: Option<i32>,
    ) -> Result<Vec<ArchivedOrderAction>> {
        let mut conn = self.pool().get().context("failed to get connection")?;
        let cutoff_naive = age_cutoff.naive_utc();

        let mut query = order_actions::table
            .into_boxed()
            .filter(order_actions::created_at.lt(cutoff_naive));
        if let Some(after_id) = after_id {
            query = query.filter(order_actions::id.gt(Some(after_id)));
        }

        let rows = query
            .order(order_actions::id.asc())
            .limit(batch_size)
            .select(OrderActionRecord::as_select())
            .load::<OrderActionRecord>(&mut conn)
            .context("failed to fetch order_actions batch")?;

        Ok(rows.into_iter().map(ArchivedOrderAction::from).collect())
    }

    fn fetch_order_updates_batch(
        &self,
        age_cutoff: &DateTime<Utc>,
        batch_size: i64,
        after_id: Option<i32>,
    ) -> Result<Vec<ArchivedOrderUpdate>> {
        let mut conn = self.pool().get().context("failed to get connection")?;
        let cutoff_naive = age_cutoff.naive_utc();
        let cutoff_ms = age_cutoff.timestamp_millis();

        let mut query = order_updates::table.into_boxed().filter(
            order_updates::created_at
                .lt(cutoff_naive)
                .or(order_updates::created_at
                    .is_null()
                    .and(order_updates::timestamp.lt(cutoff_ms))),
        );
        if let Some(after_id) = after_id {
            query = query.filter(order_updates::id.gt(Some(after_id)));
        }

        let rows = query
            .order(order_updates::id.asc())
            .limit(batch_size)
            .select(OrderUpdateRecord::as_select())
            .load::<OrderUpdateRecord>(&mut conn)
            .context("failed to fetch order_updates batch")?;

        Ok(rows.into_iter().map(ArchivedOrderUpdate::from).collect())
    }
}

// ---------------------------------------------------------------------------
// ArchiverWriter
// ---------------------------------------------------------------------------

impl ArchiverWriter for DatabaseHandler {
    fn delete_events(&self, event_ids: &[i64]) -> Result<usize> {
        let mut conn = self.pool().get().context("failed to get connection")?;

        let deleted =
            diesel::delete(engine_events::table.filter(engine_events::event_id.eq_any(event_ids)))
                .execute(&mut conn)
                .context("failed to delete engine_events")?;

        Ok(deleted)
    }

    fn delete_commands(&self, command_ids: &[i64]) -> Result<usize> {
        let mut conn = self.pool().get().context("failed to get connection")?;

        let deleted = conn
            .transaction::<_, diesel::result::Error, _>(|conn| {
                // Delete digests first (FK constraint)
                diesel::delete(
                    engine_command_digests::table
                        .filter(engine_command_digests::command_id.eq_any(command_ids)),
                )
                .execute(conn)?;

                diesel::delete(
                    engine_commands::table.filter(engine_commands::command_id.eq_any(command_ids)),
                )
                .execute(conn)
            })
            .context("failed to delete engine_commands")?;

        Ok(deleted)
    }

    fn delete_order_actions(&self, ids: &[i32]) -> Result<usize> {
        let mut conn = self.pool().get().context("failed to get connection")?;

        let deleted = diesel::delete(order_actions::table.filter(order_actions::id.eq_any(ids)))
            .execute(&mut conn)
            .context("failed to delete order_actions")?;

        Ok(deleted)
    }

    fn delete_order_updates(&self, ids: &[i32]) -> Result<usize> {
        let mut conn = self.pool().get().context("failed to get connection")?;

        let deleted = diesel::delete(order_updates::table.filter(order_updates::id.eq_any(ids)))
            .execute(&mut conn)
            .context("failed to delete order_updates")?;

        Ok(deleted)
    }

    fn delete_notifications_before(
        &self,
        cutoff: &DateTime<Utc>,
        batch_size: i64,
    ) -> Result<usize> {
        let mut conn = self.pool().get().context("failed to get connection")?;

        let sql = r#"
            DELETE FROM notifications
            WHERE id IN (
                SELECT id FROM notifications
                WHERE created_at < $1
                ORDER BY id ASC
                LIMIT $2
            )
        "#;
        let deleted = diesel::sql_query(sql)
            .bind::<diesel::sql_types::Timestamptz, _>(cutoff)
            .bind::<BigInt, _>(batch_size)
            .execute(&mut conn)
            .context("failed to delete notifications batch")?;

        Ok(deleted)
    }

    fn delete_notifications_over_per_user_cap(
        &self,
        max_per_user: i64,
        batch_size: i64,
    ) -> Result<usize> {
        let mut conn = self.pool().get().context("failed to get connection")?;

        let sql = r#"
            DELETE FROM notifications
            WHERE id IN (
                SELECT id FROM (
                    SELECT id, ROW_NUMBER() OVER (PARTITION BY wallet_address ORDER BY id DESC) AS rn
                    FROM notifications
                ) sub
                WHERE sub.rn > $1
                LIMIT $2
            )
        "#;
        let deleted = diesel::sql_query(sql)
            .bind::<BigInt, _>(max_per_user)
            .bind::<BigInt, _>(batch_size)
            .execute(&mut conn)
            .context("failed to delete notifications over per-user cap")?;

        Ok(deleted)
    }
}

#[allow(dead_code)]
fn _unused_notifications_schema_guard() {
    let _ = notifications::table;
}

#[cfg(test)]
mod tests {
    use crate::test_helpers::TestDb;
    use chrono::{Duration, Utc};
    use diesel::prelude::*;
    use diesel::sql_query;
    use diesel::sql_types::BigInt;
    use hypercall_db::*;

    /// Helper: upsert an engine_snapshot_boundary row.
    fn seed_snapshot_boundary(test_db: &TestDb, last_command_id: i64) {
        let mut conn = test_db.handler.pool().get().unwrap();
        sql_query(
            "INSERT INTO engine_snapshot_boundary (id, last_command_id, updated_at) \
             VALUES (1, $1, NOW()) \
             ON CONFLICT (id) DO UPDATE SET last_command_id = EXCLUDED.last_command_id, updated_at = NOW()",
        )
        .bind::<BigInt, _>(last_command_id)
        .execute(&mut conn)
        .unwrap();
    }

    /// Helper: insert an engine_command row.
    fn seed_engine_command(test_db: &TestDb, command_id: i64, created_at_hours_ago: i64) {
        let mut conn = test_db.handler.pool().get().unwrap();
        let ts = Utc::now() - Duration::hours(created_at_hours_ago);
        sql_query(
            "INSERT INTO engine_commands (command_id, received_ts_ms, request_uuid, created_at, command_type_enum) \
             VALUES ($1, $2, gen_random_uuid(), $3, 'CreateOrder')",
        )
        .bind::<BigInt, _>(command_id)
        .bind::<BigInt, _>(command_id * 1000)
        .bind::<diesel::sql_types::Timestamptz, _>(ts)
        .execute(&mut conn)
        .unwrap();
    }

    /// Helper: insert an engine_event row.
    /// Uses `event_id` as the `event_idx` to avoid unique constraint violations
    /// when inserting multiple events for the same command_id.
    fn seed_engine_event(
        test_db: &TestDb,
        event_id: i64,
        command_id: i64,
        created_at_hours_ago: i64,
    ) {
        let mut conn = test_db.handler.pool().get().unwrap();
        let ts = Utc::now() - Duration::hours(created_at_hours_ago);
        sql_query(
            "INSERT INTO engine_events (event_id, command_id, event_idx, event_data, created_at, publish_attempts, event_type_enum) \
             VALUES ($1, $2, $1::int, E'\\\\x00', $3, 0, 'OrderAction')",
        )
        .bind::<BigInt, _>(event_id)
        .bind::<BigInt, _>(command_id)
        .bind::<diesel::sql_types::Timestamptz, _>(ts)
        .execute(&mut conn)
        .unwrap();
    }

    /// Helper: insert an engine_command_digest row.
    fn seed_command_digest(test_db: &TestDb, command_id: i64) {
        let mut conn = test_db.handler.pool().get().unwrap();
        sql_query(
            "INSERT INTO engine_command_digests (command_id, duration_ms, created_at, pre_digest_data, post_digest_data) \
             VALUES ($1, 10, NOW(), E'\\\\x00', E'\\\\x00')",
        )
        .bind::<BigInt, _>(command_id)
        .execute(&mut conn)
        .unwrap();
    }

    /// Helper: insert an order_actions row.
    fn seed_order_action(test_db: &TestDb, created_at_hours_ago: i64) -> i32 {
        let mut conn = test_db.handler.pool().get().unwrap();
        let ts = (Utc::now() - Duration::hours(created_at_hours_ago)).naive_utc();
        #[derive(diesel::QueryableByName)]
        struct IdRow {
            #[diesel(sql_type = diesel::sql_types::Integer)]
            id: i32,
        }
        let row = sql_query(
            "INSERT INTO order_actions (timestamp, wallet, action, symbol, price, size, side, tif, created_at) \
             VALUES (1000, E'\\\\x0000000000000000000000000000000000000001', 'place', 'BTC-25DEC26-100000-C', 0.05, 1, 'buy', 'GTC', $1) \
             RETURNING id",
        )
        .bind::<diesel::sql_types::Timestamp, _>(ts)
        .get_result::<IdRow>(&mut conn)
        .unwrap();
        row.id
    }

    /// Helper: insert an order_updates row.
    fn seed_order_update(test_db: &TestDb, created_at_hours_ago: i64) -> i32 {
        let mut conn = test_db.handler.pool().get().unwrap();
        let ts = (Utc::now() - Duration::hours(created_at_hours_ago)).naive_utc();
        #[derive(diesel::QueryableByName)]
        struct IdRow {
            #[diesel(sql_type = diesel::sql_types::Integer)]
            id: i32,
        }
        let row = sql_query(
            "INSERT INTO order_updates (timestamp, status, filled_size, symbol, price, size, side, tif, created_at) \
             VALUES (1000, 'FILLED', 1, 'BTC-25DEC26-100000-C', 0.05, 1, 'buy', 'GTC', $1) \
             RETURNING id",
        )
        .bind::<diesel::sql_types::Timestamp, _>(ts)
        .get_result::<IdRow>(&mut conn)
        .unwrap();
        row.id
    }

    /// Helper: insert a notifications row.
    fn seed_notification(test_db: &TestDb, wallet: &str, created_at_hours_ago: i64) -> i64 {
        let mut conn = test_db.handler.pool().get().unwrap();
        let ts = Utc::now() - Duration::hours(created_at_hours_ago);
        #[derive(diesel::QueryableByName)]
        struct IdRow {
            #[diesel(sql_type = diesel::sql_types::BigInt)]
            id: i64,
        }
        let row = sql_query(
            "INSERT INTO notifications (wallet_address, notification_type, payload, created_at) \
             VALUES ($1, 'order_filled', E'\\\\x00', $2) \
             RETURNING id",
        )
        .bind::<diesel::sql_types::Text, _>(wallet)
        .bind::<diesel::sql_types::Timestamptz, _>(ts)
        .get_result::<IdRow>(&mut conn)
        .unwrap();
        row.id
    }

    // =========================================================================
    // ArchiverReader tests
    // =========================================================================

    #[tokio::test]
    async fn find_safe_boundary_returns_snapshot_boundary() {
        let test_db = TestDb::new().await.unwrap();
        seed_snapshot_boundary(&test_db, 500);

        let reader: &dyn ArchiverReader = test_db.handler.as_ref();
        let boundary = reader.find_safe_boundary(24).unwrap();
        assert_eq!(boundary.max_command_id, 500);
    }

    #[tokio::test]
    async fn find_safe_boundary_no_snapshot_returns_zero() {
        let test_db = TestDb::new().await.unwrap();
        let reader: &dyn ArchiverReader = test_db.handler.as_ref();
        let boundary = reader.find_safe_boundary(24).unwrap();
        assert_eq!(boundary.max_command_id, 0);
    }

    #[tokio::test]
    async fn fetch_events_batch_filters_by_command_id_and_age() {
        let test_db = TestDb::new().await.unwrap();
        // Command 1, old enough (48h ago)
        seed_engine_command(&test_db, 1, 48);
        seed_engine_event(&test_db, 1, 1, 48);
        // Command 2, old enough but command_id too high
        seed_engine_command(&test_db, 100, 48);
        seed_engine_event(&test_db, 2, 100, 48);
        // Command 3, recent (not old enough)
        seed_engine_command(&test_db, 3, 1);
        seed_engine_event(&test_db, 3, 3, 1);

        let reader: &dyn ArchiverReader = test_db.handler.as_ref();
        let age_cutoff = Utc::now() - Duration::hours(24);
        let batch = reader
            .fetch_events_batch(50, &age_cutoff, 100, None)
            .unwrap();
        // Should only get event_id=1 (command_id=1 <= 50, and 48h > 24h cutoff)
        assert_eq!(batch.len(), 1);
        assert_eq!(batch[0].event_id, 1);
        assert_eq!(batch[0].command_id, 1);
    }

    #[tokio::test]
    async fn fetch_events_batch_respects_after_event_id() {
        let test_db = TestDb::new().await.unwrap();
        seed_engine_command(&test_db, 1, 48);
        seed_engine_event(&test_db, 10, 1, 48);
        seed_engine_event(&test_db, 20, 1, 48);

        let reader: &dyn ArchiverReader = test_db.handler.as_ref();
        let age_cutoff = Utc::now() - Duration::hours(24);
        let batch = reader
            .fetch_events_batch(100, &age_cutoff, 100, Some(10))
            .unwrap();
        assert_eq!(batch.len(), 1);
        assert_eq!(batch[0].event_id, 20);
    }

    #[tokio::test]
    async fn delete_events_removes_matching_rows() {
        let test_db = TestDb::new().await.unwrap();
        seed_engine_command(&test_db, 1, 48);
        seed_engine_event(&test_db, 10, 1, 48);
        seed_engine_event(&test_db, 20, 1, 48);

        let writer: &dyn ArchiverWriter = test_db.handler.as_ref();
        let deleted = writer.delete_events(&[10]).unwrap();
        assert_eq!(deleted, 1);

        // Verify only event 20 remains
        let reader: &dyn ArchiverReader = test_db.handler.as_ref();
        let age_cutoff = Utc::now() - Duration::hours(1);
        let remaining = reader
            .fetch_events_batch(100, &age_cutoff, 100, None)
            .unwrap();
        assert_eq!(remaining.len(), 1);
        assert_eq!(remaining[0].event_id, 20);
    }

    #[tokio::test]
    async fn fetch_orphan_commands_batch_returns_commands_without_events() {
        let test_db = TestDb::new().await.unwrap();
        // Command with events (not orphan)
        seed_engine_command(&test_db, 1, 48);
        seed_engine_event(&test_db, 1, 1, 48);
        // Command without events (orphan)
        seed_engine_command(&test_db, 2, 48);

        let reader: &dyn ArchiverReader = test_db.handler.as_ref();
        let age_cutoff = Utc::now() - Duration::hours(24);
        let batch = reader
            .fetch_orphan_commands_batch(100, &age_cutoff, 100, None)
            .unwrap();
        assert_eq!(batch.len(), 1);
        assert_eq!(batch[0].command_id, 2);
    }

    #[tokio::test]
    async fn delete_commands_cascades_to_digests() {
        let test_db = TestDb::new().await.unwrap();
        seed_engine_command(&test_db, 5, 48);
        seed_command_digest(&test_db, 5);

        let writer: &dyn ArchiverWriter = test_db.handler.as_ref();
        let deleted = writer.delete_commands(&[5]).unwrap();
        assert_eq!(deleted, 1);

        // Verify command is gone
        let reader: &dyn ArchiverReader = test_db.handler.as_ref();
        let age_cutoff = Utc::now() - Duration::hours(1);
        let orphans = reader
            .fetch_orphan_commands_batch(100, &age_cutoff, 100, None)
            .unwrap();
        assert!(orphans.is_empty());
    }

    #[tokio::test]
    async fn fetch_order_actions_batch_filters_by_age() {
        let test_db = TestDb::new().await.unwrap();
        let _old_id = seed_order_action(&test_db, 48);
        let _new_id = seed_order_action(&test_db, 1);

        let reader: &dyn ArchiverReader = test_db.handler.as_ref();
        let age_cutoff = Utc::now() - Duration::hours(24);
        let batch = reader
            .fetch_order_actions_batch(&age_cutoff, 100, None)
            .unwrap();
        assert_eq!(batch.len(), 1);
    }

    #[tokio::test]
    async fn delete_order_actions_removes_rows() {
        let test_db = TestDb::new().await.unwrap();
        let id = seed_order_action(&test_db, 48);

        let writer: &dyn ArchiverWriter = test_db.handler.as_ref();
        let deleted = writer.delete_order_actions(&[id]).unwrap();
        assert_eq!(deleted, 1);

        // Verify gone
        let reader: &dyn ArchiverReader = test_db.handler.as_ref();
        let age_cutoff = Utc::now() - Duration::hours(1);
        let remaining = reader
            .fetch_order_actions_batch(&age_cutoff, 100, None)
            .unwrap();
        assert!(remaining.is_empty());
    }

    #[tokio::test]
    async fn fetch_order_updates_batch_filters_by_age() {
        let test_db = TestDb::new().await.unwrap();
        let _old_id = seed_order_update(&test_db, 48);
        let _new_id = seed_order_update(&test_db, 1);

        let reader: &dyn ArchiverReader = test_db.handler.as_ref();
        let age_cutoff = Utc::now() - Duration::hours(24);
        let batch = reader
            .fetch_order_updates_batch(&age_cutoff, 100, None)
            .unwrap();
        assert_eq!(batch.len(), 1);
    }

    #[tokio::test]
    async fn delete_order_updates_removes_rows() {
        let test_db = TestDb::new().await.unwrap();
        let id = seed_order_update(&test_db, 48);

        let writer: &dyn ArchiverWriter = test_db.handler.as_ref();
        let deleted = writer.delete_order_updates(&[id]).unwrap();
        assert_eq!(deleted, 1);

        let reader: &dyn ArchiverReader = test_db.handler.as_ref();
        let age_cutoff = Utc::now() - Duration::hours(1);
        let remaining = reader
            .fetch_order_updates_batch(&age_cutoff, 100, None)
            .unwrap();
        assert!(remaining.is_empty());
    }

    #[tokio::test]
    async fn delete_notifications_before_removes_old_rows() {
        let test_db = TestDb::new().await.unwrap();
        let _old = seed_notification(&test_db, "0xabc", 48);
        let _new = seed_notification(&test_db, "0xabc", 1);

        let writer: &dyn ArchiverWriter = test_db.handler.as_ref();
        let cutoff = Utc::now() - Duration::hours(24);
        let deleted = writer.delete_notifications_before(&cutoff, 100).unwrap();
        assert_eq!(deleted, 1);
    }

    #[tokio::test]
    async fn delete_notifications_over_per_user_cap_enforces_limit() {
        let test_db = TestDb::new().await.unwrap();
        // Insert 5 notifications for same wallet
        for i in 0..5 {
            seed_notification(&test_db, "0xwallet1", i);
        }

        let writer: &dyn ArchiverWriter = test_db.handler.as_ref();
        // Cap at 2 per user: should delete 3
        let deleted = writer
            .delete_notifications_over_per_user_cap(2, 100)
            .unwrap();
        assert_eq!(deleted, 3);
    }
}