[package]
name = "sqlsrv"
version = "0.1.0"
edition = "2021"
license = "0BSD"
categories = [ "database" ]
keywords = [ "sqlite", "server" ]
repository = "https://repos.qrnch.tech/pub/sqlsrv"
description = "Utility functions for managing SQLite connections in a server application."
rust-version = "1.56"
exclude = [
  ".fossil-settings",
  ".efiles",
  ".fslckout",
  "build_docs.sh",
  "examples",
  "www",
  "rustfmt.toml"
]

[features]
default = ["tpool"]
tpool = ["dep:swctx", "dep:threadpool"]

[dependencies]
parking_lot = { version = "0.12.1" }
r2d2 = { version = "0.8.10" }
r2d2_sqlite = { version = "0.23.0" }
rusqlite = { version = "0.30.0", features = ["hooks"] }
swctx = { version = "0.2.2", optional = true }
threadpool = { version = "1.8.1", optional = true }

[package.metadata.docs.rs]
all-features = true
rustdoc-args = ["--cfg", "docsrs", "--generate-link-to-definition"]

#!/bin/sh

RUSTFLAGS="--cfg docsrs" RUSTDOCFLAGS="--cfg docsrs" \
cargo +nightly doc --all-features

# vim: set ft=sh et sw=2 ts=2 sts=2 cinoptions=2 tw=79 :

  id   INTEGER  PRIMARY KEY,
  data TEXT     UNIQUE NOT NULL,
  ts   DATETIME DEFAULT CURRENT_TIMESTAMP,
  CHECK (length(data) == 64)
);",
        &[] as &[&dyn ToSql]
      )?;

      conn.execute(
        "CREATE TABLE IF NOT EXISTS whoop (
  id   INTEGER  PRIMARY KEY,
  name TEXT,
  ts   DATETIME DEFAULT CURRENT_TIMESTAMP
);",
        &[] as &[&dyn ToSql]
      )?;
    }

    Ok(())
  }
}


fn main() {
  let schema = Box::new(Schema {});
  #[allow(unused_mut)]
  let mut bldr = sqlsrv::Builder::new(schema).incremental_autoclean(10, None);

  #[cfg(feature = "tpool")]
  bldr.worker_threads_r(None);

  let connpool = bldr.build("test.sqlite").unwrap();

  // Acquire the writer connection and use it to conditionally create the
  // `stuff` table.
  {
    let wconn = connpool.writer();
    wconn

    let rconn = connpool.reader().unwrap();
    let mut stmt = rconn.prepare_cached(SQL_LOOKUP_TABLE).unwrap();
    let have = stmt
      .query_row(params!("stuff"), |row| row.get::<usize, bool>(0))
      .unwrap();
    assert!(have);
  }

  #[cfg(feature = "tpool")]
  connpool
    .ro_run(|conn| {
      const SQL: &str = "SELECT * FROM snarks;";
      let mut _stmt = conn.prepare_cached(SQL).unwrap();
    })
    .unwrap();

  #[cfg(feature = "tpool")]
  connpool.rw_run(|conn| {
    const SQL: &str = "INSERT INTO whoop (name) VALUES (?);";
    let mut stmt = conn.prepare_cached(SQL).unwrap();
    stmt.execute(params!["test"]).unwrap();
    Some(1)
  });

  #[cfg(feature = "tpool")]
  connpool.shutdown();
}

// vim: set ft=rust et sw=2 ts=2 sts=2 cinoptions=2 tw=79 :

//! A library for implementing an in-process SQLite database server.
//!
//! # Connection pooling
//! sqlsrv implements connection pooling that reflects the concurrency model
//! for SQLite:  It supports only one writer but multiple readers.
//!
//! # Connection task pooling
//! In addition to pooling connections, the library can pool threads to run
//! database operations on.
//!
//! # Incremental auto-clean
//! The connection pool has built-in support for setting up incremental
//! autovacuum, and can be configured to implicitly run incremental vacuuming.
//!
//! To use this feature, a "maximum dirt" value is configured on the connection
//! pool.  Whenever the writer connection performs changes to the database it
//! can add "dirt" to the connection.  When the writer connection is returned
//! to the connection pool it checks to see if the amount of dirt is equal to
//! or greater than the configured "maximum dirt" threshold.  If the threshold
//! has been reached, an incremental autovacuum is performed.

