//! 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,

    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::{

    OptionalExtension, ToSql, Transaction, TransactionBehavior, named_params, params,

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

};

use saturating_time::SaturatingTime;

use tor_basic_utils::RngExt;

use tor_dircommon::config::DirTolerance;

use tor_error::into_internal;

use tor_netdoc::doc::{

    authcert::{AuthCert, AuthCertKeyIds},

    netstatus::ConsensusFlavor,

};

use crate::{err::DatabaseError, types::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([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 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, 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 unsigned_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> {

    /// Obtains the (valid) consensuses from the database.

    ///

    /// This function queries the database using a [`Transaction`] in order to

    /// have a consistent view upon it.  It will return an [`Option`] containing

    /// a consensus.  In order to obtain a *valid* consensus, a [`Timestamp`]

    /// plus a [`DirTolerance`] are supplied, which will be used for querying

    /// the database in a time-constrained fashion.

    ///

    /// Supplying [`None`] as the [`Timestamp`] simply returns the consensus

    /// with the highest valid-after value, regardless of the current system

    /// time.

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

        //

        // The `valid_after` and `valid_until` cells must be a member of the range:

        // `[valid_after - pre_valid_tolerance; valid_after + post_valid_tolerance]`

        // (inclusively).

        // 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 server descriptors for this consensus.

        // Select the missing router descriptors.

        //

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

        // `consensus_router_descriptor_member` but not in `router_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 router_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.

    /// 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 server descriptors we have.

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

        // server descriptor.

        //

        // It works by doing an inner join from

        // `consensus_router_descriptor_member` to `router_descriptor` because

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

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

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

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

        // the respective extra-info.

        //

        // Parameters:

        // :docid - The docid of the consensus.

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

        // Select the missing micro descriptors.

        //

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

        // `consensus_router_descriptor_member` but not in `router_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 router_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.

}

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

    /// Obtain the most recently published and valid certificate for each authority.

    ///

    /// Returns the found [`AuthCertMeta`] items as well as the missing

    /// [`AuthCertKeyIds`].

    ///

    /// # Performance

    ///

    /// This function has a performance between `O(n * log n)` and `O(n^2)`

    /// because it performs `signatories.len()` database queries, with each

    /// database query potentially taking something between `O(log n)` to

    /// `O(n)` to execute.  However, given that this respective value is

    /// oftentimes fairly small, it should not be much of a big concern.

        // For every key pair in `signatories`, get the most recent valid cert.

        //

        // This query selects the most recent timestamp valid certificate from

        // the database for a single given key pair.  It means that this query

        // has to be executed as many times as there are entries in

        // `signatories`.

        //

        // Unfortunately, there is no neater way to do this, because the

        // alternative would involve using a nested set which SQLite does not

        // support, even with the carray extension.  An alternative might be to

        // precompute that string and then insert it here using `format!` but

        // that feels hacky, error- and injection-prone.

        //

        // Parameters:

        // :id_rsa: The RSA identity key fingerprint in uppercase hexadecimal.

        // :sk_rsa: The RSA signing key fingerprint in uppercase hexadecimal.

        // :now: The current system timestamp.

        // :pre_tolerance: The tolerance for not-yet-valid certificates.

        // :post_tolerance: The tolerance for expired certificates.

        // Keep track of the found (and parsed) certificates and the missing ones.

        // Iterate over every key pair and query it, adding it to found if it exists

        // and was parsed successfully or to missing if it does not exist within the

        // database.

            // Query the certificate from the database.

                },

                |row| 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,

        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_dircommon::config::DirToleranceBuilder;

    use tor_llcrypto::pk::rsa::RsaIdentity;

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

    use crate::testdata2;

    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!(

            "

            SELECT algorithm

            FROM compressed_document

            WHERE identity_docid = 'C3AB8FF13720E8AD9047DD39466B3C8974E592C2FA383D4A3960714CAEF0C4F2'

            "

        )).unwrap();

        let algorithms = stmt

            .query_map(params![], |row| row.get::<_, String>(0))

            .unwrap();

        let algorithms = algorithms.map(|x| x.unwrap()).collect::<HashSet<_>>();

        assert_eq!(

            algorithms,

            HashSet::from([

                "deflate".to_string(),

                "gzip".to_string(),

                "x-zstd".to_string(),

                "x-tor-lzma".to_string()

            ])

        );

        // Now insert the same thing a second time again and see whether the

        // ON CONFLICT magic works.

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

        assert_eq!(docid, docid_second);

        // Remove a few compressed entries and get them again.

        let n = tx

            .execute(

                sql!(

                    "

                    DELETE FROM

                    compressed_document

                    WHERE algorithm IN ('deflate', 'x-zstd')

                    "

                ),

                params![],

            )

            .unwrap();

        assert_eq!(n, 2);

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

        assert_eq!(docid, docid_third);

        let algorithms = stmt

            .query_map(params![], |row| row.get::<_, String>(0))

            .unwrap();

        let algorithms = algorithms.map(|x| x.unwrap()).collect::<HashSet<_>>();

        assert_eq!(

            algorithms,

            HashSet::from([

                "deflate".to_string(),

                "gzip".to_string(),

                "x-zstd".to_string(),

                "x-tor-lzma".to_string()

            ])

        );

    }

    #[test]

    fn compress_test() {

        /// Asserts that `res` contains `encoding`.

        fn contains(encoding: ContentEncoding, res: &[(ContentEncoding, Vec<u8>)]) {

            assert!(res.iter().any(|x| x.0 == encoding));

        }

        const INPUT: &[u8] = "foobar".as_bytes();

        // Check whether everything was encoded.

        let res = ContentEncoding::iter()

            .map(|encoding| (encoding, compress(INPUT, encoding).unwrap()))

            .collect::<Vec<_>>();

        assert_eq!(res.len(), 5);

        contains(ContentEncoding::Identity, &res);

        contains(ContentEncoding::Deflate, &res);

        contains(ContentEncoding::Gzip, &res);

        contains(ContentEncoding::XTorLzma, &res);

        contains(ContentEncoding::XZstd, &res);

        // Check if we can decode it.

        for (encoding, compressed) in res {

            let mut decompressed = Vec::new();

            match encoding {

                ContentEncoding::Identity => decompressed = compressed,

                ContentEncoding::Deflate => {

                    DeflateDecoder::new(Cursor::new(compressed))

                        .read_to_end(&mut decompressed)

                        .unwrap();

                }

                ContentEncoding::Gzip => {

                    GzDecoder::new(Cursor::new(compressed))

                        .read_to_end(&mut decompressed)

                        .unwrap();

                }

                ContentEncoding::XTorLzma => {

                    lzma_rs::lzma_decompress(&mut Cursor::new(compressed), &mut decompressed)

                        .unwrap();

                }

                ContentEncoding::XZstd => {

                    decompressed = zstd::decode_all(Cursor::new(compressed)).unwrap();

                }

            }

            assert_eq!(decompressed, INPUT);

        }

    }

    /// Tests whether consensuses are queried properly from the database given

    /// a pre-defined data.

    ///

    /// It also tests various constraints and edge-cases, including the use of

    /// tolerances.

    #[test]

    fn recent_consensus() {

        let pool = testdata2::test_db();

        let no_tolerance = DirToleranceBuilder::default()

            .pre_valid_tolerance(Duration::ZERO)

            .post_valid_tolerance(Duration::ZERO)

            .build()

            .unwrap();

        let liberal_tolerance = DirToleranceBuilder::default()

            .pre_valid_tolerance(Duration::from_secs(60 * 60)) // 1h before

            .post_valid_tolerance(Duration::from_secs(60 * 60)) // 1h after

            .build()

            .unwrap();

        let docid = Sha256::digest(testdata2::current_consensus_ns().1.as_bytes());

        let lifetime = testdata2::current_consensus_ns().0.preamble.lifetime;