rhei 2.0.0

//! Rhei — lightweight serverless HTAP engine for Rust.
//!
//! Rhei pairs **Rusqlite** (SQLite) for low-latency OLTP writes with a
//! pluggable OLAP backend (**Apache DataFusion** or **DuckDB**) for analytical
//! queries, bridged by trigger-based Change Data Capture (CDC) replication and
//! automatic SQL query routing. No separate server process is required — the
//! engine runs fully in-process.
//!
//! # Architecture
//!
//! ```text
//! Standard HTAP mode:
//!
//! Client
//!   |
//!   v
//! HtapEngine  (facade — this crate)
//!   |-- SqlParserRouter  ──► AST-based routing to OLTP or OLAP
//!   |-- RusqliteEngine   ──► WAL-mode SQLite, write conn + read pool
//!   |     └── RusqliteCdcProducer  ──► trigger-based _rhei_cdc_log
//!   |-- OlapBackend      ──► DuckDB or DataFusion (feature-gated)
//!   └── CdcSyncEngine    ──► polls CDC, applies DML to OLAP
//!
//! Sidecar mode:
//!
//! External DB (SQLite / PostgreSQL)
//!   │  polls by updated_at > watermark
//!   ▼
//! TimestampCdcConsumer<S: SourceConnector>
//!   │  CdcEvent stream
//!   ▼
//! CdcSyncEngine  (temporal or destructive)
//!   │  DML -> OLAP
//!   ▼
//! OlapBackend (DuckDB / DataFusion)
//! ```
//!
//! # Operating modes
//!
//! ## Standard HTAP mode
//!
//! The default mode. A local SQLite database handles writes; CDC triggers
//! capture every INSERT / UPDATE / DELETE into `_rhei_cdc_log`. A background
//! sync loop (or manual [`HtapEngine::sync_now`]) applies those events to the
//! OLAP engine so analytical queries see fresh data.
//!
//! ## Sidecar mode
//!
//! When [`HtapConfig::sidecar`] is `Some(SidecarConfig { .. })`, the engine
//! *follows* an external database (SQLite or PostgreSQL) via timestamp-based
//! CDC polling. No local SQLite write path is needed unless
//! [`SidecarConfig::enable_local_oltp`] is `true`. Useful for adding OLAP
//! capabilities to an existing application database without schema changes.
//!
//! # Sync modes
//!
//! | Mode | Behaviour |
//! |------|-----------|
//! | [`rhei_core::SyncMode::Destructive`] | Mirror semantics: UPDATE overwrites the row, DELETE removes it. Default. |
//! | [`rhei_core::SyncMode::Temporal`] | SCD Type 2: every change appends a new version with `_rhei_valid_from` / `_rhei_valid_to` / `_rhei_operation` columns. Enables point-in-time queries. |
//!
//! # Feature flags
//!
//! | Feature | Default | Description |
//! |---------|---------|-------------|
//! | `datafusion-backend` | **yes** | Apache DataFusion OLAP engine (pure Rust, natively async) |
//! | `duckdb-backend` | no | DuckDB OLAP engine (C++ bundled via `spawn_blocking`) |
//! | `full` | no | Both OLAP backends simultaneously |
//! | `sidecar` | no | Timestamp-based CDC from SQLite or PostgreSQL external sources |
//! | `rocksdb-cdc` | no | RocksDB-backed durable CDC log (5–7× faster writes than SQLite triggers) |
//! | `flight-sql` | no | Arrow Flight SQL gRPC server (see `rhei-flight` crate) |
//! | `metrics` | no | Counters/gauges emitted via the `metrics` crate facade |
//! | `metrics-exporter` | no | Prometheus HTTP endpoint (configured via `metrics_port` in TOML) |
//! | `cloud-storage` | no | S3 / GCS object-store backends for DataFusion Parquet storage |
//!
//! # Quick start
//!
//! ```rust,no_run
//! use rhei::{HtapConfig, HtapEngine, TableSchema};
//! use arrow::datatypes::{DataType, Field, Schema};
//! use std::sync::Arc;
//!
//! #[tokio::main]
//! async fn main() -> Result<(), Box<dyn std::error::Error>> {
//!     // 1. Configure the engine (in-memory OLAP, default DataFusion backend)
//!     let config = HtapConfig {
//!         oltp_path: "my_app.db".to_string(),
//!         olap_in_memory: true,
//!         ..HtapConfig::default()
//!     };
//!
//!     // 2. Start the engine
//!     let mut engine = HtapEngine::new(config).await?;
//!
//!     // 3. Register a table for HTAP replication
//!     let schema = TableSchema {
//!         name: "events".to_string(),
//!         arrow_schema: Arc::new(Schema::new(vec![
//!             Field::new("id",    DataType::Int64,  false),
//!             Field::new("label", DataType::Utf8,   true),
//!         ])),
//!         primary_key: vec!["id".to_string()],
//!     };
//!     engine.register_table(schema).await?;
//!
//!     // 4. Write via OLTP (auto-routed by SqlParserRouter)
//!     engine.execute("INSERT INTO events VALUES (1, 'hello')", &[]).await?;
//!
//!     // 5. Sync CDC events to OLAP
//!     engine.sync_now().await?;
//!
//!     // 6. Query via OLAP (aggregate → routed to DataFusion)
//!     let batches = engine.query("SELECT COUNT(*) FROM events").await?;
//!     println!("{} batches returned", batches.len());
//!
//!     // 7. Shut down cleanly
//!     engine.shutdown().await;
//!     Ok(())
//! }
//! ```
//!
//! # Key types
//!
//! - [`HtapEngine`] — the main engine facade; start here.
//! - [`HtapConfig`] — builder-style configuration.
//! - [`SidecarConfig`] — sidecar-specific settings (watermark persistence,
//!   delete detection, OLTP toggle).
//! - [`CdcSource`] — unified CDC consumer wrapper used internally by the sync
//!   engine.

pub mod error;

use std::sync::Arc;
use std::time::Duration;

use arrow::record_batch::RecordBatch;
use tokio_util::sync::CancellationToken;
use tracing::{debug, info, warn};

pub use error::HtapError;

// Re-export core types for convenience
pub use rhei_core::types::*;
pub use rhei_core::{
    CdcConsumer, OlapEngine, OltpEngine, QueryRouter, SchemaRegistry, SyncEngine, TableSchema,
};

// Re-export OLAP dispatcher and backend types
#[cfg(feature = "datafusion-backend")]
pub use rhei_olap::{DataFusionEngine, SharedDataFusionEngine, StorageMode};
#[cfg(feature = "duckdb-backend")]
pub use rhei_olap::{DuckDbEngine, SharedDuckDbEngine};
pub use rhei_olap::{OlapBackend, OlapError, OltpBackend, OltpError};
pub use rhei_oltp_rusqlite::{RusqliteCdcProducer, RusqliteEngine, RusqliteOltpError};

// Re-export sync components
pub use rhei_sync::{
    spawn_sync_loop, temporalize_schema, CdcSyncEngine, HeuristicRouter, SqlParserRouter,
};

// Re-export sidecar components
#[cfg(feature = "sidecar")]
pub use rhei_sidecar::{
    DeleteDetection, NullWatermarkStore, RocksDbWatermarkStore, SidecarError, SourceConnector,
    TimestampCdcConfig, TimestampCdcConsumer, TimestampTableConfig, WatermarkStore,
};

/// Concrete sidecar consumer type for SQLite sources (used by the facade).
#[cfg(feature = "sidecar")]
pub type SqliteTimestampConsumer =
    rhei_sidecar::TimestampCdcConsumer<connector_arrow::rusqlite::SQLiteConnection>;

/// Concrete sidecar consumer type for PostgreSQL sources (used by the facade).
#[cfg(feature = "sidecar")]
pub type PostgresTimestampConsumer =
    rhei_sidecar::TimestampCdcConsumer<connector_arrow::postgres::PostgresConnection>;

/// Source database for the sidecar CDC consumer.
#[cfg(feature = "sidecar")]
pub enum SidecarSource {
    /// Path to a SQLite database file.
    Sqlite(String),
    /// PostgreSQL connection string (e.g. "host=localhost user=postgres dbname=mydb").
    Postgres(String),
}

