//! Access to the database schema.

//!

//! This module is not intended to provide a high-level ORM, instead it serves

//! the purpose of initializing and upgrading the database, if necessary.

//!

//! # Synchronous or Asynchronous?

//!

//! The question on whether the database and access to it shall be synchronous

//! or asynchronous has been fairly long debate that eventually got settled

//! after realizing that an asynchronous approach does not work.  This comment

//! should serve as a reminder for future devs, wondering why we use certain

//! synchronous primitives in an otherwise asynchronous codebase.

//!

//! Early on, it was clear that we would need some sort of connection pool,

//! primarily for two reasons:

//! 1. Performing frequent open and close calls in every task would be costly.

//! 2. Sharing a single connection object with a Mutex would be a waste

//!

//! Because the application itself is primarily asynchronous, we decided to go

//! with an asynchronous connection pool as well, leading to the choose of

//! `deadpool` initially.

//!

//! However, soon thereafter, problems with `deadpool` became evident.  Those

//! problems mostly stemmed from the synchronous nature of SQLite itself.  In our

//! case, this problem was initially triggered by figuring out a way to solve

//! `SQLITE_BUSY` handling.  In the end, we decided to settle upon the following

//! approach: Set `PRAGMA busy_timeout` to a certain value and create write

//! transactions with `BEGIN EXCLUSIVE`.  This way, SQLite would try to obtain

//! a write transaction for `busy_timeout` milliseconds by blocking the current

//! thread.  Due to this blocking, async no longer made any sense and was in

//! fact quite counter-productive because those potential sleep could screw a

//! lot of things up, which became very evident while trying to test this.

//!

//! Besides, throughout refactoring the code base, we realized that, even while

//! still using `deadpool`, the actual "asynchronous" calls interfacing with the

//! database became smaller and smaller.  In the end, the asynchronous code just

//! involved parts of obtaining a connection and creating a transaction,

//! eventually resulting in a calling a synchronous function taking the

//! transaction handle to perform the lion's share of the operation.

// TODO DIRMIRROR: This could benefit from methods by wrapping the pool into a

// custom type.

use std::{

    collections::HashSet,

    fmt::Display,

    io::{Cursor, Write},

    marker::PhantomData,

    num::NonZero,

    ops::{Add, Sub},

    path::Path,

    rc::Rc,

    time::{Duration, SystemTime},

};

use digest::Digest;

use educe::Educe;

use flate2::write::{DeflateEncoder, GzEncoder};

use r2d2::Pool;

use r2d2_sqlite::SqliteConnectionManager;

use rand::Rng;

use rusqlite::{

    ToSql, Transaction, TransactionBehavior, named_params, params,

    types::{FromSql, FromSqlError, FromSqlResult, ToSqlOutput, Value, ValueRef},

};

use saturating_time::SaturatingTime;

use tor_basic_utils::RngExt;

use tor_error::{internal, into_internal};

use tor_netdoc::doc::{authcert::AuthCert, netstatus::ConsensusFlavor};

use crate::{

    err::DatabaseError,

    types::{FlavoredConsensusBody, FlavoredConsensusSignatures, FlavoredConsensusUnverified},

};

/// Version 1 of the database schema.

///

/// TODO DIRMIRROR: Before the release, figure out where to use rowid and where

/// to use docid.

const V1_SCHEMA: &str = include_str!("schema_v1.sql");

/// Global options set in every connection.

const GLOBAL_OPTIONS: &str = sql!(

    "

PRAGMA journal_mode=WAL;

PRAGMA foreign_keys=ON;

PRAGMA busy_timeout=1000;

"

);

/// Convenience macro for implementing a hash type in a rusqlite compatible fashion.

///

/// This macro accepts the following parameters:

/// 1. `name` for specifying an identifier of the type, such as [`Sha256`].

/// 2. `algo` for specifying the type from the rust-crypto [`digest`] ecosystem,

///    such as [`tor_llcrypto::d::Sha256`].

/// 3. The size in bytes of the hash output, such as `32` for [`Sha256`].

///     * Unfortunately, we cannot use something like [`Digest::output_size()`]

///       because it is not a constant.

///

/// It generates a struct with `name` as the identifier, which implements the

/// following methods:

/// * `digest` for wrapping around [`Digest::digest()`].

///