#![cfg_attr(docsrs, feature(doc_cfg))]

mod changehook;
mod err;
mod rawhook;
mod wrconn;

use std::{
  mem::ManuallyDrop, num::NonZeroUsize, path::Path, str::FromStr, sync::Arc
};

use parking_lot::{Condvar, Mutex};

use r2d2::{CustomizeConnection, PooledConnection};

use r2d2_sqlite::SqliteConnectionManager;

pub use rusqlite;

use rusqlite::{params, Connection, OpenFlags};

#[cfg(feature = "tpool")]
use threadpool::ThreadPool;

pub use changehook::ChangeLogHook;
pub use err::Error;
pub use rawhook::{Action, Hook};
pub use wrconn::WrConn;

    conn.pragma_update(None, "foreign_keys", "ON")?;
    Ok(())
  }

  fn on_release(&self, _conn: rusqlite::Connection) {}
}


#[cfg(feature = "tpool")]
#[derive(Default)]
enum ThrdPool {
  #[default]
  Disable,
  Enable {
    nthreads: Option<usize>
  }
}

/// Builder for constructing a [`ConnPool`] object.
pub struct Builder {
  schmgr: Box<dyn SchemaMgr>,
  full_vacuum: bool,
  max_readers: usize,
  #[cfg(feature = "tpool")]
  thrdpool: ThrdPool,
  autoclean: Option<AutoClean>,
  hook: Option<Arc<dyn Hook + Send + Sync>>
}

/// Internal methods.
impl Builder {
  /// Open the writer connection.

    let roconn_initterm = RoConn {};
    let max_readers = u32::try_from(self.max_readers).unwrap();
    r2d2::Pool::builder()
      .max_size(max_readers)
      .connection_customizer(Box::new(roconn_initterm))
      .build(manager)
  }

  #[cfg(feature = "tpool")]
  fn init_tpool(&self) -> Option<ThreadPool> {
    match self.thrdpool {
      ThrdPool::Disable => None,
      ThrdPool::Enable { nthreads } => {
        let nthreads = if let Some(nthreads) = nthreads {
          nthreads
        } else {
          self.max_readers + 1
        };
        let tpool = ThreadPool::new(nthreads);
        Some(tpool)
      }
    }
  }
}


impl Builder {
  /// Create a new `Builder` for constructing a [`ConnPool`] object.
  ///
  /// Default to not run a full vacuum of the database on initialization and
  /// create 2 read-only connections for the pool.
  /// No workers thread pool will be used.
  pub fn new(schmgr: Box<dyn SchemaMgr>) -> Self {
    Self {
      schmgr,
      full_vacuum: false,
      max_readers: 2,
      #[cfg(feature = "tpool")]
      thrdpool: ThrdPool::default(),
      autoclean: None,
      hook: None
    }
  }

  /// Trigger a full vacuum when initializing the connection pool.
  ///

  /// Set maximum number of readers in the connection pool.
  ///
  /// Operates on a borrowed `Builder` object.
  pub fn max_readers_r(&mut self, n: usize) -> &mut Self {
    self.max_readers = n;
    self
  }

  /// Enable a thread pool for running connection tasks on.
  ///
  /// Unless this is called, no thread pool will be allocated and the
  /// [`ConnPool::ro_run()`] and [`ConnPool::rw_run()`] (and associated
  /// methods) will panic.
  ///
  /// The `nthreads` can be used to specify the number of worker threads to
  /// allocate.  If this is `None`, the number of threads will default to the
  /// number of reader connections plus one (for the writer).
  ///
  /// # Panic
  /// Panics if `nthreads` is set to `Some(0)`.
  #[cfg(feature = "tpool")]
  #[cfg_attr(docsrs, doc(cfg(feature = "tpool")))]
  pub fn worker_threads(mut self, nthreads: Option<usize>) -> Self {
    self.worker_threads_r(nthreads);
    self
  }