// Re-export RocksDB CDC components
#[cfg(feature = "rocksdb-cdc")]
pub use rhei_cdc_rocksdb::{RocksDbCdcConfig, RocksDbCdcError, RocksDbCdcLog};

// OlapBackend is now provided by rhei-olap (re-exported above)

// ---------------------------------------------------------------------------
// CdcSource — unified CDC consumer wrapper (feature-gated)
// ---------------------------------------------------------------------------

/// When the `sidecar` feature is disabled, CdcSource wraps the Rusqlite CDC producer.
#[cfg(not(feature = "sidecar"))]
pub enum CdcSource {
    /// Rusqlite-based CDC (trigger).
    Rusqlite(RusqliteCdcProducer),
    /// RocksDB-backed CDC log (standalone, append-only).
    #[cfg(feature = "rocksdb-cdc")]
    RocksDb(Arc<RocksDbCdcLog>),
    /// Bridge mode: SQLite triggers fire into `_rhei_cdc_log`, a bridge consumer
    /// moves events into RocksDB for durable buffering, and the sync engine reads
    /// FROM RocksDB. Provides crash durability: events survive a mid-cycle crash.
    #[cfg(feature = "rocksdb-cdc")]
    RocksDbBridge {
        /// The Rusqlite CDC producer that reads from `_rhei_cdc_log`.
        sqlite: RusqliteCdcProducer,
        /// The RocksDB log into which SQLite events are durably moved.
        rocksdb: Arc<RocksDbCdcLog>,
    },
}

#[cfg(not(feature = "sidecar"))]
impl CdcConsumer for CdcSource {
    type Error = HtapError;

    async fn poll(
        &self,
        after_seq: Option<i64>,
        limit: u32,
    ) -> Result<Vec<rhei_core::types::CdcEvent>, Self::Error> {
        match self {
            Self::Rusqlite(p) => p.poll(after_seq, limit).await.map_err(HtapError::from),
            #[cfg(feature = "rocksdb-cdc")]
            Self::RocksDb(p) => p.poll(after_seq, limit).await.map_err(HtapError::from),
            #[cfg(feature = "rocksdb-cdc")]
            Self::RocksDbBridge { sqlite, rocksdb } => {
                bridge_poll(sqlite, rocksdb, after_seq, limit).await
            }
        }
    }

    async fn latest_seq(&self) -> Result<Option<i64>, Self::Error> {
        match self {
            Self::Rusqlite(p) => p.latest_seq().await.map_err(HtapError::from),
            #[cfg(feature = "rocksdb-cdc")]
            Self::RocksDb(p) => p.latest_seq().await.map_err(HtapError::from),
            #[cfg(feature = "rocksdb-cdc")]
            Self::RocksDbBridge { sqlite, rocksdb } => bridge_latest_seq(sqlite, rocksdb).await,
        }
    }

    async fn prune(&self, up_to_seq: i64) -> Result<u64, Self::Error> {
        match self {
            Self::Rusqlite(p) => p.prune(up_to_seq).await.map_err(HtapError::from),
            #[cfg(feature = "rocksdb-cdc")]
            Self::RocksDb(p) => p.prune(up_to_seq).await.map_err(HtapError::from),
            #[cfg(feature = "rocksdb-cdc")]
            Self::RocksDbBridge { rocksdb, .. } => {
                rocksdb.prune(up_to_seq).await.map_err(HtapError::from)
            }
        }
    }
}

/// When the `sidecar` feature is enabled, CdcSource wraps both standard and sidecar consumers.
#[cfg(feature = "sidecar")]
pub enum CdcSource {
    /// Rusqlite-based CDC (trigger).
    Rusqlite(RusqliteCdcProducer),
    /// Sidecar: timestamp-based polling of an external SQLite source.
    SidecarSqlite(SqliteTimestampConsumer),
    /// Sidecar: timestamp-based polling of an external PostgreSQL source.
    SidecarPostgres(PostgresTimestampConsumer),
    /// RocksDB-backed CDC log (standalone, append-only).
    #[cfg(feature = "rocksdb-cdc")]
    RocksDb(Arc<RocksDbCdcLog>),
    /// Bridge mode: SQLite triggers fire into `_rhei_cdc_log`, a bridge consumer
    /// moves events into RocksDB for durable buffering, and the sync engine reads
    /// FROM RocksDB. Provides crash durability: events survive a mid-cycle crash.
    #[cfg(feature = "rocksdb-cdc")]
    RocksDbBridge {
        /// The Rusqlite CDC producer that reads from `_rhei_cdc_log`.
        sqlite: RusqliteCdcProducer,
        /// The RocksDB log into which SQLite events are durably moved.
        rocksdb: Arc<RocksDbCdcLog>,
    },
}

#[cfg(feature = "sidecar")]
impl CdcConsumer for CdcSource {
    type Error = HtapError;

    async fn poll(
        &self,
        after_seq: Option<i64>,
        limit: u32,
    ) -> Result<Vec<rhei_core::types::CdcEvent>, Self::Error> {
        match self {
            Self::Rusqlite(p) => p.poll(after_seq, limit).await.map_err(HtapError::from),
            Self::SidecarSqlite(p) => p.poll(after_seq, limit).await.map_err(HtapError::from),
            Self::SidecarPostgres(p) => p.poll(after_seq, limit).await.map_err(HtapError::from),
            #[cfg(feature = "rocksdb-cdc")]
            Self::RocksDb(p) => p.poll(after_seq, limit).await.map_err(HtapError::from),
            #[cfg(feature = "rocksdb-cdc")]
            Self::RocksDbBridge { sqlite, rocksdb } => {
                bridge_poll(sqlite, rocksdb, after_seq, limit).await
            }
        }
    }

    async fn latest_seq(&self) -> Result<Option<i64>, Self::Error> {
        match self {
            Self::Rusqlite(p) => p.latest_seq().await.map_err(HtapError::from),
            Self::SidecarSqlite(p) => p.latest_seq().await.map_err(HtapError::from),
            Self::SidecarPostgres(p) => p.latest_seq().await.map_err(HtapError::from),
            #[cfg(feature = "rocksdb-cdc")]
            Self::RocksDb(p) => p.latest_seq().await.map_err(HtapError::from),
            #[cfg(feature = "rocksdb-cdc")]
            Self::RocksDbBridge { sqlite, rocksdb } => bridge_latest_seq(sqlite, rocksdb).await,
        }
    }

    async fn prune(&self, up_to_seq: i64) -> Result<u64, Self::Error> {
        match self {
            Self::Rusqlite(p) => p.prune(up_to_seq).await.map_err(HtapError::from),
            Self::SidecarSqlite(p) => p.prune(up_to_seq).await.map_err(HtapError::from),
            Self::SidecarPostgres(p) => p.prune(up_to_seq).await.map_err(HtapError::from),
            #[cfg(feature = "rocksdb-cdc")]
            Self::RocksDb(p) => p.prune(up_to_seq).await.map_err(HtapError::from),
            #[cfg(feature = "rocksdb-cdc")]
            Self::RocksDbBridge { rocksdb, .. } => {
                rocksdb.prune(up_to_seq).await.map_err(HtapError::from)
            }
        }
    }
}

// ---------------------------------------------------------------------------
// Bridge helpers (shared across both CdcSource feature-gate variants)
// ---------------------------------------------------------------------------

/// Compute an *effective* latest sequence number for the bridge that accounts
/// for SQLite CDC events that have not yet been moved into RocksDB.
///
/// SQLite seqs and RocksDB seqs live in different namespaces, so we cannot
/// return a single meaningful seq that covers both. Instead we return:
///
///   `rocksdb_latest_seq + pending_sqlite_count`
///
/// where `pending_sqlite_count` is `sqlite.latest_seq - bridge_watermark`
/// (the number of rows in `_rhei_cdc_log` that are newer than the last bridged
/// watermark). This overestimates the RocksDB seq for the pending events but
/// produces a correct **nonzero** lag signal: `CdcSyncEngine::status()` computes
/// `lag = latest_seq - last_synced_seq`, so as long as pending events inflate
/// `latest_seq`, the dashboard correctly shows pending work.
///
/// After a full bridge+sync cycle the pending count drops to zero and the RocksDB
/// seq equals the last synced seq, giving `lag == 0` as expected.
///
/// # Note
/// The semantic is intentionally an approximation. The concrete RocksDB sequence
/// numbers for un-bridged events are only assigned at the next `poll` cycle.
#[cfg(feature = "rocksdb-cdc")]
async fn bridge_latest_seq(
    sqlite: &RusqliteCdcProducer,
    rocksdb: &Arc<RocksDbCdcLog>,
) -> Result<Option<i64>, HtapError> {
    use rhei_core::CdcConsumer as _;

    let rocksdb_seq = rocksdb.latest_seq().await.map_err(HtapError::from)?;
    let sqlite_latest = sqlite.latest_seq().await.map_err(HtapError::from)?;

    // How many SQLite rows are ahead of the watermark?
    let pending_count: i64 = match sqlite_latest {
        None => 0,
        Some(s) => {
            let watermark = rocksdb
                .get_bridge_watermark()
                .map_err(HtapError::from)?
                .unwrap_or(0);
            (s - watermark).max(0)
        }
    };

    if pending_count == 0 {
        return Ok(rocksdb_seq);
    }

    // Inflate the RocksDB seq by the number of pending SQLite events.
    let base = rocksdb_seq.unwrap_or(0);
    Ok(Some(base + pending_count))
}