/// It also implements the following traits:

/// * [`Display`]

/// * [`FromSql`]

/// * [`ToSql`]

/// * [`PartialEq<&str>`] for base16 comparisons

/// * [`From<u8; $size>`]

macro_rules! impl_hash_wrapper {

    ($name:ident, $algo:ty, $size:literal) => {

        /// Database wrapper type for $name.

        ///

        /// Serves as a database friendly wrapper around [`tor_llcrypto::d`]

        /// with features such as SQL support.

        #[derive(Debug, Copy, Clone, PartialEq, Eq, Hash)]

        pub(crate) struct $name(pub [u8; $size]);

        impl $name {

            /// Computes the hash from arbitrary data.

        }

        impl Display for $name {

            /// Formats the hash in uppercase hexadecimal.

        }

        impl FromSql for $name {

                // We read the hash as a hexadecimal string from the database.

                // Convert it to binary data and check length afterwards.

        }

        impl ToSql for $name {

                // Because Self is only constructed with FromSql and digest

                // data, it is safe to assume it is valid.

        }

        impl From<$name> for Value {

        }

        impl PartialEq<&str> for $name {

        }

        impl From<[u8; $size]> for $name {

        }

    };

}

impl_hash_wrapper!(Sha1, tor_llcrypto::d::Sha1, 20);

impl_hash_wrapper!(Sha256, tor_llcrypto::d::Sha256, 32);

impl_hash_wrapper!(Sha3_256, tor_llcrypto::d::Sha3_256, 32);

/// The identifier for documents in the content-addressable cache.

///

/// Right now, this is a [`Sha256`] hash, but this may change in future.

pub(crate) type DocumentId = Sha256;

/// The supported content encodings.

#[derive(

    Debug, Clone, Copy, PartialEq, Eq, Hash, strum::EnumString, strum::Display, strum::EnumIter,

)]

#[strum(serialize_all = "kebab-case", ascii_case_insensitive)]

pub(crate) enum ContentEncoding {

    /// RFC2616 section 3.5.

    Identity,

    /// RFC2616 section 3.5.

    Deflate,

    /// RFC2616 section 3.5.

    Gzip,

    /// The zstandard compression algorithm (www.zstd.net).

    XZstd,

    /// The lzma compression algorithm with a "present" value no higher than 6.

    XTorLzma,

}

/// A wrapper around [`SystemTime`] with convenient features.

///

/// Please use this type throughout the crate internally, instead of

/// [`SystemTime`].

///

/// # Conversion

///

/// This type can be safely converted from and into a [`SystemTime`], because

/// it is just a wrapper type.

///

/// # Saturating Arithmetic

///

/// This type implements [`Add`] and [`Sub`] for [`Duration`] and [`Timestamp`]

/// ([`Sub`] only) using saturating arithmetic from the [`saturating_time`]

/// crate.  It means that addition and subtraction can be safely performed

/// without the potential risk of an unexpected panic, instead wrapping to

/// a local maximum/minimum or [`Duration::ZERO`] depending on the type.

///

/// Note that we don't provide a saturating version of [`Duration`], so addition

/// or subtraction of two [`Duration`]s still needs care to avoid panics.

///

/// # SQLite Interaction

///

/// This type implements [`FromSql`] and [`ToSql`], making it convenient to

/// integrate into SQL statements, as the database schema represents timestamps

/// internally using a non-negative [`i64`] storing the seconds since the epoch.

#[derive(Copy, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Debug)]

pub(crate) struct Timestamp(SystemTime);

impl From<SystemTime> for Timestamp {

}

impl From<Timestamp> for SystemTime {

}

impl Add<Duration> for Timestamp {

    type Output = Self;

    /// Performs a saturating addition wrapping to [`SystemTime::max_value()`].

}

impl Sub<Duration> for Timestamp {

    type Output = Self;

    /// Performs a saturating subtraction wrapping to [`SystemTime::min_value()`].

}

impl Sub<Timestamp> for Timestamp {

    type Output = Duration;

    /// Performs a saturating duration_since wrapping to [`Duration::ZERO`].

        #[allow(unstable_name_collisions)]

}

impl FromSql for Timestamp {

}

impl ToSql for Timestamp {

        #[allow(unstable_name_collisions)]

}

/// Representation of consensus metadata from the database.

