alkstore/alkstore-sqlite/src/store.rs

//! The SQLite engine's store: the `open` constructor, the connection
//! architecture (engine-sqlite.md's "Connection architecture"), and
//! the [`Store`] trait wiring.
//!
//! Connection architecture: one writer connection behind the
//! substrate's `Writer` slot (WAL single-writer modeled honestly), a
//! pool of reader connections (fresh opens on demand up to
//! [`SqliteOpts::max_readers`]), and one `SharedUpdateWatcher` poll
//! thread over the db path. Every connection — writer and pooled
//! readers alike — runs the substrate's full bootstrap at open:
//! pragmas (WAL, `synchronous=NORMAL`, busy timeout), the notify
//! machinery, the substrate's SQL function set, and the schema
//! bootstrap.
//!
//! Watcher spawn is fallible in the substrate (ADR-012 §4's W-2); the
//! failure maps into [`Error::Database`](alkstore::Error::Database)
//! engine-side and `open` fails if the watcher cannot start. On
//! watcher death the substrate's `WatcherDeathGuard` closes every
//! subscriber (ADR-006) — the death-guard plumbing exists from open
//! onward; the consumer-visible receiver bridging arrives with
//! `listen`.
//!
//! Drop/close: the watcher's poll thread is joined (releasing its
//! connection and the db file handle) and the writer slot + reader
//! pool are closed; [`Drop`](SqliteStore::drop) delegates to
//! [`SqliteStore::close`].
//!
//! Every engine call round-trips the `spawn_blocking` seam
//! ([`crate::seam`]) — the substrate is sync machinery (std threads,
//! blocking iterators) bridged at the trait seam, per the family
//! REQ-TTY-01 posture. Single-host by nature: file-backed, one
//! machine; cross-process on one host is verified ground
//! (`data_version` is cross-process by nature); NFS
//! two-writers-unsupported (the lineage's honesty
//! posture, inherited — ADR-016).

use alkstore::Store;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::atomic::AtomicBool;

use crate::opts::SqliteOpts;
use crate::seam::{database_error, sqlite_error};
use crate::substrate::{Readers, SharedUpdateWatcher, Writer};

/// The SQLite engine's store handle. Consumers hold
/// [`alkstore::Store`] — the trait is the contract surface; the type
/// exists so the engine can own its machinery and so tests can reach
/// the close path directly.
pub struct SqliteStore {
    writer: Arc<Writer>,
    readers: Arc<Readers>,
    watcher: Arc<SharedUpdateWatcher>,
    db_path: PathBuf,
    closed: Arc<AtomicBool>,
    /// The store's commit-fault arm (`sqlite-commit-error-arm`) — born
    /// disarmed; a test arms it through [`SqliteStore::arm_commit_fault`]
    /// so the next handle commit fails through the real commit error
    /// arm. Dead in production builds.
    #[cfg(test)]
    commit_fault: crate::seam::commit_fault::Flag,
    /// The store's rollback-fault arm (review 003 Finding 3) — born
    /// disarmed; a test arms it through
    /// [`SqliteStore::arm_rollback_fault`] so the next ROLLBACK on a
    /// drop path (the handle's or the cancelled op's lease's) fails
    /// through the real rollback-failure drop arms. Dead in production
    /// builds.
    #[cfg(test)]
    rollback_fault: crate::seam::rollback_fault::Flag,
}

impl std::fmt::Debug for SqliteStore {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SqliteStore")
            .field("db_path", &self.db_path)
            .finish_non_exhaustive()
    }
}

impl SqliteStore {
    /// Explicit close: stops the watcher, disconnects every subscriber
    /// (the death-guard close — receivers see them close), closes the
    /// reader pool, and closes the writer slot. Idempotent. [`Drop`]
    /// delegates here.
    pub fn close(&self) {
        self.closed.store(true, std::sync::atomic::Ordering::SeqCst);
        let _ = self.watcher.close();
        self.readers.close();
        self.writer.close();
    }

    /// Arm the commit-fault seam (`sqlite-commit-error-arm`): the
    /// next `commit` through a handle of this store fails with a
    /// `SQLITE_FULL`-shaped driver error and replenishes the writer
    /// slot through the real commit error arm. One-store scoping keeps
    /// parallel test stores unaffected.
    #[cfg(test)]
    pub(crate) fn arm_commit_fault(&self) {
        crate::seam::commit_fault::arm(&self.commit_fault)
    }

