Something went wrong. Try again.
Chris’s PDS implementation
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174//! `SQLite` persistence, divided by the domain state each query serves.
mod account;mod blob;mod blockstore;mod oauth;mod query;mod record;mod repo;mod schema;
use std::{ path::Path, sync::{Arc, Mutex, MutexGuard},};
use rusqlite::{Connection, MAIN_DB, OpenFlags, Params, Row, Transaction, TransactionBehavior};
use crate::error::{Error, Result};
pub(crate) use crate::clock::now;pub(crate) use account::{parse_account_status, require_account_active};pub(crate) use blob::active_blob_ref_cids;pub(crate) use blockstore::RepoBlockStore;pub(crate) use query::{constraint_conflict, query, query_optional, sql_row};pub(crate) use repo::{ LifecycleEvent, ListedRepoHead, PendingRepoCommit, RepoHead, StoredAccountEvent, StoredCommitEvent, StoredIdentityEvent, StoredRepoEvent,};
#[derive(Clone)]pub struct Database { connection: Arc<Mutex<Connection>>,}
impl Database { pub(crate) fn query<T>( &self, sql: &str, params: impl Params, decode: impl FnMut(&Row<'_>) -> Result<T>, ) -> Result<Vec<T>> { query(&*self.connection()?, sql, params, decode) }
pub(crate) fn query_one<T>( &self, sql: &str, params: impl Params, decode: impl FnOnce(&Row<'_>) -> Result<T>, ) -> Result<T> { self.connection()?.query_row_and_then(sql, params, decode) }
pub(crate) fn query_optional<T>( &self, sql: &str, params: impl Params, decode: impl FnOnce(&Row<'_>) -> Result<T>, ) -> Result<Option<T>> { query_optional(&*self.connection()?, sql, params, decode) }
pub(crate) fn transaction<T>( &self, operation: impl FnOnce(&Transaction<'_>) -> Result<T>, ) -> Result<T> { let mut connection = self.connection()?; let transaction = connection.transaction()?; let value = operation(&transaction)?; transaction.commit()?; Ok(value) }
/// Open or create a database file and apply the current schema. /// /// # Errors /// /// Returns an error if `SQLite` cannot open or migrate the database. pub fn open(path: impl AsRef<Path>) -> Result<Self> { Self::from_connection(Connection::open(path)?, true) }
/// Open and migrate an existing database without creating a replacement. /// /// # Errors /// /// Returns an error if the path does not contain initialized CPDS storage or /// cannot be opened and migrated. pub fn open_existing(path: impl AsRef<Path>) -> Result<Self> { let connection = Connection::open_with_flags(path, OpenFlags::SQLITE_OPEN_READ_WRITE)?; Self::from_connection(connection, false) }
/// Create an isolated database suitable for tests and ephemeral servers. /// /// # Errors /// /// Returns an error if the in-memory database cannot be initialized. pub fn open_in_memory() -> Result<Self> { Self::from_connection(Connection::open_in_memory()?, true) }
fn from_connection(mut connection: Connection, initialize: bool) -> Result<Self> { schema::apply(&mut connection, initialize)?; Ok(Self { connection: Arc::new(Mutex::new(connection)), }) }
pub(crate) fn connection(&self) -> Result<MutexGuard<'_, Connection>> { self.connection .lock() .map_err(|_| Error::Internal("database lock was poisoned".to_owned())) }
/// Verify that the live connection and expected schema remain available. /// /// # Errors /// /// Returns an error when the connection cannot be locked or queried, its /// policy changed, or its schema version is no longer supported. pub fn check(&self) -> Result<()> { let connection = self.connection()?; schema::check(&connection)?; connection.query_row( "SELECT COALESCE(MAX(seq), 0), MIN(created_at) FROM repo_outbox", [], |row| <(i64, Option<i64>)>::try_from(row), )?; Ok(()) }
/// Checkpoint committed WAL frames and truncate the WAL after request drain. /// /// # Errors /// /// Returns an error when the database cannot be locked or the checkpoint /// remains busy after all HTTP work has drained. pub fn checkpoint(&self) -> Result<()> { let busy: i64 = self.connection()? .query_row("PRAGMA wal_checkpoint(TRUNCATE)", [], |row| row.get(0))?; if busy != 0 { return Err(Error::Internal( "SQLite WAL checkpoint remained busy after request drain".to_owned(), )); } Ok(()) }
/// Copy a consistent `SQLite` snapshot while holding writes around related /// storage capture. /// /// # Errors /// /// Returns an error when the write lock, `SQLite` backup, related capture, or /// final transaction commit fails. pub(crate) fn backup_with( &self, source: impl AsRef<Path>, destination: impl AsRef<Path>, capture_related: impl FnOnce() -> Result<()>, ) -> Result<()> { let mut connection = self.connection()?; let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?; let snapshot = Connection::open_with_flags(source, OpenFlags::SQLITE_OPEN_READ_ONLY)?; snapshot.backup(MAIN_DB, destination, None)?; capture_related()?; transaction.commit()?; Ok(()) }}