#[derive(Educe)]

#[educe(Debug, Clone, Copy, PartialEq, Eq)]

pub(crate) struct ConsensusMeta<T> {

    /// The document id uniquely identifying the consensus.

    pub docid: DocumentId,

    /// The SHA3 of the unsigned part of the consensus.

    pub sha3_256: Sha3_256,

    /// The time after which this consensus is valid.

    pub valid_after: Timestamp,

    /// The time after which this consensus stops being fresh.

    pub fresh_until: Timestamp,

    /// The time after which this consensus stops being valid.

    pub valid_until: Timestamp,

    /// The flavor of the consensus; determined at compile time.

    flavor: PhantomData<T>,

}

impl<T: FlavoredConsensusUnverified> ConsensusMeta<T> {

    /// Select the missing router descriptors.

    ///

    /// A router descriptor is considered missing if it exists in

    /// `consensus_descriptor_member` but not in `descriptor`

    /// because the first entry is added once the consensus got parsed,

    /// whereas the second entry is added once we have actually retrieved it.

    ///

    /// It works by doing a left join on descriptor and filtering for

    /// all entries where the join is NULL, as that implies we are aware of

    /// the descriptor but not have it stored.

    ///

    /// Parameters:

    /// :docid - The docid of the consensus.

    /// :exclude - The descriptors to skip.

    /// :limit - The maximum number of descriptors to return.

    //

    // TODO DIRMIRROR: Potentially constify more queries.

    const MISSING_ROUTERS_QUERY: &'static str = sql!(

        "

        SELECT cr.sha1, RANDOM() AS rand

        FROM consensus_descriptor_member AS cr

          LEFT JOIN descriptor AS router ON cr.sha1 = router.sha1

        WHERE

          cr.consensus_docid = :docid

          AND cr.sha1 IS NOT NULL

          AND cr.sha1 NOT IN rarray(:exclude)

          AND router.sha1 IS NULL

        ORDER BY rand

        LIMIT :limit