    /// Arm the rollback-fault seam (review 003 Finding 3): the next
    /// `ROLLBACK` on a drop path of this store (a transaction handle
    /// dropped open, or a tx-op future dropped mid-op) fails with a
    /// `SQLITE_FULL`-shaped driver error and replenishes the writer
    /// slot through the real rollback-failure drop arm. One-store
    /// scoping (disarms on first consumption) keeps parallel test
    /// stores unaffected.
    #[cfg(test)]
    pub(crate) fn arm_rollback_fault(&self) {
        crate::seam::rollback_fault::arm(&self.rollback_fault)
    }
}

impl Drop for SqliteStore {
    fn drop(&mut self) {
        self.close();
    }
}

/// Open a SQLite-backed store at `path`.
///
/// The path is a **plain filesystem path** (ADR-023 §3): no URI
/// interpretation — a `?name=value` suffix is part of the literal
/// filename (`:memory:` works without URI mode). The full connection
/// architecture boots: the writer slot, the reader pool, and the
/// update watcher (whose spawn is fallible — open fails with
/// [`Error::Database`](alkstore::Error::Database) if it cannot start).
pub fn open(path: &str, opts: SqliteOpts) -> alkstore::Result<Box<dyn Store>> {
    Ok(Box::new(open_store(path, opts)?))
}

pub(crate) fn open_store(path: &str, opts: SqliteOpts) -> alkstore::Result<SqliteStore> {
    let config = opts.watcher_config()?;

    let writer_conn = crate::substrate::open_conn_bootstrapped(path)
        .map_err(|e| match e {
            crate::substrate::SchemaError::Sqlite(inner) => sqlite_error(inner),
        })
        .map_err(|e| database_error(format!("failed to open/bootstrap writer connection: {e}")))?;

    let watcher = SharedUpdateWatcher::new_with_config(PathBuf::from(path), config)
        .map_err(|e| database_error(format!("failed to spawn update watcher: {e}")))?;

    Ok(SqliteStore {
        writer: Arc::new(Writer::new(path.to_string(), writer_conn)),
        readers: Arc::new(Readers::new(path.to_string(), opts.max_readers)),
        watcher: Arc::new(watcher),
        db_path: PathBuf::from(path),
        closed: Arc::new(AtomicBool::new(false)),
        #[cfg(test)]
        commit_fault: crate::seam::commit_fault::disarmed(),
        #[cfg(test)]
        rollback_fault: crate::seam::rollback_fault::disarmed(),
    })
}

