dpm-store 0.1.0

use crate::StoreLineage;
use crate::{
    StoreError,
    error::database_error,
    schema::{self, Layout},
};
use dpm_engine::{Operation, apply_command};
use dpm_model::{LineageId, Plan};
use rusqlite::{Connection, OpenFlags, TransactionBehavior};
use std::{
    path::{Path, PathBuf},
    time::Duration,
};

mod history;
mod integrity;
mod lineage;
mod recovery;
use recovery::files;
mod replay;
mod runs;
pub(crate) use crate::{HistoryEntry, RecordedOperation};
pub use recovery::{IntegrityReport, restore_store_blocking, verify_store_blocking};
pub use runs::{
    RUN_ACTIVITY_LIMIT, RUN_STORE_VERSION, RunBinding, RunStore, RunStoreReport, Written,
};
mod snapshot;
mod snapshot_cache;
use snapshot::load_blocking;

/// How long a write waits for another connection's lock before reporting the store busy.
const BUSY_TIMEOUT: Duration = Duration::from_secs(5);

/// A synchronous local database with validated snapshots and immutable operation history.
pub struct SqliteStore {
    connection: Connection,
    cache: snapshot_cache::SnapshotCache,
    path: PathBuf,
}

impl SqliteStore {
    /// Create or open a database without writing to it; [`Self::initialize_blocking`] creates the
    /// schema.
    pub fn open_blocking(path: impl AsRef<Path>) -> Result<Self, StoreError> {
        let path = path.as_ref().to_path_buf();
        check_existing_read_only_blocking(&path)?;
        let connection = Connection::open(&path).map_err(database_error(&path, "open database"))?;
        Self::prepare_blocking(connection, path)
    }

    /// Open an existing database without creating a new file or writing to it; a store in a
    /// retired or newer layout is refused unchanged.
    pub fn open_existing_blocking(path: impl AsRef<Path>) -> Result<Self, StoreError> {
        let path = path.as_ref().to_path_buf();
        check_existing_read_only_blocking(&path)?;
        let connection = Connection::open_with_flags(&path, OpenFlags::SQLITE_OPEN_READ_WRITE)
            .map_err(database_error(&path, "open existing database"))?;
        Self::prepare_blocking(connection, path)
    }

    /// Create an isolated in-memory database for ephemeral work and tests.
    pub fn in_memory_blocking() -> Result<Self, StoreError> {
        let path = PathBuf::from(":memory:");
        let connection =
            Connection::open_in_memory().map_err(database_error(&path, "open database"))?;
        Self::prepare_blocking(connection, path)
    }

    fn prepare_blocking(connection: Connection, path: PathBuf) -> Result<Self, StoreError> {
        connection
            .busy_timeout(BUSY_TIMEOUT)
            .map_err(database_error(&path, "set busy timeout"))?;
        // Refused formats are checked before this read-write connection opens; schema creation
        // happens inside the initializing write transaction.
        schema::check_blocking(&connection, &path)?;
        connection
            .execute_batch("PRAGMA foreign_keys = ON;")
            .map_err(database_error(&path, "configure database"))?;
        Ok(Self {
            connection,
            path,
            cache: Default::default(),
        })
    }

    /// Database location, or the in-memory label.
    pub fn path(&self) -> &Path {
        &self.path
    }

    /// Wait at most `timeout` for another connection's lock before failing as busy.
    pub fn set_busy_timeout_blocking(&self, timeout: Duration) -> Result<(), StoreError> {
        self.connection
            .busy_timeout(timeout)
            .map_err(database_error(&self.path, "set busy timeout"))
    }

    /// The lineage this file continues, if initialized.
    pub fn lineage_blocking(&self) -> Result<Option<StoreLineage>, StoreError> {
        if schema::check_blocking(&self.connection, &self.path)? == Layout::Empty {
            return Ok(None);
        }
        lineage::read_blocking(&self.connection, &self.path).map(Some)
    }

    /// The committed revision and its lineage, read in one transaction without the snapshot text,
    /// so polling for changes costs two row reads; `None` for an uninitialized store.
    pub fn revision_blocking(&self) -> Result<Option<crate::StoreRevision>, StoreError> {
        read_revision_blocking(&self.connection, &self.path)
    }