        "

    );

    /// Obtains all consensuses found in the database.

    ///

    /// This should be reasonable size-wise, given that a ConsensusMeta instance

    /// is very small and that the garbage collector removes old consensuses

    /// anyways.  If this becomes a problem, we may want to add an optional

    /// limit.

        // Select the most recent flavored consensus document from the database.

        // Actually execute the query.

            },

                Ok(Self {

                })

    /// Queries the raw data of a [`ConsensusMeta`].

    /// Calculates the [`Timestamp`] at which the authorities will be queried again.

    ///

    /// # Specifications

    ///

    /// * <https://spec.torproject.org/dir-spec/directory-cache-operation.html#download-ns-from-auth>

    /// Returns the missing router descriptors for this consensus.

    ///

    /// `limit` may be given to specify an optional upper limit, in which case

    /// the result will contain at most `limit` missing descriptors.

    ///

    /// # Performance

    ///

    /// The performance here is O(n * log n) for each 64-element batch, leading

    /// to an overall performance of O(n^2 * (log n)/64) per consensus period,

    /// which sould be acceptable.  See the link below for a further discussion

    /// on performance related matters:

    ///

    /// <https://gitlab.torproject.org/tpo/core/arti/-/merge_requests/4378#note_3467300>

    // TODO DIRMIRROR: This has a large overlap with missing_extras and

    // missing_micros, we should genericise or macrofy this at one point.

    /// Returns the missing extra infos for this consensus to the best of our abilities.

    ///

    /// Keep in mind that this does not return **all** missing extra infos but

    /// only the missing extra infos of router descriptors we have.

    ///

    /// `limit` may be given to specify an optional upper limit, in which case

    /// the result will contain at most `limit` missing extra-infos.

    ///

    /// See [`ConsensusMeta::missing_routers()`] for a discussion on

    /// performance.

        // Select the missing extra infos for this consensus.

        //

        // This return value is not complete because we only know the missing

        // extra-infos to the best of our abilities.  In other words: We are

        // only aware of a missing extra-info if we have parsed the respective

        // router descriptor.

        //

        // It works by doing an inner join from

        // `consensus_descriptor_member` to `descriptor` because

        // we can only know about the extra-infos of which we have the router

        // descriptors from.  Afterwards, we do a left join with the

        // `extra_info` table and filter for all results where the left

        // join result is null, hence where we have a router descriptor but not

        // the respective extra-info.

        //

        // Parameters:

        // :docid - The docid of the consensus.

        // :exclude - The descriptors to skip.

        // :limit - The maximum number of descriptors to return.

    /// Returns the missing micro descriptors for this consensus.

    ///

    /// `limit` may be given to specify an optional upper limit, in which case

    /// the result will contain at most `limit` missing descriptors.

    ///

    /// See [`ConsensusMeta::missing_routers()`] for a discussion on

    /// performance.

        // Select the missing micro descriptors.

        //

        // A micro descriptor is considered missing if it exists in

        // `consensus_descriptor_member` but not in `descriptor`

        // because the first entry is added once the consensus got parsed,

        // whereas the second entry is added once we have actually retrieved it.

        //

        // It works by doing a left join on descriptor and filtering for

        // all entries where the join is NULL, as that implies we are aware of

        // the descriptor but not have it stored.

        //

        // Parameters:

        // :docid - The docid of the consensus.

        // :limit - The maximum number of descriptors to return.

    /// Inserts a verified consensus into the database.

    ///

    /// This method *DOES NOT* compute any consensus diffs but populates

    /// associated meta tables such as `consensus_descriptor_member`

    /// and `consensus_authority_voter` properly.

    {

        // Insert a consensus into the consensus meta table.

        //

        // This requires the consensus to already be present in the `store`

        // table.

        //

        // Parameters:

        // :docid - The docid of the consensus.

        // :sha3_256 - The SHA3-256 sum of the part of the consensus

        //   without signatures but a trailing `\ndirectory-signature<SPACE>`.

        // :flavor - The consensus flavor to use.

        // :valid_after, :fresh_until, :valid_until - The respective validities.

        // Insert the relationship between a router status to a consensus.

        //

        // Parameters:

        // :docid - The consensus docid.

        // :sha1 - The router descriptor digest (plain consensus only, NULL otherwise).

        // :sha2 - The micro descriptor digest (microdesc consensus only, NULL otherwise).

        // Insert the relationship between authority certificate to consensus votes.

        //

        // Parameters:

        // :cons - The consensus docid.

        // :auth - The authority certificate docid.

        // First, insert the consensus as is.

        // TODO DIRMIRROR: We *need* this in tor-netdoc!!!

        );

        // Insert all router descriptor member relationships.

        //

        // Many small queries are fast in sqlite, so this shouldn't be a perf problem.

        // https://sqlite.org/np1queryprob.html

        //

        // Depending on the flavor, we map the document digests to an iterator

        // of a tuple of Option's, where one of the values is always None.

        // This represents the fact that router descriptors use SHA-1 and micro

        // descriptors SHA-256.

        }

        }

}

/// Representation of authority certificate metadata from the database.

#[derive(Debug, Clone, Copy, PartialEq, Eq)]

pub(crate) struct AuthCertMeta {

    /// The document id uniquely identifying the consensus.

    pub docid: DocumentId,

    /// The SHA-1 fingerprint of the identity key.

    // TODO DIRMIRROR: Change this to RsaIdentity.

    pub kp_auth_id_rsa_sha1: Sha1,

    /// The SHA-1 fingerprint of the signign key.

    // TODO DIRMIRROR: Change this to RsaIdentity.

    pub kp_auth_sign_rsa_sha1: Sha1,

    /// The timestamp after which this certificate will be valid.

    pub dir_key_published: Timestamp,

    /// The timestamp until this certificate will be valid.

    pub dir_key_expires: Timestamp,

}

impl AuthCertMeta {

    /// Obtains the authority certificates from the database.

        // Obtain all certificates from the database.

        //

        // This is okay because the set is not very big.

        //

        // In the unlikely edge case of on identity-signing key pair having

        // multiple certificates, the most recently published certificate is

        // going to be used.

        //

        // TODO DIRMIRROR: Perhaps we should modify the auth_certs table to

        // add a UNIQUE constraint on that combination, while modifying the

        // insertion logic to replace with the newer one in the case of a

        // conflict.

                Ok(Self {

                })

    /// Queries the raw data of an [`AuthCertMeta`].

    /// Inserts a new authority certificate into the database.

    ///