impl Store for SqliteStore {
    fn begin_tx(
        &self,
    ) -> alkstore::BoxedFuture<'_, alkstore::Result<Box<dyn alkstore::TxHandle + Send>>> {
        let writer = self.writer.clone();
        let db_path = self.db_path.to_string_lossy().into_owned();
        let reopen: std::sync::Arc<
            dyn Fn() -> alkstore::Result<rusqlite::Connection> + Send + Sync,
        > = std::sync::Arc::new(move || crate::seam::open_writer_connection(&db_path));
        #[cfg(test)]
        let fut = crate::tx::SqliteTxHandle::begin(
            writer,
            reopen,
            self.commit_fault.clone(),
            self.rollback_fault.clone(),
        );
        #[cfg(not(test))]
        let fut = crate::tx::SqliteTxHandle::begin(writer, reopen);
        fut
    }

    fn notify<'a>(
        &'a self,
        channel: &str,
        payload: serde_json::Value,
    ) -> alkstore::BoxedFuture<'a, alkstore::Result<()>> {
        crate::notify::notify(self.writer.clone(), channel, payload)
    }

    fn listen<'a>(
        &'a self,
        channel: &str,
    ) -> alkstore::BoxedFuture<'a, alkstore::Result<Box<dyn alkstore::WakeReceiver>>> {
        let watcher = self.watcher.clone();
        let closed = self.closed.clone();
        let channel = channel.to_string();
        Box::pin(async move { crate::notify::listen(watcher, closed, &channel) })
    }

    fn stream<'a>(
        &'a self,
        name: &str,
    ) -> alkstore::BoxedFuture<'a, alkstore::Result<Box<dyn alkstore::StreamHandle>>> {
        if let Err(e) = alkstore::validate_shared_name(name) {
            return Box::pin(async move { Err(e) });
        }
        if self.closed.load(std::sync::atomic::Ordering::Acquire) {
            return Box::pin(async { Err(crate::seam::database_error("the store is closed")) });
        }
        let name = name.to_string();
        let writer = self.writer.clone();
        let readers = self.readers.clone();
        let watcher = self.watcher.clone();
        let closed = self.closed.clone();
        Box::pin(async move {
            Ok(Box::new(crate::stream::SqliteStreamHandle::new(
                name, writer, readers, watcher, closed,
            )) as Box<dyn alkstore::StreamHandle>)
        })
    }

    fn queue<'a>(
        &'a self,
        name: &str,
        opts: alkstore::QueueOpts,
    ) -> alkstore::BoxedFuture<'a, alkstore::Result<Box<dyn alkstore::Queue>>> {
        if let Err(e) = alkstore::validate_shared_name(name) {
            return Box::pin(async move { Err(e) });
        }
        if self.closed.load(std::sync::atomic::Ordering::Acquire) {
            return Box::pin(async { Err(crate::seam::database_error("the store is closed")) });
        }
        let name = name.to_string();
        let writer = self.writer.clone();
        let readers = self.readers.clone();
        let closed = self.closed.clone();
        Box::pin(async move {
            Ok(Box::new(crate::queue::SqliteQueueHandle::new(
                name, opts, writer, readers, closed,
            )) as Box<dyn alkstore::Queue>)
        })
    }

    fn outbox<'a>(
        &'a self,
        name: &str,
    ) -> alkstore::BoxedFuture<'a, alkstore::Result<Box<dyn alkstore::Outbox>>> {
        if self.closed.load(std::sync::atomic::Ordering::Acquire) {
            return Box::pin(async { Err(crate::seam::closed_store_error()) });
        }
        let name = name.to_string();
        let writer = self.writer.clone();
        let closed = self.closed.clone();
        Box::pin(async move { crate::outbox::new_outbox_handle(&name, writer, closed) })
    }

    fn try_lock<'a>(
        &'a self,
        name: &str,
        owner: &str,
        ttl: i64,
    ) -> alkstore::BoxedFuture<'a, alkstore::Result<Option<Box<dyn alkstore::Lock>>>> {
        let name = name.to_string();
        let owner = owner.to_string();
        let writer = self.writer.clone();
        let closed = self.closed.clone();
        crate::lock::try_lock(&name, &owner, ttl, writer, closed)
    }

    fn schedule<'a>(
        &'a self,
        name: &str,
        spec: &str,
        queue: &str,
        payload: serde_json::Value,
        opts: alkstore::ScheduleOpts,
    ) -> alkstore::BoxedFuture<'a, alkstore::Result<alkstore::Schedule>> {
        let writer = self.writer.clone();
        let closed = self.closed.clone();
        let name = name.to_string();
        let spec = spec.to_string();
        let queue = queue.to_string();
        crate::scheduler::schedule(&name, &spec, &queue, payload, opts, writer, closed)
    }

    fn unschedule<'a>(&'a self, name: &str) -> alkstore::BoxedFuture<'a, alkstore::Result<bool>> {
        let writer = self.writer.clone();
        let closed = self.closed.clone();
        let name = name.to_string();
        crate::scheduler::unschedule(&name, writer, closed)
    }

    fn run_schedules<'a>(
        &'a self,
        stop: alkstore::StopToken,
    ) -> alkstore::BoxedFuture<'a, alkstore::Result<()>> {
        let writer = self.writer.clone();
        let closed = self.closed.clone();
        let db_path = self.db_path.to_string_lossy().into_owned();
        #[cfg(test)]
        let fut = crate::scheduler::run_schedules(
            stop,
            writer,
            closed,
            db_path,
            self.commit_fault.clone(),
        );
        #[cfg(not(test))]
        let fut = crate::scheduler::run_schedules(stop, writer, closed, db_path);
        fut
    }
}

#[cfg(test)]
mod lock_tests;
#[cfg(test)]
mod notify_tests;
#[cfg(test)]
mod open_tests;
#[cfg(test)]
mod outbox_tests;
#[cfg(test)]
mod queue_tests;
#[cfg(test)]
mod scheduler_tests;
#[cfg(test)]
mod seam_tests;
#[cfg(test)]
mod stream_tests;
#[cfg(test)]
mod tx_tests;