    /// The plan as it stood before the operation at local sequence `sequence`, replayed from the
    /// genesis plan; a resent plan change is compared against the state it was first applied to.
    pub fn plan_before_blocking(&self, sequence: u64) -> Result<Plan, StoreError> {
        let transaction = self
            .connection
            .unchecked_transaction()
            .map_err(database_error(&self.path, "begin replay"))?;
        let genesis = snapshot::genesis_blocking(&transaction, &self.path)?.ok_or_else(|| {
            StoreError::MissingGenesis {
                path: self.path.clone(),
            }
        })?;
        let plan = replay::replay_before_blocking(&transaction, &self.path, genesis, sequence)?;
        transaction
            .commit()
            .map_err(database_error(&self.path, "finish replay"))?;
        Ok(plan)
    }

    /// Read and validate the authoritative snapshot, if initialized.
    pub fn load_blocking(&self) -> Result<Option<Plan>, StoreError> {
        if schema::check_blocking(&self.connection, &self.path)? == Layout::Empty {
            return Ok(None);
        }
        snapshot::load_cached_blocking(&self.connection, &self.path, &self.cache)
    }

    /// Initialize an empty database with `plan` as both the snapshot and the genesis plan the
    /// history replays from; an existing snapshot or operation history is never erased, and a
    /// refused initialization writes nothing.
    pub fn initialize_blocking(&mut self, plan: &Plan) -> Result<(), StoreError> {
        plan.validate()?;
        if occupied_blocking(&self.connection, &self.path)? {
            return Err(StoreError::AlreadyInitialized(self.path.clone()));
        }
        // The journal mode cannot change inside a transaction.
        self.connection
            .execute_batch("PRAGMA journal_mode = WAL;")
            .map_err(database_error(&self.path, "configure database"))?;
        let transaction = self
            .connection
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(database_error(&self.path, "begin initialization"))?;
        // A concurrent creator may have committed between the unlocked check and this lock.
        prepare_write_blocking(&transaction, &self.path)?;
        if occupied_blocking(&transaction, &self.path)? {
            return Err(StoreError::AlreadyInitialized(self.path.clone()));
        }
        snapshot::write_initial_blocking(&transaction, &self.path, plan)?;
        transaction
            .commit()
            .map_err(database_error(&self.path, "commit initialization"))?;
        Ok(())
    }

    /// Atomically append one valid operation and the exact snapshot produced by that command,
    /// under this file's lineage.
    ///
    /// Inside the write transaction, an identity that is already recorded is answered with
    /// [`StoreError::DuplicateOperation`] before any precondition, an archive is refused, and so
    /// is a lineage other than `expected_lineage` or a revision other than the operation's base.
    pub fn persist_blocking(
        &mut self,
        plan: &Plan,
        operation: &Operation,
        expected_lineage: Option<LineageId>,
    ) -> Result<HistoryEntry, StoreError> {
        if operation.base_revision.wrapping_add(1) != operation.resulting_revision
            || plan.revision != operation.resulting_revision
        {
            return Err(StoreError::InvalidOperationRevision {
                base: operation.base_revision,
                result: operation.resulting_revision,
                plan: plan.revision,
            });
        }
        plan.validate()?;
        let transaction = self
            .connection
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(database_error(&self.path, "begin operation"))?;
        if prepare_write_blocking(&transaction, &self.path)? == Layout::Empty {
            return Err(StoreError::NotInitialized(self.path.clone()));
        }
        if let Some(recorded) = history::recorded_blocking(&transaction, &self.path, operation.id)?
        {
            return Err(StoreError::DuplicateOperation {
                recorded: Box::new(recorded),
            });
        }
        let found = lineage::read_blocking(&transaction, &self.path)?;
        found.check_writable(expected_lineage, &self.path)?;
        let mut expected = snapshot::load_cached_blocking(&transaction, &self.path, &self.cache)?
            .ok_or_else(|| StoreError::NotInitialized(self.path.clone()))?;
        if expected.revision != operation.base_revision {
            return Err(StoreError::RevisionConflict {
                expected: operation.base_revision,
                actual: expected.revision,
            });
        }
        apply_command(
            &mut expected,
            operation.actor.clone(),
            operation.command.clone(),
            operation.timestamp,
            operation.id,
        )?;
        if expected != *plan {
            return Err(StoreError::SnapshotMismatch(operation.id));
        }
        let sequence = snapshot::write_operation_blocking(
            &transaction,
            &self.path,
            plan,
            operation,
            found.lineage_id,
        )?;
        transaction
            .commit()
            .map_err(database_error(&self.path, "commit operation"))?;
        Ok(HistoryEntry {
            sequence,
            operation: RecordedOperation {
                operation: operation.clone(),
                workspace_id: plan.workspace.id,
                lineage_id: found.lineage_id,
            },
        })
    }