/// Bridge poll: drain new events from SQLite CDC into RocksDB, then serve
/// the caller's request from RocksDB.
///
/// The dual-seq problem is handled as follows:
///   1. We read the persisted bridge watermark from RocksDB. On the first call
///      this is `None`; on subsequent calls it is the max SQLite seq that was
///      already bridged in a previous cycle.
///   2. We poll SQLite with `after_seq = bridge_watermark` so we only fetch
///      rows that have **not yet** been moved to RocksDB. This makes the bridge
///      idempotent: if a previous cycle wrote events but crashed before pruning
///      SQLite, the watermark still reflects what is in RocksDB, so we skip
///      those rows on the next cycle.
///   3. We capture the *max SQLite seq* before calling the atomic write because
///      the write overwrites every `event.seq` with RocksDB's counter.
///   4. We atomically write the new events **and** the updated bridge watermark
///      into RocksDB in a single `WriteBatch`. If the process crashes here,
///      either both land or neither does, preserving idempotency.
///   5. After the atomic write, we best-effort prune the SQLite log. If pruning
///      fails the bridge watermark already guards against re-bridging the same
///      rows, so the only downside is a larger `_rhei_cdc_log` table.
///   6. We serve the caller from RocksDB using their `after_seq` (a RocksDB
///      sequence number). The sync engine's watermark is always a RocksDB
///      sequence number after the first bridging cycle.
#[cfg(feature = "rocksdb-cdc")]
async fn bridge_poll(
    sqlite: &RusqliteCdcProducer,
    rocksdb: &Arc<RocksDbCdcLog>,
    after_seq: Option<i64>,
    limit: u32,
) -> Result<Vec<rhei_core::types::CdcEvent>, HtapError> {
    use rhei_core::CdcConsumer as _;

    // Step 1: read the persisted bridge watermark so we know which SQLite rows
    // are already in RocksDB from previous cycles.
    let bridge_watermark = rocksdb.get_bridge_watermark().map_err(HtapError::from)?;

    // Step 2: poll only rows that have NOT yet been bridged.
    let mut new_events = sqlite
        .poll(bridge_watermark, limit)
        .await
        .map_err(HtapError::from)?;

    if !new_events.is_empty() {
        // Step 3: capture SQLite seqs before they are overwritten by the atomic write.
        let max_sqlite_seq = new_events.iter().map(|e| e.seq).max().unwrap_or(0);

        // Step 4: atomically write events + updated watermark into RocksDB.
        // On crash mid-write, either both land or neither does.
        rocksdb
            .append_and_set_bridge_watermark(&mut new_events, max_sqlite_seq)
            .map_err(HtapError::from)?;

        // Step 5: prune the SQLite log (best-effort — the watermark already
        // prevents re-bridging on the next cycle even if this fails).
        if let Err(e) = sqlite.prune(max_sqlite_seq).await {
            warn!(
                max_sqlite_seq,
                error = %e,
                "bridge_poll: sqlite.prune failed (best-effort); \
                 bridge watermark is committed, events will not be re-bridged"
            );
        }
    }

    // Step 6: serve the caller from RocksDB using their watermark (RocksDB seqs).
    rocksdb
        .poll(after_seq, limit)
        .await
        .map_err(HtapError::from)
}

type SyncEngineType = CdcSyncEngine<CdcSource, OlapBackend>;

/// Selects the OLAP query engine used by the [`HtapEngine`].
///
/// Only variants whose corresponding feature flag is enabled are available at
/// compile time. The `datafusion-backend` feature is on by default, so
/// [`OlapBackendType::DataFusion`] is always reachable with the default feature
/// set.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum OlapBackendType {
    /// DuckDB — embeds the C++ DuckDB library and dispatches all operations
    /// through `spawn_blocking`. Offers excellent out-of-the-box SQL
    /// compatibility and MVCC-based concurrent reads.
    ///
    /// Available on crate feature `duckdb-backend` only.
    #[cfg(feature = "duckdb-backend")]
    DuckDb,
    /// Apache DataFusion — pure-Rust, natively async OLAP engine. The default
    /// when the `datafusion-backend` feature is enabled.
    ///
    /// Available on crate feature `datafusion-backend` only.
    #[cfg(feature = "datafusion-backend")]
    DataFusion,
}

/// Configuration for the [`HtapEngine`].
///
/// Construct via [`HtapConfig::default`] and override the fields you care
/// about. All fields have sensible defaults suitable for development and
/// testing.
///
/// # Example
///
/// ```rust,no_run
/// use rhei::HtapConfig;
/// use std::time::Duration;
///
/// let config = HtapConfig {
///     oltp_path: "/var/lib/myapp/oltp.db".to_string(),
///     olap_in_memory: false,
///     olap_path: Some("/var/lib/myapp/olap".to_string()),
///     sync_interval: Some(Duration::from_millis(200)),
///     ..HtapConfig::default()
/// };
/// ```
pub struct HtapConfig {
    /// Path to the local Rusqlite (SQLite) database file used for OLTP writes.
    ///
    /// The file is created if it does not exist. WAL mode and
    /// `PRAGMA synchronous=NORMAL` are applied automatically.
    pub oltp_path: String,

    /// `true` (default) to keep the OLAP engine entirely in memory.
    ///
    /// In-memory mode is fastest and ideal for tests and short-lived processes.
    /// Set to `false` and provide [`olap_path`][Self::olap_path] for durable
    /// analytical storage across restarts.
    pub olap_in_memory: bool,

    /// Filesystem path for the OLAP engine's durable storage.
    ///
    /// Ignored when [`olap_in_memory`][Self::olap_in_memory] is `true`. For
    /// DuckDB this is a `.duckdb` file; for DataFusion it is a directory used
    /// by the configured [`datafusion_storage`][Self::datafusion_storage] mode.
    pub olap_path: Option<String>,

    /// Maximum number of CDC events consumed per sync cycle (default: 1000).
    ///
    /// Larger batches amortise per-transaction overhead and increase throughput
    /// at the cost of higher per-cycle latency. For append-heavy workloads,
    /// values between 5 000 and 50 000 are typical.
    pub sync_batch_size: u32,

    /// Whether to delete processed CDC events from `_rhei_cdc_log` after each
    /// successful sync cycle (default: `true`).
    ///
    /// Set to `false` only for debugging or when you need to replay events.
    /// Leaving this `true` prevents unbounded growth of the CDC log table.
    pub prune_after_sync: bool,

    /// When `Some(duration)`, `HtapEngine::new` automatically spawns a
    /// background sync loop that polls CDC at the given interval.
    ///
    /// Choose the interval based on your freshness-vs-overhead trade-off:
    /// - **≤ 50 ms** — near-real-time; adds measurable SQLite I/O pressure.
    /// - **100–500 ms** — balanced; recommended for most HTAP workloads.
    /// - **≥ 1 s** — low overhead; suitable for batch-oriented analytics.
    ///
    /// `None` (default) means you drive sync manually via [`HtapEngine::sync_now`].
    pub sync_interval: Option<Duration>,

    /// Which OLAP backend to instantiate (default: DataFusion).
    pub olap_backend: OlapBackendType,

    /// Number of read connections in the OLTP connection pool (default: 4).
    ///
    /// Increase for workloads with many concurrent analytical read-backs to
    /// OLTP. Values below 1 are clamped to 1 internally.
    pub read_pool_size: usize,