  /// Enable a thread pool for running connection tasks on.
  ///
  /// This does the same as [`Builder::worker_threads()`], but operates on a
  /// borrowed object.
  #[cfg(feature = "tpool")]
  #[cfg_attr(docsrs, doc(cfg(feature = "tpool")))]
  pub fn worker_threads_r(&mut self, nthreads: Option<usize>) -> &mut Self {
    assert_ne!(nthreads, Some(0));
    self.thrdpool = ThrdPool::Enable { nthreads };
    self
  }

  /// Request that a "raw" update hook be added to the writer connection.
  ///
  /// Operates on an owned `Builder` object.
  pub fn hook(mut self, hook: Arc<dyn Hook + Send + Sync>) -> Self {
    self.hook = Some(hook);
    self

    // Register a callback hook
    //
    if let Some(ref hook) = self.hook {
      rawhook::hook(&conn, Arc::clone(hook));
    }

    //
    // Create thread pool if requested to do so
    //
    #[cfg(feature = "tpool")]
    let tpool = self.init_tpool();

    //
    // Set up connection pool for read-only connections.
    //
    let rpool = self.create_ro_pool(fname)?;

    //
    // Prepare shared data
    //
    let iconn = InnerWrConn { conn, dirt: 0 };
    let inner = Inner { conn: Some(iconn) };
    let sh = Arc::new(Shared {
      inner: Mutex::new(inner),
      signal: Condvar::new(),
      autoclean: self.autoclean.clone()
    });

    Ok(ConnPool {
      rpool,
      sh,
      #[cfg(feature = "tpool")]
      tpool
    })
  }


  /// Construct a connection pool.
  ///
  /// Same as [`Builder::build()`], but register a change log callback on the
  /// writer as well.

    //
    // Register a callback hook
    //
    changehook::hook(&conn, hook);

    //
    // Create thread pool if requested to do so
    //
    #[cfg(feature = "tpool")]
    let tpool = self.init_tpool();

    //
    // Set up connection pool for read-only connections.
    //
    let rpool = self.create_ro_pool(fname)?;

    //
    // Prepare shared data
    //
    let iconn = InnerWrConn { conn, dirt: 0 };
    let inner = Inner { conn: Some(iconn) };
    let sh = Arc::new(Shared {
      inner: Mutex::new(inner),
      signal: Condvar::new(),
      autoclean: self.autoclean.clone()
    });

    Ok(ConnPool {
      rpool,
      sh,
      #[cfg(feature = "tpool")]
      tpool
    })
  }
}


/// Inner writer connection object.
///
/// When the writer is acquired from the connection pool it passes an instance

/// SQLite connection pool.
///
/// This is a somewhat specialized connection pool that only allows a single
/// writer but multiple readers.
pub struct ConnPool {
  sh: Arc<Shared>,
  rpool: r2d2::Pool<SqliteConnectionManager>,
  #[cfg(feature = "tpool")]
  tpool: Option<ThreadPool>
}

impl ConnPool {
  /// Acquire a read-only connection.
  pub fn reader(
    &self
  ) -> Result<PooledConnection<SqliteConnectionManager>, r2d2::Error> {

    Some(WrConn {
      sh: Arc::clone(&self.sh),
      inner: ManuallyDrop::new(conn)
    })
  }

  /// Run a closure with a read-only connection
  ///
  /// # Panic
  /// Panics if the connection pool was not configured to use a thread pool.
  #[cfg(feature = "tpool")]
  #[cfg_attr(docsrs, doc(cfg(feature = "tpool")))]
  pub fn ro_run<F>(&self, f: F) -> Result<(), r2d2::Error>
  where
    F: FnOnce(&Connection) + Send + 'static
  {
    let Some(ref tpool) = self.tpool else {
      panic!("Connection pool does not have a thread pool");
    };

    let roconn = self.reader()?;

    tpool.execute(move || f(&roconn));

    Ok(())
  }