    /// Count committed semantic operations.
    pub fn operation_count_blocking(&self) -> Result<u64, StoreError> {
        if schema::check_blocking(&self.connection, &self.path)? == Layout::Empty {
            return Ok(0);
        }
        let count: u64 = self
            .connection
            .query_row("SELECT COUNT(*) FROM operations", [], |row| row.get(0))
            .map_err(database_error(&self.path, "count operations"))?;
        Ok(count)
    }
}

/// Check the version and layout of an existing file through a read-only connection first.
///
/// Closing the last read-write connection to a WAL store checkpoints the log into the main file and
/// deletes the side files, so a store this binary refuses must never be opened read-write at all.
/// Revision and lineage of the store at `path` through a read-only connection, without decoding
/// its snapshot, for a client polling a store it has not opened (a repointable project locator);
/// `None` for an uninitialized store. Stores in another layout are refused as by an open.
pub fn store_revision_blocking(path: &Path) -> Result<Option<crate::StoreRevision>, StoreError> {
    let mode = files::read_mode_blocking(path)?;
    let connection = files::open_read_only_blocking(path, mode)?;
    read_revision_blocking(&connection, path)
}

fn read_revision_blocking(
    connection: &Connection,
    path: &Path,
) -> Result<Option<crate::StoreRevision>, StoreError> {
    if schema::check_blocking(connection, path)? == Layout::Empty {
        return Ok(None);
    }
    let transaction = connection
        .unchecked_transaction()
        .map_err(database_error(path, "begin revision query"))?;
    let revision = snapshot::revision_blocking(&transaction, path)?;
    let lineage = lineage::read_blocking(&transaction, path)?;
    let history_head: u64 = transaction
        .query_row(
            "SELECT COALESCE(MAX(sequence), 0) FROM operations",
            [],
            |row| row.get(0),
        )
        .map_err(database_error(path, "read history head"))?;
    transaction
        .commit()
        .map_err(database_error(path, "finish revision query"))?;
    Ok(revision.map(|revision| crate::StoreRevision {
        revision,
        lineage,
        history_head,
    }))
}

fn check_existing_read_only_blocking(path: &Path) -> Result<(), StoreError> {
    let exists = path.try_exists().map_err(|source| StoreError::Io {
        action: "inspect",
        path: path.to_path_buf(),
        source,
    })?;
    if !exists {
        return Ok(());
    }
    let mode = files::read_mode_blocking(path)?;
    let connection = files::open_read_only_blocking(path, mode)?;
    if schema::check_blocking(&connection, path)? == Layout::Current {
        snapshot::load_blocking(&connection, path)?;
        snapshot::genesis_blocking(&connection, path)?;
        lineage::read_blocking(&connection, path)?;
    }
    Ok(())
}

fn occupied_blocking(connection: &Connection, path: &Path) -> Result<bool, StoreError> {
    if schema::check_blocking(connection, path)? == Layout::Empty {
        return Ok(false);
    }
    connection
        .query_row(
            "SELECT EXISTS(SELECT 1 FROM plan_state) OR EXISTS(SELECT 1 FROM operations)",
            [],
            |row| row.get(0),
        )
        .map_err(database_error(path, "check initialization"))
}

/// Create the schema of an empty store inside a caller-owned write transaction and return the
/// layout found before that.
///
/// The exact layout is re-checked under the write lock, so a schema object added by another
/// process after open never runs inside this write.
fn prepare_write_blocking(connection: &Connection, path: &Path) -> Result<Layout, StoreError> {
    let found = schema::check_blocking(connection, path)?;
    if found == Layout::Empty {
        schema::create_blocking(connection, path)?;
    }
    Ok(found)
}

#[cfg(test)]
mod tests;