    /// How CDC events are applied to the OLAP mirror (default:
    /// [`rhei_core::SyncMode::Destructive`]).
    ///
    /// Use [`rhei_core::SyncMode::Temporal`] (SCD Type 2) when you need
    /// point-in-time queries: every change appends a new row with
    /// `_rhei_valid_from` / `_rhei_valid_to` / `_rhei_operation` metadata
    /// instead of overwriting the previous version.
    pub sync_mode: SyncMode,

    /// DataFusion storage backend (default: `StorageMode::InMemory`).
    ///
    /// Controls how DataFusion persists data between queries:
    /// - `InMemory` — no durability, fastest.
    /// - `ArrowIpc` — durable Arrow IPC files, zero serialisation overhead.
    /// - `Parquet` — durable columnar files with compression and predicate
    ///   pushdown; slower writes, smaller on-disk footprint.
    /// - `S3Parquet` / `GcsParquet` — cloud object store (requires the
    ///   `cloud-storage` feature).
    ///
    /// Ignored when `olap_backend` is `DuckDb`.
    ///
    /// Available on crate feature `datafusion-backend` only.
    #[cfg(feature = "datafusion-backend")]
    pub datafusion_storage: rhei_olap::StorageMode,

    /// Sidecar CDC configuration.
    ///
    /// When `Some`, the engine switches to **sidecar mode**: CDC events are
    /// sourced from an external SQLite or PostgreSQL database via
    /// timestamp-based polling instead of local Rusqlite triggers. Set
    /// [`SidecarConfig::enable_local_oltp`] to `true` if you also need a local
    /// write path.
    ///
    /// Available on crate feature `sidecar` only.
    #[cfg(feature = "sidecar")]
    pub sidecar: Option<SidecarConfig>,

    /// Path to a RocksDB directory used as a durable CDC event buffer.
    ///
    /// When set, SQLite CDC triggers still fire into `_rhei_cdc_log`, but a
    /// bridge drains them into RocksDB on every poll cycle. The sync engine
    /// then reads exclusively from RocksDB, giving crash-durable event
    /// buffering: events survive a mid-cycle process crash. Processed events
    /// are pruned from the SQLite log after they are safely in RocksDB.
    ///
    /// Ignored in sidecar mode (the sidecar has its own CDC source).
    ///
    /// Available on crate feature `rocksdb-cdc` only.
    #[cfg(feature = "rocksdb-cdc")]
    pub rocksdb_cdc_path: Option<String>,

    /// Path to a JSON file where the schema registry is persisted across
    /// restarts.
    ///
    /// When set, [`HtapEngine::new`] loads existing table schemas from this
    /// file on startup and [`HtapEngine::register_table`] writes the updated
    /// registry back after each successful registration (atomic write). On the
    /// next start, all previously registered tables are restored automatically
    /// — no need to call `register_table` again. The file is created on first
    /// use.
    pub schema_registry_path: Option<String>,
}

/// Configuration for sidecar (external-database CDC) mode.
///
/// Pass this as [`HtapConfig::sidecar`] to switch the engine into sidecar
/// mode. In this mode CDC events are sourced from `source` via
/// timestamp-based polling rather than from local Rusqlite triggers.
///
/// Available on crate feature `sidecar` only.
#[cfg(feature = "sidecar")]
pub struct SidecarConfig {
    /// The external database to follow.
    ///
    /// Use [`SidecarSource::Sqlite`] for a local SQLite file or
    /// [`SidecarSource::Postgres`] for a remote PostgreSQL instance.
    pub source: SidecarSource,

    /// Per-table CDC settings: which tables to follow, which columns carry the
    /// timestamp and primary key, batch size, and delete-detection strategy.
    ///
    /// See [`rhei_sidecar::TimestampCdcConfig`] and
    /// [`rhei_sidecar::TimestampTableConfig`] for details.
    pub timestamp_config: rhei_sidecar::TimestampCdcConfig,

    /// Whether to also create a local Rusqlite OLTP engine.
    ///
    /// - `false` (default for pure sidecar) — no local SQLite; all writes
    ///   return [`crate::HtapError::OltpNotAvailable`].
    /// - `true` — a local Rusqlite engine is created at
    ///   [`HtapConfig::oltp_path`] and is available for writes via
    ///   [`HtapEngine::execute`].
    pub enable_local_oltp: bool,

    /// Optional path to a RocksDB directory for persisting the CDC watermark.
    ///
    /// When set, the sidecar consumer saves its polling position (the
    /// `(updated_at, pk…)` tuple for the last seen row) to RocksDB. On the
    /// next start, polling resumes from that position, avoiding a full replay
    /// from timestamp zero. Without this path, the watermark is kept in memory
    /// only and is lost on restart.
    ///
    /// Requires the `rocksdb-watermark` feature on the `rhei-sidecar` crate
    /// (automatically enabled when this crate's `sidecar` feature is on).
    pub watermark_path: Option<String>,
}

impl Default for HtapConfig {
    fn default() -> Self {
        Self {
            oltp_path: "rhei.db".to_string(),
            olap_in_memory: true,
            olap_path: None,
            sync_batch_size: 1000,
            prune_after_sync: true,
            sync_interval: None,
            #[cfg(feature = "datafusion-backend")]
            olap_backend: OlapBackendType::DataFusion,
            #[cfg(all(feature = "duckdb-backend", not(feature = "datafusion-backend")))]
            olap_backend: OlapBackendType::DuckDb,
            read_pool_size: 4,
            sync_mode: SyncMode::default(),
            #[cfg(feature = "datafusion-backend")]
            datafusion_storage: rhei_olap::StorageMode::InMemory,
            #[cfg(feature = "sidecar")]
            sidecar: None,
            #[cfg(feature = "rocksdb-cdc")]
            rocksdb_cdc_path: None,
            schema_registry_path: None,
        }
    }
}

/// Tracks whether the background CDC sync task is active.
///
/// This is an internal implementation detail of [`HtapEngine`] and is not
/// exposed in the public API.
enum SyncLoopState {
    /// No background sync loop is running.
    Stopped,
    /// A background sync loop is active.
    Running {
        /// Handle to the spawned Tokio task.
        handle: tokio::task::JoinHandle<()>,
        /// Token used to signal the sync loop to stop gracefully.
        cancel: CancellationToken,
    },
}

/// Main HTAP engine facade.
///
/// `HtapEngine` is the single entry point for all Rhei operations. It
/// orchestrates:
///
/// - **OLTP** — a local Rusqlite (SQLite) engine for low-latency writes
///   (absent in pure sidecar mode).
/// - **OLAP** — a DataFusion or DuckDB engine for analytical queries.
/// - **CDC sync** — a [`CdcSyncEngine`] that polls change events and applies
///   them to the OLAP mirror.
/// - **Query routing** — a [`SqlParserRouter`] that inspects the SQL AST and
///   sends writes / simple SELECTs to OLTP and aggregations / JOINs / CTEs to
///   OLAP automatically.
///
/// # Lifecycle
///
/// 1. **Create** — [`HtapEngine::new`] with a [`HtapConfig`].
/// 2. **Register tables** — [`HtapEngine::register_table`] for each table you
///    want replicated to OLAP. Call once per table; idempotent on restart when
///    [`HtapConfig::schema_registry_path`] is set.
/// 3. **Read / write** — [`HtapEngine::query`] for auto-routed SQL;
///    [`HtapEngine::execute`] for explicit OLTP writes.
/// 4. **Sync** — driven automatically by the background loop started via
///    [`HtapConfig::sync_interval`], or manually via [`HtapEngine::sync_now`].
/// 5. **Shut down** — [`HtapEngine::shutdown`] cancels the background loop and
///    flushes in-flight work.
///
/// # Query routing
///
/// [`HtapEngine::query`] parses the SQL with `sqlparser-rs` (SQLite dialect)
/// and routes:
///
/// - **→ OLTP**: DML (INSERT / UPDATE / DELETE), DDL, transactions, simple
///   `SELECT` with `WHERE` / `LIMIT`.
/// - **→ OLAP**: aggregates, `GROUP BY` / `HAVING`, `JOIN`s, window functions,
///   CTEs, set operations, subqueries.
/// - **Fallback**: heuristic keyword matching if AST parsing fails; defaults to
///   OLTP (safety-first).
///
/// In pure sidecar mode (no OLTP), reads fall through to OLAP automatically;
/// write statements return [`HtapError::OltpNotAvailable`].
pub struct HtapEngine {
    oltp: Option<OltpBackend>,
    olap: OlapBackend,
    router: SqlParserRouter,
    sync_engine: Arc<SyncEngineType>,
    schema_registry: SchemaRegistry,
    sync_loop: SyncLoopState,
    sync_mode: SyncMode,
    /// Whether the CDC source is the local trigger-based producer (vs sidecar).
    /// When false (sidecar mode), CDC triggers are not installed on OLTP tables.
    uses_local_cdc: bool,
    /// DDL lock: schema operations take a write lock, sync cycles take a read lock.
    ddl_lock: Arc<tokio::sync::RwLock<()>>,
    /// Optional path for persisting the schema registry between restarts.
    schema_registry_path: Option<String>,
}