  /// Run a closure with read-only connection, returning a channel end-point
  /// for retreiving result.
  ///
  /// # Panic
  /// Panics if the connection pool was not configured to use a thread pool.
  #[cfg(feature = "tpool")]
  #[cfg_attr(docsrs, doc(cfg(feature = "tpool")))]
  pub fn ro_run_result<F, T, E>(
    &self,
    f: F
  ) -> Result<swctx::WaitCtx<T, (), E>, r2d2::Error>
  where
    F: FnOnce(&Connection, swctx::SetCtx<T, (), E>) + Send + 'static,
    T: Send + 'static,
    E: Send + 'static
  {
    let Some(ref tpool) = self.tpool else {
      panic!("Connection pool does not have a thread pool");
    };

    let roconn = self.reader()?;

    let (sctx, wctx) = swctx::mkpair();

    tpool.execute(move || f(&roconn, sctx));

    Ok(wctx)
  }

  /// Run a closure with read/write connection
  ///
  /// # Panic
  /// Panics if the connection pool was not configured to use a thread pool.
  #[cfg(feature = "tpool")]
  #[cfg_attr(docsrs, doc(cfg(feature = "tpool")))]
  pub fn rw_run<F>(&self, f: F)
  where
    F: FnOnce(&mut Connection) -> Option<usize> + Send + 'static
  {
    let Some(ref tpool) = self.tpool else {
      panic!("Connection pool does not have a thread pool");
    };

    let mut rwconn = self.writer();

    tpool.execute(move || {
      let dirt = f(&mut rwconn);
      if let Some(dirt) = dirt {
        rwconn.add_dirt(dirt);
      }
    });
  }


  /// Run a closure with read/write connection, returning a channel end-point
  /// for retreiving result.
  ///
  /// # Panic
  /// Panics if the connection pool was not configured to use a thread pool.
  #[cfg(feature = "tpool")]
  #[cfg_attr(docsrs, doc(cfg(feature = "tpool")))]
  pub fn rw_run_result<F, T, E>(&self, f: F) -> swctx::WaitCtx<T, (), E>
  where
    F: FnOnce(&mut Connection, swctx::SetCtx<T, (), E>) -> Option<usize>
      + Send
      + 'static,
    T: Send + 'static,
    E: Send + 'static
  {
    let Some(ref tpool) = self.tpool else {
      panic!("Connection pool does not have a thread pool");
    };

    let mut rwconn = self.writer();

    let (sctx, wctx) = swctx::mkpair();

    tpool.execute(move || {
      let dirt = f(&mut rwconn, sctx);
      if let Some(dirt) = dirt {
        rwconn.add_dirt(dirt);
      }
    });

    wctx
  }

  /// Perform an incremental vacuum.
  ///
  /// `n` is the number of freelist nodes to reclaim.  If `None` all nodes will
  /// be reclaimed.
  #[cfg(feature = "tpool")]
  #[cfg_attr(docsrs, doc(cfg(feature = "tpool")))]
  pub fn incremental_vacuum(
    &self,
    n: Option<usize>
  ) -> swctx::WaitCtx<(), (), rusqlite::Error> {
    self.rw_run_result(move |wconn, sctx| {
      let res = if let Some(n) = n {
        wconn.execute("PRAGMA incremental_vacuum(?);", params![n])
      } else {
        wconn.execute("PRAGMA incremental_vacuum;", params![])
      };
      match res {
        Ok(_) => sctx.set(()),
        Err(e) => sctx.fail(e)
      }

      None
    })
  }

  /// Consume self and wait for all threads in thread pool to complete.
  #[cfg(feature = "tpool")]
  #[cfg_attr(docsrs, doc(cfg(feature = "tpool")))]
  pub fn shutdown(self) {
    if let Some(tpool) = self.tpool {
      tpool.join();
    }
  }
}

// vim: set ft=rust et sw=2 ts=2 sts=2 cinoptions=2 tw=79 :

# Change Log

## [Unreleased]

[Details](/vdiff?from=sqlsrv-0.1.0&to=trunk)

### Added

### Changed

### Removed

---

## [0.1.0] - 2024-01-21

[Details](/vdiff?from=sqlsrv-0.0.4&to=sqlsrv-0.1.0)

### Changed

- Make the internal thread pool optional.  It is disabled by default.
- Put thread pool features behind a ferature gate (`tpool`), which is enabled
  by default.

---

## [0.0.4] - 2024-01-19

[Details](/vdiff?from=sqlsrv-0.0.3&to=sqlsrv-0.0.4)

### Changed

- Allow `dirt` to be returned from read/write thread wrappers.