    /// Keep in mind that the data in the [`AuthCert`] should correspond to the

    /// data found in `data`, as this method performs no parsing.

        // Inserts a new certificate into the meta table.

        //

        // Parameters:

        // :docid - The document id.

        // :id_rsa - The identity key fingerprint.

        // :sign_rsa - The signing key fingerprint

        // :published - The published timestamp.

        // :expires - The expires timestamp.

}

/// A no-op macro just returning the supplied.

///

/// The purpose of this macro is to semantically mark [`str`] literals to be

/// SQL statement.

///

/// Keep in mind that the compiler will not notice if you forget this macro.

/// Unfortunately, you have to ensure it yourself.

macro_rules! sql {

    ($s:literal) => {

        $s

    };

}

pub(crate) use sql;

/// Opens a database from disk, creating a [`Pool`] for it.

///

/// This function should be the entry point for all things requiring a database

/// handle, as this function prepares all necessary steps required for operating

/// on the database correctly, such as:

/// * Schema initialization.

/// * Schema upgrade.

/// * Setting connection specific settings.

///

/// # `SQLITE_BUSY` Caveat

///

/// There is a problem with the handling of `SQLITE_BUSY` when opening an

/// SQLite database.  In WAL, opening a database might acquire an exclusive lock

/// for a very short amount of time, in order to perform clean-up from previous

/// connections alongside other tasks for maintaining database integrity?  This

/// means, that opening multiple SQLite databases simultaneously will result in

/// a busy error regardless of a busy handler, as setting a busy handler will

/// require an existing connection, something we are unable to obtain in the

/// first place.

///

/// In order to mitigate this issue, the recommended way in the SQLite community

/// is to simply ensure that database connections are opened sequentially,

/// by urging calling applications to just use a single [`Pool`] instance.

///

/// Testing this is hard unfortunately.

        // Prepare the database, doing the following steps:

        // 1. Checking the database schema.

        // 2. Upgrading (in future) or initializing the database schema (if empty).

            sql!(

            ),

        ) {

        };

                }

            }

        } else {

        }

/// Executes a closure `op` with a given read-only [`Transaction`].

///

/// The [`Transaction`] always gets rolled back the moment `op` returns.

///

/// The [`Transaction`] gets initialized with the global pragma options set.

///

/// **The closure shall not perform write operations!**

/// Not only do they get rolled back anyways, but upgrading the [`Transaction`]

/// from a read to a write transaction will lead to other simultaneous write upgrades

/// to fail.  Unfortunately, there is no real programmatic way to ensure this.

{

/// Executes a closure `op` with a given read-write [`Transaction`].

///

/// The [`Transaction`] always gets committed the moment `op` returns.

///

/// The [`Transaction`] gets initialized with the global pragma options set.

///

/// The [`Transaction`] gets created with [`TransactionBehavior::Immediate`],

/// meaning it will immediately exist as a write connection, retrying in the

/// case of a [`rusqlite::ErrorCode::DatabaseBusy`] until it failed after 1s.

{

/// Inserts `data` into store while also compressing it with given encodings.

///

/// Returns the [`DocumentId`] of `data`.

///

/// This function inserts `data` into store and also compresses it into all

/// given compression formats.

///

/// Duplicates get re-encoded and replaced in the database, including

/// [`ContentEncoding::Identity`].

    // The statement to insert some data into the store.

    //

    // Parameters:

    // :docid - The docid.

    // :content - The binary data.

    // The statement to insert a compressed document into the metatable.

    //

    // Parameters:

    // :algorithm - The name of the encoding algorithm.

    // :identity_docid - The docid of the plain-text document in the store.

    // :compressed_docid - The docid of the encoded document in the store.

    // Insert the plain document into the store.

    // Compress it into all formats and insert it into store and compressed.

            // Ignore identity because we inserted that above.

        // We map a compression error to a bug because there is no good reason

        // on why it should fail, given that we compress from memory data to

        // memory data.  Probably because it uses the std::io::Writer interface

        // which itself demands use of std::io::Result.

    }

/// Compresses `data` into a specified [`ContentEncoding`].

///

/// Returns a [`Vec`] containing the encoded data.

        ContentEncoding::Deflate => {

        }

        ContentEncoding::Gzip => {

        }

        ContentEncoding::XTorLzma => {

        }

    }