impl HtapEngine {
    /// Create a new HTAP engine with the given configuration.
    pub async fn new(config: HtapConfig) -> Result<Self, HtapError> {
        // Initialize OLAP backend
        let olap = Self::create_olap_backend(&config)?;

        // Shared schema registry — load from disk if a persistence path is set.
        let schema_registry = if let Some(ref path) = config.schema_registry_path {
            SchemaRegistry::load_from_disk(path).map_err(|e| {
                HtapError::Other(format!("failed to load schema registry from '{path}': {e}"))
            })?
        } else {
            SchemaRegistry::new()
        };

        // Determine CDC source: sidecar (timestamp polling) or standard (Rusqlite CDC)
        #[cfg(feature = "sidecar")]
        if config.sidecar.is_some() {
            return Self::new_sidecar(config, olap, schema_registry).await;
        }

        // Standard HTAP path: OLTP + Rusqlite CDC
        Self::new_standard(config, olap, schema_registry).await
    }

    /// Standard HTAP construction: Rusqlite OLTP engine + trigger CDC.
    async fn new_standard(
        config: HtapConfig,
        olap: OlapBackend,
        schema_registry: SchemaRegistry,
    ) -> Result<Self, HtapError> {
        let rusqlite_engine =
            RusqliteEngine::new_local(&config.oltp_path, config.read_pool_size).await?;

        // Give CDC producer its own independent connection with busy_timeout
        let cdc_conn = rusqlite_engine.new_connection().await?;

        // Set up trigger-based CDC
        rhei_oltp_rusqlite::cdc_setup::ensure_cdc_log_table(&cdc_conn).await?;
        info!("trigger-based CDC enabled (rusqlite)");

        let cdc_producer = RusqliteCdcProducer::new(cdc_conn);
        let oltp = OltpBackend::Rusqlite(rusqlite_engine);

        // Use RocksDB bridge if configured, otherwise use SQLite trigger CDC directly.
        // Bridge mode: SQLite triggers still fire into `_rhei_cdc_log`, but every poll
        // cycle drains them into RocksDB for crash-durable buffering. The sync engine
        // then reads exclusively from RocksDB.
        #[cfg(feature = "rocksdb-cdc")]
        let cdc_source = if let Some(ref path) = config.rocksdb_cdc_path {
            let rocksdb_config = rhei_cdc_rocksdb::RocksDbCdcConfig {
                path: path.clone(),
                create_if_missing: true,
            };
            let rocksdb_log = rhei_cdc_rocksdb::RocksDbCdcLog::open(&rocksdb_config)?;
            info!(
                path = path.as_str(),
                "RocksDB CDC bridge enabled (SQLite triggers -> RocksDB durable log)"
            );
            CdcSource::RocksDbBridge {
                sqlite: cdc_producer,
                rocksdb: Arc::new(rocksdb_log),
            }
        } else {
            CdcSource::Rusqlite(cdc_producer)
        };
        #[cfg(not(feature = "rocksdb-cdc"))]
        let cdc_source = CdcSource::Rusqlite(cdc_producer);

        let sync_engine = Arc::new(
            CdcSyncEngine::new(
                cdc_source,
                olap.clone(),
                schema_registry.clone(),
                config.sync_batch_size,
            )
            .with_prune_after_sync(config.prune_after_sync)
            .with_sync_mode(config.sync_mode),
        );

        let backend_name = olap.backend_name();

        info!(
            oltp_path = config.oltp_path.as_str(),
            olap_in_memory = config.olap_in_memory,
            olap_backend = backend_name,
            oltp_backend = "rusqlite",
            prune_after_sync = config.prune_after_sync,
            "HTAP engine initialized"
        );

        let mut engine = Self {
            oltp: Some(oltp),
            olap,
            router: SqlParserRouter::new(),
            sync_engine,
            schema_registry,
            sync_loop: SyncLoopState::Stopped,
            sync_mode: config.sync_mode,
            uses_local_cdc: true,
            ddl_lock: Arc::new(tokio::sync::RwLock::new(())),
            schema_registry_path: config.schema_registry_path,
        };

        if let Some(interval) = config.sync_interval {
            engine.start_sync(interval);
        }

        Ok(engine)
    }

    /// Sidecar construction: timestamp-based CDC from external DB.
    #[cfg(feature = "sidecar")]
    async fn new_sidecar(
        config: HtapConfig,
        olap: OlapBackend,
        schema_registry: SchemaRegistry,
    ) -> Result<Self, HtapError> {
        let sidecar_config = config
            .sidecar
            .as_ref()
            .ok_or_else(|| HtapError::Other("sidecar config required for new_sidecar".into()))?;

        // Build watermark store (RocksDB if path provided, otherwise in-memory)
        let watermark_store: Box<dyn rhei_sidecar::WatermarkStore> =
            if let Some(ref wm_path) = sidecar_config.watermark_path {
                info!(path = wm_path.as_str(), "opening RocksDB watermark store");
                Box::new(
                    rhei_sidecar::RocksDbWatermarkStore::open(wm_path).map_err(|e| {
                        HtapError::Other(format!("failed to open watermark store: {e}"))
                    })?,
                )
            } else {
                // NullWatermarkStore — backward-compatible in-memory-only behavior
                // (constructed via the default `new()` path)
                Box::new(rhei_sidecar::NullWatermarkStore)
            };

        // Connect to external source and build the CDC consumer
        let (cdc_source, source_description) = match &sidecar_config.source {
            SidecarSource::Sqlite(path) => {
                use connector_arrow::rusqlite::{rusqlite, SQLiteConnection};
                let raw_conn = rusqlite::Connection::open(path)
                    .map_err(|e| HtapError::Other(format!("failed to open SQLite source: {e}")))?;
                let sqlite_conn = SQLiteConnection::new(raw_conn);
                let consumer = rhei_sidecar::TimestampCdcConsumer::try_with_watermark_store(
                    sqlite_conn,
                    sidecar_config.timestamp_config.clone(),
                    watermark_store,
                )
                .map_err(|e| HtapError::Other(format!("failed to load watermark state: {e}")))?;
                (CdcSource::SidecarSqlite(consumer), format!("sqlite:{path}"))
            }
            SidecarSource::Postgres(conn_str) => {
                use connector_arrow::postgres::{postgres::NoTls, PostgresConnection};
                // `postgres::Client::connect` is synchronous and calls `block_on`
                // internally, which panics on a tokio runtime thread.  Move it
                // to a blocking thread pool thread via `spawn_blocking`.
                let conn_str_owned = conn_str.clone();
                let pg_client = tokio::task::spawn_blocking(move || {
                    connector_arrow::postgres::postgres::Client::connect(&conn_str_owned, NoTls)
                })
                .await
                .map_err(|e| HtapError::Other(format!("spawn_blocking panicked: {e}")))?
                .map_err(|e| HtapError::Other(format!("failed to open PostgreSQL source: {e}")))?;
                let pg_conn = PostgresConnection::new(pg_client);
                let consumer = rhei_sidecar::TimestampCdcConsumer::try_with_watermark_store(
                    pg_conn,
                    sidecar_config.timestamp_config.clone(),
                    watermark_store,
                )
                .map_err(|e| HtapError::Other(format!("failed to load watermark state: {e}")))?;
                (
                    CdcSource::SidecarPostgres(consumer),
                    "postgres:<connection_string>".to_string(),
                )
            }
        };

        // Optionally create local OLTP
        let oltp: Option<OltpBackend> = if sidecar_config.enable_local_oltp {
            Some(OltpBackend::Rusqlite(
                RusqliteEngine::new_local(&config.oltp_path, config.read_pool_size).await?,
            ))
        } else {
            None
        };

        let sync_engine = Arc::new(
            CdcSyncEngine::new(
                cdc_source,
                olap.clone(),
                schema_registry.clone(),
                config.sync_batch_size,
            )
            .with_prune_after_sync(false) // Nothing to prune from external source
            .with_sync_mode(config.sync_mode),
        );

        info!(
            source = source_description.as_str(),
            enable_local_oltp = sidecar_config.enable_local_oltp,
            sync_mode = ?config.sync_mode,
            "sidecar engine initialized"
        );

        let mut engine = Self {
            oltp,
            olap,
            router: SqlParserRouter::new(),
            sync_engine,
            schema_registry,
            sync_loop: SyncLoopState::Stopped,
            sync_mode: config.sync_mode,
            uses_local_cdc: false, // sidecar mode — CDC comes from external source
            ddl_lock: Arc::new(tokio::sync::RwLock::new(())),
            schema_registry_path: config.schema_registry_path,
        };

        if let Some(interval) = config.sync_interval {
            engine.start_sync(interval);
        }

        Ok(engine)
    }

