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

    num::NonZero,

    ops::{Add, Sub},

    path::Path,

    time::{Duration, SystemTime},

};

use digest::Digest;

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

use getset::CopyGetters;

use r2d2::Pool;

use r2d2_sqlite::SqliteConnectionManager;

use rand::Rng;

use rusqlite::{

    named_params, params,

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

    OptionalExtension, ToSql, Transaction, TransactionBehavior,

};

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;

/// 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>`] but only in tests

macro_rules! impl_hash_wrapper {

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

        /// Database wrapper type.

        #[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 {

        }

        #[cfg(test)]

        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(Debug, Clone, Copy, PartialEq, Eq, CopyGetters)]

#[get_copy = "pub(crate)"]

pub(crate) struct ConsensusMeta {

    /// The document id uniquely identifying the consensus.

    docid: DocumentId,

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

    unsigned_sha3_256: Sha3_256,

    /// The flavor of the consensus.

    flavor: ConsensusFlavor,

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

    valid_after: Timestamp,

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

    fresh_until: Timestamp,

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

    valid_until: Timestamp,

}

impl ConsensusMeta {

    /// Obtains the most recent valid consensus 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.

    ///

    /// The [`None`] case implies that no valid consensus has been found, that

    /// is, no consensus at all or no consensus whose `valid-before` or

    /// `valid-after` lies within the range composed by `now` and `tolerance`.

        // 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; a None is totally valid and considered as

        // no consensus being present in the current database.

            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, CopyGetters)]

#[get_copy = "pub(crate)"]

pub(crate) struct AuthCertMeta {

    /// The document id uniquely identifying the consensus.

    docid: DocumentId,

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

    // TODO DIRMIRROR: Change this to RsaIdentity.

    kp_auth_id_rsa_sha1: Sha1,

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

    // TODO DIRMIRROR: Change this to RsaIdentity.

    kp_auth_sign_rsa_sha1: Sha1,

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