#[cfg(test)]

mod test {

    // @@ begin test lint list maintained by maint/add_warning @@

    #![allow(clippy::bool_assert_comparison)]

    #![allow(clippy::clone_on_copy)]

    #![allow(clippy::dbg_macro)]

    #![allow(clippy::mixed_attributes_style)]

    #![allow(clippy::print_stderr)]

    #![allow(clippy::print_stdout)]

    #![allow(clippy::single_char_pattern)]

    #![allow(clippy::unwrap_used)]

    #![allow(clippy::unchecked_time_subtraction)]

    #![allow(clippy::useless_vec)]

    #![allow(clippy::needless_pass_by_value)]

    #![allow(clippy::string_slice)] // See arti#2571

    //! <!-- @@ end test lint list maintained by maint/add_warning @@ -->

    use std::{

        collections::HashSet,

        io::Read,

        iter,

        sync::{Arc, Once},

    };

    use flate2::read::{DeflateDecoder, GzDecoder};

    use rusqlite::Connection;

    use strum::IntoEnumIterator;

    use tempfile::tempdir;

    use tor_basic_utils::test_rng::testing_rng;

    use tor_netdoc::doc::netstatus::{md, plain};

    use crate::testdata2::{self, current_consensus_ns};

    use super::*;

    type Plain = plain::NetworkStatusUnverified;

    type Md = md::NetworkStatusUnverified;

    #[test]

    fn open_test() {

        let db_dir = tempdir().unwrap();

        let db_path = db_dir.path().join("db");

        open(&db_path).unwrap();

        let conn = Connection::open(&db_path).unwrap();

        // Check if the version was initialized properly.

        let version = conn

            .query_one(

                "SELECT version FROM arti_dirserver_schema_version WHERE rowid = 1",

                params![],

                |row| row.get::<_, String>(0),

            )

            .unwrap();

        assert_eq!(version, "1");

        // Set the version to something unknown.

        conn.execute(

            "UPDATE arti_dirserver_schema_version SET version = 42",

            params![],

        )

        .unwrap();

        drop(conn);

        assert_eq!(

            open(&db_path).unwrap_err().to_string(),

            "incompatible schema version: 42"

        );

    }

    #[test]

    fn read_tx_test() {

        let db_dir = tempdir().unwrap();

        let db_path = db_dir.path().join("db");

        let pool = open(&db_path).unwrap();

        // Do a write transaction despite forbidden.

        read_tx(&pool, |tx| {

            tx.execute_batch("DELETE FROM arti_dirserver_schema_version")

                .unwrap();

            let e = tx

                .query_one(

                    sql!("SELECT version FROM arti_dirserver_schema_version"),

                    params![],

                    |row| row.get::<_, String>(0),

                )

                .unwrap_err();

            assert_eq!(e, rusqlite::Error::QueryReturnedNoRows);

        })

        .unwrap();

        // Normal check.

        let version: String = read_tx(&pool, |tx| {

            tx.query_one(

                sql!("SELECT version FROM arti_dirserver_schema_version"),

                params![],

                |row| row.get(0),

            )

            .unwrap()

        })

        .unwrap();

        assert_eq!(version, "1");

    }

    #[test]

    fn rw_tx_test() {

        let db_dir = tempdir().unwrap();

        let db_path = db_dir.path().join("db");

        let pool = open(&db_path).unwrap();

        // Do a write transaction.

        rw_tx(&pool, |tx| {

            tx.execute_batch("DELETE FROM arti_dirserver_schema_version")

                .unwrap();

        })

        .unwrap();

        // Check that it was deleted.

        read_tx(&pool, |tx| {

            let e = tx

                .query_one(

                    sql!("SELECT version FROM arti_dirserver_schema_version"),

                    params![],

                    |row| row.get::<_, String>(0),

                )

                .unwrap_err();

            assert_eq!(e, rusqlite::Error::QueryReturnedNoRows);

        })

        .unwrap();

    }

    /// Tests whether our SQLite busy error handling works in normal situations.

    ///

    /// A normal situations means a situation where a lock is never held for

    /// more than 1000ms.  In our case, we will work with two threads.

    /// t1 will acquire an exclusive lock and inform t2 about it.  t2 waits

    /// until t1 has acquired this lock and then immediately informs t1, that