    fn create_olap_backend(config: &HtapConfig) -> Result<OlapBackend, HtapError> {
        match config.olap_backend {
            #[cfg(feature = "duckdb-backend")]
            OlapBackendType::DuckDb => {
                let engine = if config.olap_in_memory {
                    DuckDbEngine::in_memory().map_err(rhei_olap::OlapError::from)?
                } else {
                    let path = config.olap_path.as_deref().unwrap_or("rhei_olap.duckdb");
                    DuckDbEngine::persistent(path).map_err(rhei_olap::OlapError::from)?
                };
                Ok(OlapBackend::DuckDb(SharedDuckDbEngine::new(engine)))
            }
            #[cfg(feature = "datafusion-backend")]
            OlapBackendType::DataFusion => {
                let engine = DataFusionEngine::with_storage(config.datafusion_storage.clone())
                    .map_err(rhei_olap::OlapError::from)?;
                Ok(OlapBackend::DataFusion(SharedDataFusionEngine::new(engine)))
            }
        }
    }

    /// Register a table for HTAP replication.
    ///
    /// This:
    /// 1. Creates the mirror table in the OLAP engine
    /// 2. Sets up CDC triggers on the OLTP side
    /// 3. Registers the schema in the shared registry (commit point)
    ///
    /// Acquires the DDL write lock to prevent sync from running concurrently.
    pub async fn register_table(&self, schema: TableSchema) -> Result<(), HtapError> {
        let _ddl_guard = self.ddl_lock.write().await;

        // Validate upfront before any side effects (OLAP table creation, CDC triggers).
        // schema.validate() checks identifiers and PK; duplicate-name check is separate.
        schema.validate()?;
        if let Ok(existing) = self.schema_registry.get(&schema.name) {
            // Idempotency: if the schema matches the persisted one (e.g. on rh serve restart),
            // silently skip re-registration.  If the schema has changed, surface a clear error
            // rather than silently ignoring the mismatch.
            if existing.arrow_schema == schema.arrow_schema
                && existing.primary_key == schema.primary_key
            {
                debug!(
                    table = schema.name.as_str(),
                    "register_table: table already registered with matching schema, skipping"
                );
                return Ok(());
            }
            return Err(HtapError::Other(format!(
                "table '{}' is already registered with a different schema; \
                 use add_column/drop_column for schema evolution",
                schema.name
            )));
        }

        let table_name = schema.name.clone();
        let arrow_schema = schema.arrow_schema.clone();

        // In temporal mode, extend the OLAP schema with validity columns
        let olap_schema = if self.sync_mode == SyncMode::Temporal {
            for reserved in &["_rhei_valid_from", "_rhei_valid_to", "_rhei_operation"] {
                if arrow_schema.field_with_name(reserved).is_ok() {
                    return Err(HtapError::Other(format!(
                        "table '{}' already contains reserved temporal column '{}'",
                        table_name, reserved
                    )));
                }
            }
            temporalize_schema(&arrow_schema)
        } else {
            arrow_schema
        };

        // 1. Create mirror table in OLAP (before registry, so sync can't see it yet).
        // In Temporal (SCD Type 2) mode, multiple versions of the same business
        // key coexist (each close-and-insert cycle appends a new row), so the
        // business PK cannot be a uniqueness constraint. Skip the PK clause for
        // temporal mirror tables; PK info is still tracked in the schema
        // registry and used by the sync converter for WHERE clauses.
        let create_pk: &[String] = match self.sync_mode {
            SyncMode::Destructive => &schema.primary_key,
            SyncMode::Temporal => &[],
        };
        self.olap
            .create_table(&table_name, &olap_schema, create_pk)
            .await?;

        // 2. Set up CDC triggers (only in standard mode, not sidecar)
        if self.uses_local_cdc {
            if let Some(ref oltp) = self.oltp {
                // Reach through OltpBackend::Rusqlite for rusqlite-specific CDC DDL.
                // Future OLTP backends will need equivalent trigger-setup hooks here.
                let OltpBackend::Rusqlite(engine) = oltp;
                rhei_oltp_rusqlite::cdc_setup::setup_cdc(&engine.connection(), &table_name).await?;
            }
        }

        // 3. Register in schema registry last (commit point -- sync can now see this table)
        // Validation already done above; this should not fail.
        self.schema_registry.register(schema)?;

        // 4. Persist the updated registry if a path is configured.
        if let Some(ref path) = self.schema_registry_path {
            if let Err(e) = self.schema_registry.save_to_disk(path) {
                warn!(
                    path = path.as_str(),
                    error = %e,
                    "failed to persist schema registry after register_table"
                );
            }
        }

        info!(
            table = table_name.as_str(),
            "table registered for HTAP replication"
        );
        Ok(())
    }

    /// Execute a read-only SQL query and return all matching rows as Arrow
    /// `RecordBatch`es.
    ///
    /// The configured [`QueryRouter`] (default: [`rhei_sync::SqlParserRouter`])
    /// inspects the SQL and decides which engine to use:
    ///
    /// - Simple point lookups, writes, and DDL are sent to **OLTP** (Rusqlite).
    /// - Aggregates, GROUP BY/HAVING, JOINs, window functions, CTEs, and set
    ///   operations are sent to **OLAP** (DuckDB or DataFusion).
    ///
    /// In sidecar mode (no local OLTP), only `SELECT` / `WITH` / `EXPLAIN`
    /// statements fall through to OLAP; any write statement returns
    /// [`HtapError::OltpNotAvailable`].
    ///
    /// To bypass routing entirely, use [`HtapEngine::query_with_hint`] with
    /// [`QueryHint::ForceOltp`] or [`QueryHint::ForceOlap`].
    ///
    /// # Errors
    ///
    /// Returns [`HtapError::OltpNotAvailable`] if the query was routed to OLTP
    /// but OLTP is disabled (sidecar mode), or any error from the underlying
    /// engine.
    pub async fn query(&self, sql: &str) -> Result<Vec<RecordBatch>, HtapError> {
        let target = self.router.route(sql);
        #[cfg(feature = "metrics")]
        match target {
            QueryTarget::Oltp => metrics::counter!("rhei.query.routed_oltp").increment(1),
            QueryTarget::Olap => metrics::counter!("rhei.query.routed_olap").increment(1),
        };
        match target {
            QueryTarget::Oltp => {
                if let Some(ref oltp) = self.oltp {
                    let batches = oltp.query(sql, &[]).await?;
                    Ok(batches)
                } else {
                    // No OLTP available (sidecar mode).
                    // Only allow reads to fall through to OLAP; reject writes
                    // to prevent unintended OLAP mutations.
                    let upper = sql.trim_start().to_uppercase();
                    if upper.starts_with("SELECT")
                        || upper.starts_with("WITH")
                        || upper.starts_with("EXPLAIN")
                    {
                        let batches = self.olap.query(sql).await?;
                        Ok(batches)
                    } else {
                        Err(HtapError::OltpNotAvailable)
                    }
                }
            }
            QueryTarget::Olap => {
                let batches = self.olap.query(sql).await?;
                Ok(batches)
            }
        }
    }

