sqlite-collections 0.1.0

use crate::format::Format;
use crate::OpenError;
use crate::Savepointable;
use crate::{db, identifier::Identifier};
use rusqlite::{params, Connection, OptionalExtension, Savepoint};

use std::marker::PhantomData;

mod error;
mod iter;
pub use iter::Iter;

pub use error::Error;

#[derive(Debug, Clone, PartialEq, PartialOrd, Ord, Eq, Hash)]
pub struct Config<'db, 'tbl> {
    pub database: Identifier<'db>,
    pub table: Identifier<'tbl>,
}

impl Default for Config<'static, 'static> {
    fn default() -> Self {
        Config {
            database: "main".try_into().unwrap(),
            table: "ds::set".try_into().unwrap(),
        }
    }
}

/// Deterministic store set.
pub struct Set<'db, 'tbl, S, C>
where
    S: Format,
    C: Savepointable,
{
    connection: C,
    database: Identifier<'db>,
    table: Identifier<'tbl>,
    serializer: PhantomData<S>,
}

impl<S, C> Set<'static, 'static, S, C>
where
    S: Format,
    C: Savepointable,
{
    pub fn open(connection: C) -> Result<Self, OpenError> {
        Set::open_with_config(connection, Config::default())
    }

    /// Open a set without creating it or checking if it exists.  This is safe
    /// if you call a safe open in (or under) the same transaction or savepoint
    /// beforehand.
    pub fn unchecked_open(connection: C) -> Self {
        Set::unchecked_open_with_config(connection, Config::default())
    }
}
impl<'db, 'tbl, S, C> Set<'db, 'tbl, S, C>
where
    S: Format,
    C: Savepointable,
{
    pub fn open_with_config(
        mut connection: C,
        config: Config<'db, 'tbl>,
    ) -> Result<Self, OpenError> {
        let database = config.database;
        let table = config.table;

        {
            let sp = connection.savepoint()?;

            let mut version = db::setup(&sp, &database, &table, "ds::set")?;
            if version < 0 {
                return Err(OpenError::TableVersion(version));
            }
            let prev_version = version;
            if version < 1 {
                let trailer = db::strict_without_rowid();
                let sql_type = S::sql_type();

                sp.execute(
                    &format!(
                        "CREATE TABLE {database}.{table} (
                            key {sql_type} UNIQUE PRIMARY KEY NOT NULL
                        ){trailer}"
                    ),
                    [],
                )?;
                version = 1;
            }
            if version > 1 {
                return Err(OpenError::TableVersion(version));
            }
            if prev_version != version {
                db::set_version(&sp, &database, &table, version)?;
            }

            sp.commit()?;
        }
        Ok(Self {
            connection,
            database,
            table,
            serializer: PhantomData,
        })
    }

    /// Open a set without creating it or checking if it exists.  This is safe
    /// if you call a safe open in (or under) the same transaction or savepoint
    /// beforehand.
    pub fn unchecked_open_with_config(connection: C, config: Config<'db, 'tbl>) -> Self {
        let database = config.database;
        let table = config.table;

        Self {
            connection,
            database,
            table,
            serializer: PhantomData,
        }
    }

    pub fn insert(&mut self, value: &S::In) -> Result<bool, Error<S>> {
        let database = &self.database;
        let table = &self.table;
        let serialized = S::serialize(value).map_err(Error::Serialize)?;

        let sp = self.connection.savepoint()?;
        let ret = if db::has_upsert() {
            sp.prepare_cached(&format!(
                "INSERT INTO {database}.{table} (key) VALUES (?) ON CONFLICT DO NOTHING"
            ))?
            .execute(params![serialized])?;
            sp.changes() > 0
        } else if Self::contains_serialized(database, table, &sp, &serialized)? {
            false
        } else {
            sp.prepare_cached(&format!("INSERT INTO {database}.{table} (key) VALUES (?)"))?
                .execute(params![serialized])?;
            true
        };
        sp.commit()?;
        Ok(ret)
    }

    pub fn contains(&mut self, value: &S::In) -> Result<bool, Error<S>> {
        let serialized = S::serialize(value).map_err(|e| Error::Serialize(e))?;
        Self::contains_serialized(
            &self.database,
            &self.table,
            &*self.connection.savepoint()?,
            &serialized,
        )
    }

    fn contains_serialized(
        database: &Identifier,
        table: &Identifier,
        connection: &Connection,
        value: &S::Buffer,
    ) -> Result<bool, Error<S>> {
        Ok(connection
            .prepare_cached(&format!("SELECT 1 FROM {database}.{table} WHERE key = ?"))?
            .query_row(params![value], |_| Ok(()))
            .optional()?
            .is_some())
    }

    pub fn remove(&mut self, value: &S::In) -> Result<bool, Error<S>> {
        let database = &self.database;
        let table = &self.table;
        let serialized = S::serialize(value).map_err(Error::Serialize)?;

        let sp = self.connection.savepoint()?;
        let changes = sp
            .prepare_cached(&format!("DELETE FROM {database}.{table} WHERE key = ?"))?
            .execute(params![serialized])?;

        sp.commit()?;

        Ok(changes > 0)
    }

    pub fn clear(&mut self) -> Result<(), Error<S>> {
        let database = &self.database;
        let table = &self.table;
        let sp = self.connection.savepoint()?;
        sp.prepare_cached(&format!("DELETE FROM {database}.{table}"))?
            .execute([])?;
        sp.commit()?;
        Ok(())
    }

    pub fn first(&mut self) -> Result<Option<S::Out>, Error<S>> {
        let database = &self.database;
        let table = &self.table;

        let serialized: Option<S::Buffer> = self
            .connection
            .savepoint()?
            .prepare_cached(&format!(
                "SELECT key FROM {database}.{table} ORDER BY key ASC"
            ))?
            .query_row([], |row| row.get(0))
            .optional()?;

        serialized
            .map(|s| S::deserialize(&s))
            .transpose()
            .map_err(Error::Deserialize)
    }

    pub fn last(&mut self) -> Result<Option<S::Out>, Error<S>> {
        let database = &self.database;
        let table = &self.table;
        let serialized: Option<S::Buffer> = self
            .connection
            .savepoint()?
            .prepare_cached(&format!(
                "SELECT key FROM {database}.{table} ORDER BY key DESC"
            ))?
            .query_row([], |row| row.get(0))
            .optional()?;

        serialized
            .map(|s| S::deserialize(&s))
            .transpose()
            .map_err(Error::Deserialize)
    }

    pub fn len(&mut self) -> Result<u64, Error<S>> {
        let database = &self.database;
        let table = &self.table;
        Ok(self
            .connection
            .savepoint()?
            .prepare_cached(&format!("SELECT COUNT(*) FROM {database}.{table}"))?
            .query_row([], |row| row.get(0))?)
    }

    pub fn is_empty(&mut self) -> Result<bool, Error<S>> {
        let database = &self.database;
        let table = &self.table;
        Ok(self
            .connection
            .savepoint()?
            .prepare_cached(&format!("SELECT 1 FROM {database}.{table} LIMIT 1"))?
            .query_row([], |_| Ok(()))
            .optional()?
            .is_none())
    }

    pub fn iter(&mut self) -> Result<Iter<'db, 'tbl, S, Savepoint<'_>>, Error<S>> {
        Ok(Iter::new(
            self.connection.savepoint()?,
            self.database.clone(),
            self.table.clone(),
        )?)
    }

    /// Retains only the elements specified by the predicate.
    ///
    /// In other words, remove all elements e for which f(e) returns false. The
    /// elements are visited in ascending serialized order.
    ///
    /// This is all done in a single transaction.
    pub fn retain<F>(&mut self, mut f: F) -> Result<(), Error<S>>
    where
        F: FnMut(S::Out) -> bool,
    {
        let database = &self.database;
        let table = &self.table;

        let sp = self.connection.savepoint()?;
        {
            let mut maybe_serialized = sp
                .prepare_cached(&format!(
                    "SELECT key FROM {database}.{table} ORDER BY key ASC LIMIT 1"
                ))?
                .query_row([], |row| row.get(0))
                .optional()?;
            let mut deleter =
                sp.prepare_cached(&format!("DELETE FROM {database}.{table} WHERE key = ?"))?;
            let mut select_next = sp.prepare_cached(&format!(
                "SELECT key FROM {database}.{table} WHERE key > ? ORDER BY key ASC LIMIT 1"
            ))?;
            while let Some(serialized) = maybe_serialized {
                let item = S::deserialize(&serialized).map_err(|e| Error::Deserialize(e))?;
                if !f(item) {
                    deleter.execute(params![serialized])?;
                }
                maybe_serialized = select_next
                    .query_row(params![serialized], |row| row.get(0))
                    .optional()?;
            }
        }
        sp.commit()?;
        Ok(())
    }
}

#[cfg(test)]
mod test;