    /// it will now wait for a lock too.  Now, t1 will immediately terminate,

    /// thereby releasing the lock and leading t2 to eventually acquire it.

    #[test]

    fn rw_tx_busy_timeout_working() {

        let db_dir = tempdir().unwrap();

        let db_path = db_dir.path().join("db");

        let pool = open(db_path).unwrap();

        // t2 will wait on this before it starts doing stuff.

        let t1_acquired_lock = Arc::new(Once::new());

        // t1 will wait on this in order to terminate properly.

        let t2_is_waiting = Arc::new(Once::new());

        let t1 = std::thread::spawn({

            let pool = pool.clone();

            let t1_acquired_lock = t1_acquired_lock.clone();

            let t2_is_waiting = t2_is_waiting.clone();

            move || {

                rw_tx(&pool, move |_tx| {

                    // Inform t2 we have write lock.

                    t1_acquired_lock.call_once(|| ());

                    println!("t1 acquired write lock");

                    // Wait for t2 to start waiting.

                    t2_is_waiting.wait();

                })

                .unwrap();

                println!("t2 released write lock");

            }

        });

        println!("t2 waits for t1 to acquire write lock");

        t1_acquired_lock.wait();

        t2_is_waiting.call_once(|| ());

        rw_tx(&pool, |_| ()).unwrap();

        println!("t2 acquired and released write lock");

        t1.join().unwrap();

    }

    /// Tests whether our SQLite busy error handlings fails as expected.

    ///

    /// We configure SQLite to fail after 1000ms.  This test works with two

    /// threads.  t1 will acquire an exclusive lock on the database and will

    /// inform t2 about it, which itself will wait until t1 has acquired the

    /// lock.  t2 will then immediately try to also obtain an exclusive lock,

    /// which should fail after about 1000ms.  After the failure, t2 informs

    /// t1 that it has failed, causing t1 to terminate.

    #[test]

    fn rw_tx_busy_timeout_busy() {

        let db_dir = tempdir().unwrap();

        let db_path = db_dir.path().join("db");

        let pool = open(db_path).unwrap();

        // t2 will wait on this before it starts doing stuff.

        let t1_acquired_lock = Arc::new(Once::new());

        // t1 will wait on this in order to terminate properly.

        let t2_gave_up = Arc::new(Once::new());

        let t1 = std::thread::spawn({

            let pool = pool.clone();

            let t1_acquired_lock = t1_acquired_lock.clone();

            let t2_gave_up = t2_gave_up.clone();

            move || {

                rw_tx(&pool, move |_tx| {

                    // Inform t2 we have the write lock.

                    t1_acquired_lock.call_once(|| ());

                    println!("t1 acquired write lock");

                    // Wait for t2 to give up before we release (how mean from us).

                    t2_gave_up.wait();

                })

                .unwrap();

                println!("t1 released write lock");

            }

        });

        println!("t2 waits for t1 to acquire write lock");

        t1_acquired_lock.wait();

        let e = rw_tx(&pool, |_| ()).unwrap_err();

        assert_eq!(

            e.to_string(),

            "low-level rusqlite error: database is locked"

        );

        println!("t2 gave up on acquiring write lock");

        t2_gave_up.call_once(|| ());

        t1.join().unwrap();

    }

    #[test]

    fn store_insert_test() {

        let db_dir = tempdir().unwrap();

        let db_path = db_dir.path().join("db");

        open(&db_path).unwrap();

        let mut conn = Connection::open(&db_path).unwrap();

        let tx = conn.transaction().unwrap();

        let docid = store_insert(&tx, "foobar".as_bytes(), ContentEncoding::iter()).unwrap();

        assert_eq!(

            docid,

            "C3AB8FF13720E8AD9047DD39466B3C8974E592C2FA383D4A3960714CAEF0C4F2"

        );

        let res = tx

            .query_one(

                sql!(

                    "

                    SELECT content

                    FROM store

                    WHERE docid = 'C3AB8FF13720E8AD9047DD39466B3C8974E592C2FA383D4A3960714CAEF0C4F2'

                    "

                ),

                params![],

                |row| row.get::<_, Vec<u8>>(0),

            )

            .unwrap();

        assert_eq!(res, "foobar".as_bytes());

        let mut stmt = tx.prepare_cached(sql!(