    /// Execute a read-only SQL query with an explicit routing hint, bypassing
    /// the SQL parser router.
    ///
    /// Use this when the caller already knows which engine should handle the
    /// query and wants to skip parsing — e.g. forcing an analytic query to
    /// OLAP even if the parser would route it to OLTP, or running an admin
    /// query directly against OLTP.
    ///
    /// - [`QueryHint::ForceOltp`] — always OLTP; errors with
    ///   [`HtapError::OltpNotAvailable`] in sidecar mode.
    /// - [`QueryHint::ForceOlap`] — always OLAP.
    /// - [`QueryHint::Auto`] — equivalent to [`HtapEngine::query`] (uses the router).
    ///
    /// # Errors
    ///
    /// Propagates engine errors and [`HtapError::OltpNotAvailable`] when
    /// `ForceOltp` is requested without an OLTP backend.
    pub async fn query_with_hint(
        &self,
        sql: &str,
        hint: QueryHint,
    ) -> Result<Vec<RecordBatch>, HtapError> {
        match hint {
            QueryHint::ForceOltp => {
                let oltp = self.oltp.as_ref().ok_or(HtapError::OltpNotAvailable)?;
                let batches = oltp.query(sql, &[]).await?;
                Ok(batches)
            }
            QueryHint::ForceOlap => {
                let batches = self.olap.query(sql).await?;
                Ok(batches)
            }
            QueryHint::Auto => self.query(sql).await,
        }
    }

    /// Execute a single write statement (INSERT / UPDATE / DELETE / DDL)
    /// against OLTP and return the number of affected rows.
    ///
    /// `params` are bound positionally — pass an empty slice for statements
    /// without placeholders. The change becomes visible to OLAP after the
    /// next sync cycle (CDC triggers append to `_rhei_cdc_log`, which
    /// [`rhei_sync::CdcSyncEngine`] drains and applies).
    ///
    /// For multi-statement transactions, prefer [`HtapEngine::execute_batch`]
    /// — it amortizes the per-commit fsync (~50× higher throughput).
    ///
    /// # Errors
    ///
    /// Returns [`HtapError::OltpNotAvailable`] in pure sidecar mode (where
    /// `enable_local_oltp = false`); otherwise propagates the underlying
    /// OLTP error.
    pub async fn execute(&self, sql: &str, params: &[serde_json::Value]) -> Result<u64, HtapError> {
        let oltp = self.oltp.as_ref().ok_or(HtapError::OltpNotAvailable)?;
        let rows = oltp.execute(sql, params).await?;
        Ok(rows)
    }

    /// Execute multiple write statements in a single transaction.
    ///
    /// This is **much faster** than calling `execute()` in a loop -- individual
    /// statements do an fsync per commit (~350 ops/sec), while a batch wraps
    /// everything in BEGIN/COMMIT with a single fsync (~18,000+ ops/sec with
    /// CDC triggers, ~73,000+ without).
    pub async fn execute_batch(
        &self,
        statements: &[(String, Vec<serde_json::Value>)],
    ) -> Result<(), HtapError> {
        let oltp = self.oltp.as_ref().ok_or(HtapError::OltpNotAvailable)?;
        oltp.execute_batch(statements).await?;
        Ok(())
    }

    /// Perform a single sync cycle (poll CDC, apply to OLAP).
    ///
    /// Acquires the DDL read lock so schema changes cannot interleave with sync.
    pub async fn sync_now(&self) -> Result<SyncResult, HtapError> {
        let _guard = self.ddl_lock.read().await;
        let result = self.sync_engine.sync_once().await?;
        Ok(result)
    }

    /// Start a background sync loop that polls CDC on the given interval.
    ///
    /// If a sync loop is already running, it is stopped first.
    pub fn start_sync(&mut self, interval: Duration) {
        // Stop any existing loop to prevent overlap
        if let SyncLoopState::Running { cancel, handle } =
            std::mem::replace(&mut self.sync_loop, SyncLoopState::Stopped)
        {
            cancel.cancel();
            handle.abort();
        }

        let cancel = CancellationToken::new();
        let handle = spawn_sync_loop(
            Arc::clone(&self.sync_engine),
            interval,
            cancel.clone(),
            Arc::clone(&self.ddl_lock),
        );

        self.sync_loop = SyncLoopState::Running { handle, cancel };
    }

    /// Stop the background sync loop, if running.
    pub async fn stop_sync(&mut self) {
        if let SyncLoopState::Running { cancel, handle } =
            std::mem::replace(&mut self.sync_loop, SyncLoopState::Stopped)
        {
            cancel.cancel();
            let _ = handle.await;
        }
    }

    /// Check if the background sync loop is running.
    pub fn is_sync_running(&self) -> bool {
        matches!(&self.sync_loop, SyncLoopState::Running { handle, .. } if !handle.is_finished())
    }

    /// Get the current sync status.
    pub async fn sync_status(&self) -> Result<SyncStatus, HtapError> {
        let mut status = self.sync_engine.status().await?;
        status.running = self.is_sync_running();
        #[cfg(feature = "metrics")]
        metrics::gauge!("rhei.sync.cdc_lag").set(status.lag as f64);
        Ok(status)
    }

    /// Get a reference to the schema registry.
    pub fn schema_registry(&self) -> &SchemaRegistry {
        &self.schema_registry
    }

    /// Get a reference to the OLTP backend (None in pure sidecar mode).
    pub fn oltp(&self) -> Option<&OltpBackend> {
        self.oltp.as_ref()
    }

    /// Get a reference to the inner `RusqliteEngine` if the OLTP backend is
    /// `OltpBackend::Rusqlite`. Returns `None` in pure sidecar mode or when a
    /// different backend variant is active.
    ///
    /// Prefer the trait-level `oltp()` accessor for all standard OLTP
    /// operations. Use this only when rusqlite-specific access is required
    /// (e.g., advanced diagnostic tooling or direct connection borrowing).
    pub fn oltp_rusqlite(&self) -> Option<&RusqliteEngine> {
        self.oltp.as_ref().and_then(OltpBackend::as_rusqlite)
    }

    /// Get a reference to the OLAP backend.
    pub fn olap(&self) -> &OlapBackend {
        &self.olap
    }

    /// Perform an initial full sync for a registered table.
    ///
    /// Reads all rows from the OLTP table and bulk-loads them into OLAP using
    /// `load_arrow`. This is useful for cold-start scenarios where the OLAP
    /// engine needs to be populated with existing OLTP data.
    ///
    /// The table must already be registered (via `register_table`).
    pub async fn initial_sync(&self, table_name: &str) -> Result<u64, HtapError> {
        rhei_core::validate_identifier(table_name)?;
        let schema = self.schema_registry.get(table_name)?;
        info!(table = table_name, "starting initial sync (full load)");

        // Query all rows from OLTP
        let oltp = self.oltp.as_ref().ok_or(HtapError::OltpNotAvailable)?;
        let sql = format!("SELECT * FROM {table_name}");
        let batches = oltp.query(&sql, &[]).await?;

        let total_rows: usize = batches.iter().map(|b| b.num_rows()).sum();
        if batches.is_empty() || total_rows == 0 {
            info!(
                table = table_name,
                rows = 0,
                "initial sync complete (empty table)"
            );
            return Ok(0);
        }

        // All batches from a single OltpEngine::query call share the same schema --
        // this is a contract of the Arrow RecordBatch protocol and libSQL's query API.
        // The debug_assert catches any violation early in test/debug builds.
        debug_assert!(
            batches.iter().all(|b| b.num_columns() == batches[0].num_columns()),
            "OltpEngine::query returned batches with inconsistent column counts for table '{table_name}'"
        );

        // Build column index once from the first batch schema, then reuse for all batches.
        let col_indices: Vec<usize> = {
            let first_schema = batches[0].schema();
            schema
                .arrow_schema
                .fields()
                .iter()
                .map(|f| {
                    first_schema.index_of(f.name()).map_err(|_| {
                        HtapError::Other(format!(
                            "column '{}' not found in OLTP result for table '{}'",
                            f.name(),
                            table_name
                        ))
                    })
                })
                .collect::<Result<_, _>>()?
        };

        // Reorder columns to match the registered Arrow schema
        let mut aligned_batches = Vec::with_capacity(batches.len());
        for batch in &batches {
            let columns: Vec<_> = col_indices
                .iter()
                .map(|&i| batch.column(i).clone())
                .collect();
            aligned_batches.push(
                RecordBatch::try_new(schema.arrow_schema.clone(), columns)
                    .map_err(|e| HtapError::Other(e.to_string()))?,
            );
        }

        // Bulk load into OLAP
        let rows = self.olap.load_arrow(table_name, &aligned_batches).await?;
        info!(table = table_name, rows, "initial sync complete");
        Ok(rows)
    }

