//! 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.
//!
//! # 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.

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};

use threadpool::ThreadPool;

pub use changehook::ChangeLogHook;
pub use err::Error;
pub use rawhook::Hook;
pub use wrconn::WrConn;


/// Used to register application callbacks to set up database schema.
pub trait SchemaMgr {
  /// Called just after the writer connection has been created and is intended
  /// to perform database initialization (create tables, add predefined rows,
  /// etc).
  ///
  /// `newdb` will be `true` if the database file did not exist prior to
  /// initialization.
  ///
  /// While this method can be used to perform schema upgrades, there are two
  /// specialized methods (`need_upgrade()` and `upgrade()`) that can be used
  /// for this purpose instead.
  ///
  /// The default implementation does nothing but returns `Ok(())`.
  #[allow(unused_variables)]
  fn init(&self, conn: &mut Connection, newdb: bool) -> Result<(), Error> {
    Ok(())
  }

  /// Application callback used to determine if the database schema is out of
  /// date and needs to be updated.
  ///
  /// The default implementation does nothing but returns `Ok(false)`.
  #[allow(unused_variables)]
  fn need_upgrade(&self, conn: &Connection) -> Result<bool, Error> {
    Ok(false)
  }

  /// Upgrade the database schema.
  ///
  /// This is called if [`SchemaMgr::need_upgrade()`] returns `Ok(true)`.
  ///
  /// The default implementation does nothing but returns `Ok(())`.
  #[allow(unused_variables)]
  fn upgrade(&self, conn: &mut Connection) -> Result<(), Error> {
    Ok(())
  }
}


#[derive(Clone, Debug)]
struct AutoClean {
  /// The amount of dirt that must have accumulated before an incremental
  /// vacuum is triggered.
  ///
  /// If this is set to `0` the incremental vacuum will be run at every
  /// opportunity.
  dirt_threshold: usize,

  /// The number of pages to process from the freelist each time an
  /// incremental autovacuum is triggered.
  ///
  /// If this is `None` then all pages will be processed.
  npages: Option<NonZeroUsize>
}

/// Read-only connection.
#[derive(Debug)]
struct RoConn {}

impl CustomizeConnection<rusqlite::Connection, rusqlite::Error> for RoConn {
  fn on_acquire(
    &self,
    conn: &mut rusqlite::Connection
  ) -> Result<(), rusqlite::Error> {
    conn.pragma_update(None, "foreign_keys", "ON")?;
    Ok(())
  }

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


/// Builder for constructing a [`ConnPool`] object.
pub struct Builder {
  schmgr: Box<dyn SchemaMgr>,
  full_vacuum: bool,
  max_readers: usize,
  autoclean: Option<AutoClean>,
  hook: Option<Arc<dyn Hook + Send + Sync>>
}

/// Internal methods.
impl Builder {
  /// Open the writer connection.
  fn open_writer(&self, fname: &Path) -> Result<Connection, rusqlite::Error> {
    let conn = Connection::open(fname)?;
    conn.pragma_update(None, "journal_mode", "WAL")?;
    conn.pragma_update(None, "foreign_keys", "ON")?;

    // Only enable incremental auto vacuum if a dirt watermark has been
    // configured.
    if self.autoclean.is_some() {
      conn.pragma_update(None, "auto_vacuum", "INCREMENTAL")?;
    }

    Ok(conn)
  }

  /// Run a full vacuum.
  fn full_vacuum(&self, conn: &Connection) -> Result<(), rusqlite::Error> {
    conn.execute("VACUUM;", params![])?;
    Ok(())
  }