    dir_key_published: Timestamp,

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

    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)]

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

    use std::{

        collections::HashSet,

        io::Read,

        sync::{Arc, Once},

    };

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

    use lazy_static::lazy_static;

    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 super::*;

    lazy_static! {

    /// Wed Jan 01 2020 00:00:00 GMT+0000

    static ref VALID_AFTER: Timestamp =

        (SystemTime::UNIX_EPOCH + Duration::from_secs(1577836800)).into();

    /// Wed Jan 01 2020 01:00:00 GMT+0000

    static ref FRESH_UNTIL: Timestamp =

        *VALID_AFTER + Duration::from_secs(60 * 60);

    /// Wed Jan 01 2020 02:00:00 GMT+0000

    static ref FRESH_UNTIL_HALF: Timestamp =

        *FRESH_UNTIL + Duration::from_secs(60 * 60);

    /// Wed Jan 01 2020 03:00:00 GMT+0000

    static ref VALID_UNTIL: Timestamp =

        *FRESH_UNTIL + Duration::from_secs(60 * 60 * 2);

    }

    const CONSENSUS_CONTENT: &str = "Lorem ipsum dolor sit amet.";

    const CONSENSUS_MD_CONTENT: &str = "Lorem ipsum dolor sit amet!";

    const CERT_CONTENT: &[u8] = include_bytes!("../testdata/authcert-longclaw");

    lazy_static! {

        static ref CONSENSUS_DOCID: DocumentId = DocumentId::digest(CONSENSUS_CONTENT.as_bytes());

        static ref CONSENSUS_MD_DOCID: DocumentId =

            DocumentId::digest(CONSENSUS_MD_CONTENT.as_bytes());

        static ref CERT_DOCID: DocumentId = DocumentId::digest(CERT_CONTENT);

    }

    fn create_dummy_db() -> Pool<SqliteConnectionManager> {

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

        rw_tx(&pool, |tx| {

            tx.execute(

                sql!("INSERT INTO store (docid, content) VALUES (?1, ?2)"),

                params![*CONSENSUS_DOCID, CONSENSUS_CONTENT.as_bytes()],

            )

            .unwrap();

            tx.execute(

                sql!("INSERT INTO store (docid, content) VALUES (?1, ?2)"),

                params![*CONSENSUS_MD_DOCID, CONSENSUS_MD_CONTENT.as_bytes()],

            )

            .unwrap();

            tx.execute(

                sql!("INSERT INTO store (docid, content) VALUES (?1, ?2)"),

                params![*CERT_DOCID, CERT_CONTENT],

            )

            .unwrap();

            tx.execute(

                sql!("INSERT INTO store (docid, content) VALUES (?1, ?2)"),

                params![

                    DocumentId::digest(include_bytes!("../testdata/descriptor1-ns")),

                    include_bytes!("../testdata/descriptor1-ns")

                ]

            ).unwrap();

            tx.execute(

                sql!("INSERT INTO store (docid, content) VALUES (?1, ?2)"),

                params![

                    DocumentId::digest(include_bytes!("../testdata/descriptor1-extra-info")),

                    include_bytes!("../testdata/descriptor1-extra-info")

                ]

            ).unwrap();

            tx.execute(

                sql!("INSERT INTO store (docid, content) VALUES (?1, ?2)"),

                params![

                    DocumentId::digest(include_bytes!("../testdata/descriptor1-md")),

                    include_bytes!("../testdata/descriptor1-md"),

            ]).unwrap();

            // Insert descriptor into router_extra_info.

            tx.execute(sql!(

                "

                INSERT INTO router_extra_info

                (docid, unsigned_sha1, kp_relay_id_rsa_sha1)

                VALUES (?1, ?2, ?3)

                "

            ), params![

                Sha256::digest(include_bytes!("../testdata/descriptor1-extra-info")),

                Sha1::digest(include_bytes!("../testdata/descriptor1-extra-info-unsigned")),

                "000004ACBB9D29BCBA17256BB35928DDBFC8ABA9",

            ]).unwrap();

            // We only insert descriptor1 here.

            tx.execute(sql!(

                "

                INSERT INTO router_descriptor

                (docid, unsigned_sha1, unsigned_sha2, kp_relay_id_rsa_sha1, flavor, extra_unsigned_sha1)

                VALUES

                (?1, ?2, ?3, ?4, 'ns', ?5)

                "

            ), params![

                DocumentId::digest(include_bytes!("../testdata/descriptor1-ns")),

                Sha1::digest(include_bytes!("../testdata/descriptor1-ns-unsigned")),

                Sha256::digest(include_bytes!("../testdata/descriptor1-ns-unsigned")),

                Sha1::from([0, 0, 4, 172, 187, 157, 41, 188, 186, 23, 37, 107, 179, 89, 40, 221, 191, 200, 171, 169]),

                Sha1::digest(include_bytes!("../testdata/descriptor1-extra-info-unsigned")),

            ]).unwrap();

            // Only insert descriptor1's md

            tx.execute(sql!(

                "

                INSERT INTO router_descriptor

                (docid, unsigned_sha1, unsigned_sha2, kp_relay_id_rsa_sha1, flavor)

                VALUES (?1, ?2, ?3, ?4, 'microdesc')

                "

            ), params![

                DocumentId::digest(include_bytes!("../testdata/descriptor1-md")),

                Sha1::digest(include_bytes!("../testdata/descriptor1-md")),

                Sha256::digest(include_bytes!("../testdata/descriptor1-md")),

                Sha1::from([0, 0, 4, 172, 187, 157, 41, 188, 186, 23, 37, 107, 179, 89, 40, 221, 191, 200, 171, 169]),

            ]).unwrap();

            tx.execute(

                sql!(

                    "

                    INSERT INTO consensus

                    (docid, unsigned_sha3_256, flavor, valid_after, fresh_until, valid_until)

                    VALUES

                    (?1, ?2, ?3, ?4, ?5, ?6)

                    "

                ),

                params![

                    *CONSENSUS_DOCID,

                    "0000000000000000000000000000000000000000000000000000000000000000", // not the correct hash

                    ConsensusFlavor::Plain.name(),

                    *VALID_AFTER,

                    *FRESH_UNTIL,

                    *VALID_UNTIL,

                ],

            )

            .unwrap();

            tx.execute(

                sql!(

                    "

                    INSERT INTO consensus

                    (docid, unsigned_sha3_256, flavor, valid_after, fresh_until, valid_until)

                    VALUES

                    (?1, ?2, ?3, ?4, ?5, ?6)

                    "

                ),

                params![

                    *CONSENSUS_MD_DOCID,

                    "0000000000000000000000000000000000000000000000000000000000000001", // not the correct hash

                    ConsensusFlavor::Microdesc.name(),

                    *VALID_AFTER,

                    *FRESH_UNTIL,

                    *VALID_UNTIL,

                ],

            )

            .unwrap();

            tx.execute(sql!(

                "

                INSERT INTO consensus_router_descriptor_member

                (consensus_docid, unsigned_sha1, unsigned_sha2)

                VALUES

                (?1, ?2, NULL),

                (?1, ?3, NULL)

                "

            ), params![

                *CONSENSUS_DOCID,

                Sha1::digest(include_bytes!("../testdata/descriptor1-ns-unsigned")),

                Sha1::digest(include_bytes!("../testdata/descriptor2-ns-unsigned")),

            ]).unwrap();

            tx.execute(sql!(

                "

                INSERT INTO consensus_router_descriptor_member

                (consensus_docid, unsigned_sha1, unsigned_sha2)

                VALUES

                (?1, NULL, ?2),

                (?1, NULL, ?3)

                "

            ), params![

                *CONSENSUS_MD_DOCID,

                Sha256::digest(include_bytes!("../testdata/descriptor1-md")),

                Sha256::digest(include_bytes!("../testdata/descriptor2-md")),

            ]).unwrap();

            tx.execute(sql!(

                "

                INSERT INTO authority_key_certificate

                  (docid, kp_auth_id_rsa_sha1, kp_auth_sign_rsa_sha1, dir_key_published, dir_key_expires)

                VALUES

                  (:docid, :id_rsa, :sk_rsa, :published, :expires)

                "

                ),

                named_params! {

                ":docid": *CERT_DOCID,

                ":id_rsa": "49015F787433103580E3B66A1707A00E60F2D15B",

                ":sk_rsa": "C5D153A6F0DA7CC22277D229DCBBF929D0589FE0",

                ":published": 1764543578,

                ":expires": 1772492378,

            }).unwrap();

        })

        .unwrap();

        pool

    }

    #[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());