    /// Perform initial sync for all registered tables.
    pub async fn initial_sync_all(&self) -> Result<u64, HtapError> {
        let table_names = self.schema_registry.table_names();
        let mut total = 0u64;
        for name in &table_names {
            total += self.initial_sync(name).await?;
        }
        Ok(total)
    }

    /// Add a column to a registered table, propagating the change to:
    /// 1. The OLAP mirror table
    /// 2. CDC triggers (re-created to capture the new column)
    /// 3. The schema registry (updated last as commit point)
    ///
    /// The caller is responsible for executing the ALTER TABLE on OLTP first.
    ///
    /// Acquires the DDL write lock to prevent sync from running concurrently.
    /// On failure after CDC teardown, a best-effort trigger restore is attempted.
    pub async fn add_column(
        &self,
        table_name: &str,
        column_name: &str,
        data_type: arrow::datatypes::DataType,
    ) -> Result<(), HtapError> {
        let _ddl_guard = self.ddl_lock.write().await;
        info!(table = table_name, column = column_name, "adding column");

        // Validate upfront before any side effects
        rhei_core::validate_identifier(column_name)?;
        let existing = self.schema_registry.get(table_name)?;
        if existing.arrow_schema.field_with_name(column_name).is_ok() {
            return Err(rhei_core::CoreError::SchemaValidation(format!(
                "column '{}' already exists in table '{}'",
                column_name, table_name
            ))
            .into());
        }
        drop(existing);

        // 1. Propagate to OLAP first (sync can't see the new column until registry is updated)
        self.olap
            .add_column(table_name, column_name, &data_type)
            .await?;

        // 2. Rebuild CDC triggers (only in standard mode, not sidecar)
        if self.uses_local_cdc {
            if let Some(ref oltp) = self.oltp {
                self.teardown_cdc_triggers(oltp, table_name).await?;

                if let Err(e) = self.setup_cdc_triggers(oltp, table_name).await {
                    warn!(
                        table = table_name,
                        error = %e,
                        "add_column: setup_cdc failed after teardown; restoring triggers"
                    );
                    let _ = self.setup_cdc_triggers(oltp, table_name).await;
                    return Err(e);
                }
            }
        }

        // 3. Update schema registry last (commit point -- sync can now see the new column)
        self.schema_registry
            .add_column(table_name, column_name, data_type)?;

        info!(
            table = table_name,
            column = column_name,
            "column added and CDC updated"
        );
        Ok(())
    }

    /// Drop a column from a registered table, propagating the change to:
    /// 1. CDC triggers (torn down so the OLTP ALTER TABLE can proceed)
    /// 2. The OLTP table via `ALTER TABLE ... DROP COLUMN`
    /// 3. The OLAP mirror table
    /// 4. CDC triggers (re-created to reflect the post-ALTER schema)
    /// 5. The schema registry (updated last for consistency)
    ///
    /// SQLite rejects `ALTER TABLE DROP COLUMN` when a trigger references
    /// the column, so this method handles the full teardown -> ALTER -> rebuild
    /// sequence internally. The caller does NOT need to run ALTER TABLE separately.
    ///
    /// On failure after CDC teardown, a best-effort trigger restore is attempted
    /// so the table is not left without CDC coverage.
    ///
    /// Cannot drop primary key columns.
    pub async fn drop_column(&self, table_name: &str, column_name: &str) -> Result<(), HtapError> {
        let _ddl_guard = self.ddl_lock.write().await;
        info!(table = table_name, column = column_name, "dropping column");

        // Validate inputs before touching any engine, so we fail fast without
        // leaving any engine in an inconsistent state.
        rhei_core::validate_identifier(column_name)?;
        let schema = self.schema_registry.get(table_name)?;
        if schema.primary_key.contains(&column_name.to_string()) {
            return Err(rhei_core::CoreError::SchemaValidation(format!(
                "cannot drop primary key column '{}' from table '{}'",
                column_name, table_name
            ))
            .into());
        }
        if schema.arrow_schema.field_with_name(column_name).is_err() {
            return Err(rhei_core::CoreError::SchemaValidation(format!(
                "column '{}' not found in table '{}'",
                column_name, table_name
            ))
            .into());
        }
        drop(schema); // release the Arc before taking mutable engine locks

        if self.oltp.is_none() {
            // In sidecar mode without OLTP, just update OLAP + registry.
            self.olap.drop_column(table_name, column_name).await?;
            self.schema_registry.drop_column(table_name, column_name)?;
            info!(table = table_name, column = column_name, "column dropped");
            return Ok(());
        }

        let oltp = self.oltp.as_ref().ok_or(HtapError::OltpNotAvailable)?;

        // Tear down CDC triggers before ALTER TABLE DROP COLUMN (only in standard mode).
        if self.uses_local_cdc {
            self.teardown_cdc_triggers(oltp, table_name).await?;
        }

        let result = async {
            self.execute(
                &format!("ALTER TABLE {table_name} DROP COLUMN {column_name}"),
                &[],
            )
            .await?;

            self.olap.drop_column(table_name, column_name).await?;

            if self.uses_local_cdc {
                self.setup_cdc_triggers(oltp, table_name).await?;
            }

            self.schema_registry.drop_column(table_name, column_name)?;

            Ok::<(), HtapError>(())
        }
        .await;

        if let Err(ref e) = result {
            warn!(
                table = table_name,
                error = %e,
                "drop_column failed after CDC teardown; attempting to restore triggers"
            );
            if self.uses_local_cdc {
                if let Err(rebuild_err) = self.setup_cdc_triggers(oltp, table_name).await {
                    warn!(
                        table = table_name,
                        error = %rebuild_err,
                        "failed to restore CDC triggers after drop_column error; \
                         table may need manual re-registration"
                    );
                }
            }
        } else {
            info!(
                table = table_name,
                column = column_name,
                "column dropped and CDC triggers updated"
            );
        }

        result
    }

    /// Set up CDC triggers on the OLTP backend.
    ///
    /// Reaches through `OltpBackend::Rusqlite` to get the raw connection handle
    /// needed for rusqlite-specific CDC trigger DDL. Future OLTP backends will
    /// need equivalent hooks here.
    async fn setup_cdc_triggers(
        &self,
        oltp: &OltpBackend,
        table_name: &str,
    ) -> Result<(), HtapError> {
        let OltpBackend::Rusqlite(engine) = oltp;
        rhei_oltp_rusqlite::cdc_setup::setup_cdc(&engine.connection(), table_name).await?;
        Ok(())
    }

    /// Tear down CDC triggers on the OLTP backend.
    ///
    /// Reaches through `OltpBackend::Rusqlite` to get the raw connection handle
    /// needed for rusqlite-specific CDC trigger DDL. Future OLTP backends will
    /// need equivalent hooks here.
    async fn teardown_cdc_triggers(
        &self,
        oltp: &OltpBackend,
        table_name: &str,
    ) -> Result<(), HtapError> {
        let OltpBackend::Rusqlite(engine) = oltp;
        rhei_oltp_rusqlite::cdc_setup::teardown_cdc(&engine.connection(), table_name).await?;
        Ok(())
    }

    /// Gracefully shut down the engine, stopping background sync if running.
    pub async fn shutdown(&mut self) {
        self.stop_sync().await;
        info!("HTAP engine shut down");
    }
}

impl Drop for HtapEngine {
    fn drop(&mut self) {
        // Cancel the background sync loop and abort the task for prompt teardown.
        // We can't await the handle in Drop, but abort() is best-effort and
        // prevents the task from keeping resources alive unnecessarily.
        if let SyncLoopState::Running { cancel, handle } =
            std::mem::replace(&mut self.sync_loop, SyncLoopState::Stopped)
        {
            cancel.cancel();
            handle.abort();
        }
    }
}