  fn create_ro_pool(
    &self,
    fname: &Path
  ) -> Result<r2d2::Pool<SqliteConnectionManager>, r2d2::Error> {
    let fl =
      OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX;
    let manager = SqliteConnectionManager::file(fname).with_flags(fl);
    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)
  }
}


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.
  pub fn new(schmgr: Box<dyn SchemaMgr>) -> Self {
    Self {
      schmgr,
      full_vacuum: false,
      max_readers: 2,
      autoclean: None,
      hook: None
    }
  }

  /// Trigger a full vacuum when initializing the connection pool.
  ///
  /// Operates on an owned `Builder` object.
  pub fn init_vacuum(mut self) -> Self {
    self.full_vacuum = true;
    self
  }

  /// Trigger a full vacuum when initializing the connection pool.
  ///
  /// Operates on a borrowed `Builder` object.
  pub fn init_vacuum_r(&mut self) -> &mut Self {
    self.full_vacuum = true;
    self
  }

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

  /// 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
  }

  /// 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
  }

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

  /// Enable incremental autovacuum.
  ///
  /// `dirt_watermark` is used to set what amount of "dirt" is required in
  /// order to trigger an autoclean.  `nfree` is the number of blocks in the
  /// freelists to process each time the autoclean is run.
  pub fn incremental_autoclean(
    mut self,
    dirt_watermark: usize,
    npages: Option<NonZeroUsize>
  ) -> Self {
    self.autoclean = Some(AutoClean {
      dirt_threshold: dirt_watermark,
      npages
    });
    self
  }

  /// Construct a connection pool.
  pub fn build<P>(self, fname: P) -> Result<ConnPool, Error>
  where
    P: AsRef<Path>
  {
    // ToDo: Use std::path::absolute() once stabilized
    let fname = fname.as_ref();
    let db_exists = fname.exists();

    //
    // Set up the read/write connection
    //
    // This must be done before creating the read-only connection pool, because
    // at that point the database file must already exist.
    //
    let mut conn = self.open_writer(fname)?;

    //
    // Perform schema initialization.
    //
    // This must be done after auto_vacuum is set, because auto_vacuum requires
    // configuration before any tables have been created.
    // See: https://www.sqlite.org/pragma.html#pragma_auto_vacuum
    //
    self.schmgr.init(&mut conn, !db_exists)?;
    if self.schmgr.need_upgrade(&conn)? {
      self.schmgr.upgrade(&mut conn)?;
    }

    //
    // Perform a full vacuum if requested to do so.
    //
    if self.full_vacuum {
      self.full_vacuum(&conn)?;
    }

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

    //
    // Create a threadpool with one thread for each reader and the writer
    //
    let tpool = ThreadPool::new(self.max_readers + 1);

    //
    // 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, tpool })
  }


  /// Construct a connection pool.
  ///
  /// Same as [`Builder::build()`], but register a change log callback on the
  /// writer as well.
  ///
  /// This method should not be called if the application has requested to add
  /// a raw update hook.
  ///
  /// # Panic
  /// This method will panic if a hook has been added to the Builder.
  pub fn build_with_changelog_hook<P, D, T>(
    self,
    fname: P,
    hook: Box<dyn ChangeLogHook<Database = D, Table = T> + Send>
  ) -> Result<ConnPool, Error>
  where
    P: AsRef<Path>,
    D: FromStr + Send + Sized + 'static,
    T: FromStr + Send + Sized + 'static
  {
    if self.hook.is_some() {
      panic!(
        "Can't build a connection pool with both a raw and changelog hook"
      );
    }

    // ToDo: Use std::path::absolute() once stabilized
    let fname = fname.as_ref();
    let db_exists = fname.exists();

    //
    // Set up the read/write connection
    //
    // This must be done before creating the read-only connection pool, because
    // at that point the database file must already exist.
    //
    let mut conn = self.open_writer(fname)?;

    //
    // Perform schema initialization.
    //
    // This must be done after auto_vacuum is set, because auto_vacuum requires
    // configuration before any tables have been created.
    // See: https://www.sqlite.org/pragma.html#pragma_auto_vacuum
    //
    self.schmgr.init(&mut conn, !db_exists)?;
    if self.schmgr.need_upgrade(&conn)? {
      self.schmgr.upgrade(&mut conn)?;
    }

    //
    // Perform a full vacuum if requested to do so.
    //
    if self.full_vacuum {
      self.full_vacuum(&conn)?;
    }

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

    //
    // Create a threadpool with one thread for each reader and the writer
    //
    let tpool = ThreadPool::new(self.max_readers + 1);

    //
    // 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, tpool })
  }
}


/// Inner writer connection object.
///
/// When the writer is acquired from the connection pool it passes an instance
/// of this struct to the WrConn object.
struct InnerWrConn {
  /// The writer connection.
  conn: Connection,

  /// Amount of accumulated dirt.
  ///
  /// Only used when the autoclean feature is used.
  dirt: usize
}


struct Inner {
  /// The writer connection.
  ///
  /// This is `None` when a `WrConn` exists, and is set to `Some()` by
  /// `WrConn`'s Drop implementation.
  conn: Option<InnerWrConn>
}

struct Shared {
  inner: Mutex<Inner>,
  signal: Condvar,
  autoclean: Option<AutoClean>
}


/// 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>,
  tpool: ThreadPool
}

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

  /// Acquire the read/write connection.
  ///
  /// If the writer is already taken, then block and wait for it to become
  /// available.
  pub fn writer(&self) -> WrConn {
    let mut g = self.sh.inner.lock();
    let conn = loop {
      if let Some(conn) = g.conn.take() {
        break conn;
      } else {
        self.sh.signal.wait(&mut g);
      }
    };

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

  /// Attempt to acquire the writer connection.
  ///
  /// Returns `Some(conn)` if the writer connection was available at the time
  /// of the request.  Returns `None` if the writer has already been taken.
  pub fn try_writer(&self) -> Option<WrConn> {
    let mut g = self.sh.inner.lock();
    let Some(conn) = g.conn.take() else {
      return None;
    };

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

  /// Run a closure with a read-only connection
  pub fn ro_run<F>(&self, f: F) -> Result<(), r2d2::Error>
  where
    F: FnOnce(&Connection) + Send + 'static
  {
    let roconn = self.reader()?;

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

    Ok(())
  }

  /// Run a closure with read-only connection, returning a channel end-point
  /// for retreiving result.
  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 roconn = self.reader()?;

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

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

    Ok(wctx)
  }

  /// Run a closure with read/write connection
  pub fn rw_run<F>(&self, f: F)
  where
    F: FnOnce(&mut Connection) + Send + 'static
  {
    let mut rwconn = self.writer();

    self.tpool.execute(move || f(&mut rwconn));
  }

  /// Run a closure with read/write connection, returning a channel end-point
  /// for retreiving result.
  pub fn rw_run_result<F, T, E>(&self, f: F) -> swctx::WaitCtx<T, (), E>
  where
    F: FnOnce(&mut Connection, swctx::SetCtx<T, (), E>) + Send + 'static,
    T: Send + 'static,
    E: Send + 'static
  {
    let mut rwconn = self.writer();

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

    self.tpool.execute(move || f(&mut rwconn, sctx));

    wctx
  }

  /// Perform an incremental vacuum.
  ///
  /// `n` is the number of freelist nodes to reclaim.  If `None` all nodes will
  /// be reclaimed.
  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)
      }
    })
  }

  /// Consume self and wait for all threads in thread pool to complete.
  pub fn shutdown(self) {
    self.tpool.join();
  }
}

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