pub(crate) fn from_path_and_mistrust<P: AsRef<Path>>(

        path: P,

        mistrust: &fs_mistrust::Mistrust,

        mut readonly: bool,

    ) -> Result<Self> {

        let path = path.as_ref();

        let sqlpath = path.join("dir.sqlite3");

        let blobpath = path.join("dir_blobs/");

        let lockpath = path.join("dir.lock");

        let verifier = mistrust.verifier().permit_readable().check_content();

        let blob_dir = if readonly {

            verifier.secure_dir(blobpath)?

            verifier.make_secure_dir(blobpath)?

        for p in [&lockpath, &sqlpath] {

            match mistrust

                .verifier()

                .permit_readable()

                .require_file()

                .check(p)

                Ok(()) | Err(fs_mistrust::Error::NotFound(_)) => {}

        let mut lockfile = fslock::LockFile::open(&lockpath).map_err(Error::from_lockfile)?;

        if !readonly && !lockfile.try_lock().map_err(Error::from_lockfile)? {

        };

        let flags = if readonly {

            OpenFlags::SQLITE_OPEN_READ_ONLY

            OpenFlags::SQLITE_OPEN_READ_WRITE | OpenFlags::SQLITE_OPEN_CREATE

        let conn = rusqlite::Connection::open_with_flags(&sqlpath, flags)?;

        let mut store = SqliteStore::from_conn_internal(conn, blob_dir, readonly)?;

        store.sql_path = Some(sqlpath);

        store.lockfile = Some(lockfile);

        Ok(store)

    }

    fn from_conn(conn: rusqlite::Connection, blob_dir: CheckedDir) -> Result<Self> {

        Self::from_conn_internal(conn, blob_dir, false)

    }

    fn from_conn_internal(

        conn: rusqlite::Connection,

        blob_dir: CheckedDir,

        readonly: bool,

    ) -> Result<Self> {

        // sqlite (as of Jun 2024) does not enforce foreign keys automatically unless you set this

        // pragma on the connection.

        conn.pragma_update(None, "foreign_keys", "ON")?;

        let mut result = SqliteStore {

            conn,

            blob_dir,

            lockfile: None,

            sql_path: None,

        };

        result.check_schema(readonly)?;

        Ok(result)

    }

    fn check_schema(&mut self, readonly: bool) -> Result<()> {

        let tx = self.conn.transaction()?;

        let db_n_tables: u32 = tx.query_row(

            "SELECT COUNT(name) FROM sqlite_master

             WHERE type='table'

             AND name NOT LIKE 'sqlite_%'",

            [],

            |row| row.get(0),

        )?;

        let db_exists = db_n_tables > 0;

        // Update the schema from current_vsn to the latest (does not commit)

        let update_schema = |tx: &rusqlite::Transaction, current_vsn| {

            for (from_vsn, update) in UPDATE_SCHEMA.iter().enumerate() {

                let from_vsn = u32::try_from(from_vsn).expect("schema version >2^32");

                let new_vsn = from_vsn + 1;

                if current_vsn < new_vsn {

                    tx.execute_batch(update)?;

                    tx.execute(UPDATE_SCHEMA_VERSION, params![new_vsn, new_vsn])?;

            Ok::<_, Error>(())

        };

        if !db_exists {

            if !readonly {

                tx.execute_batch(INSTALL_V0_SCHEMA)?;

                update_schema(&tx, 0)?;

                tx.commit()?;

            return Ok(());

        }

        let (version, readable_by): (u32, u32) = tx.query_row(

            "SELECT version, readable_by FROM TorSchemaMeta

             WHERE name = 'TorDirStorage'",

            [],

            |row| Ok((row.get(0)?, row.get(1)?)),

        )?;

        if version < SCHEMA_VERSION {

        } else if readable_by > SCHEMA_VERSION {

            return Err(Error::UnrecognizedSchema {

                schema: readable_by,

                supported: SCHEMA_VERSION,

            });

        }

        // rolls back the transaction, but nothing was done.

        Ok(())

    }

    fn read_blob(&self, path: &str) -> Result<StdResult<InputString, AbsentBlob>> {

        let file = match self.blob_dir.open(path, OpenOptions::new().read(true)) {

            Ok(file) => file,

        InputString::load(file)

            .map_err(|err| Error::CacheFile {

            })

            .map(Ok)

    }

    fn save_blob_internal(

        &mut self,

        contents: &[u8],

        doctype: &str,

        digest_type: &str,

        digest: &[u8],

        expires: OffsetDateTime,

    ) -> Result<blob_handle::SavedBlobHandle<'_>> {

        let digest = hex::encode(digest);

        let digeststr = format!("{}-{}", digest_type, digest);

        let fname = format!("{}_{}", doctype, digeststr);

        let full_path = self.blob_dir.join(&fname)?;

        let unlinker = blob_handle::Unlinker::new(&full_path);

        self.blob_dir

            .write_and_replace(&fname, contents)

            .map_err(|e| match e {

            })?;

        let tx = self.conn.unchecked_transaction()?;

        tx.execute(INSERT_EXTDOC, params![digeststr, expires, doctype, fname])?;

        Ok(blob_handle::SavedBlobHandle::new(

            tx, fname, digeststr, unlinker,

        ))

    }

    fn latest_consensus_internal(

        &self,

        flavor: ConsensusFlavor,

        pending: Option<bool>,

    ) -> Result<StdResult<InputString, AbsentBlob>> {

        trace!(?flavor, ?pending, "Loading latest consensus from cache");

        let rv: Option<(OffsetDateTime, OffsetDateTime, String)> = match pending {

            None => self

                .conn

                .query_row(FIND_CONSENSUS, params![flavor.name()], |row| row.try_into())

                .optional()?,

            Some(pending_val) => self

                .conn

                .query_row(

                    FIND_CONSENSUS_P,

                    params![pending_val, flavor.name()],

                    |row| row.try_into(),

                )

                .optional()?,

        if let Some((_va, _vu, filename)) = rv {

            self.read_blob(&filename)

            Ok(Err(AbsentBlob::NothingToRead))

    }

    fn save_blob(

        &mut self,

        contents: &[u8],

        doctype: &str,

        digest_type: &str,

        digest: &[u8],

        expires: OffsetDateTime,

    ) -> Result<String> {

        let h = self.save_blob_internal(contents, doctype, digest_type, digest, expires)?;

        let fname = h.fname().to_string();

        h.commit()?;

        Ok(fname)

    }

    fn latest_consensus_time(&self, flavor: ConsensusFlavor) -> Result<Option<OffsetDateTime>> {

        Ok(self

            .latest_consensus_meta(flavor)?

            .map(|m| m.lifetime().valid_after().into()))

    }

    fn remove_blob_or_warn<P: AsRef<Path>>(&self, fname: P) {

        let fname = fname.as_ref();

        if let Err(e) = self.blob_dir.remove_file(fname) {

        }

    }

    fn remove_unreferenced_blobs(

        &self,

        now: OffsetDateTime,

        expiration: &ExpirationConfig,

    ) -> Result<()> {

        for ent in self.blob_dir.read_directory(".")?.flatten() {

            let md_error = |io_error| Error::CacheFile {

            if ent

                .metadata()

                .map_err(md_error)?

                .modified()

                .map_err(md_error)?

                + expiration.consensuses

                >= now

                continue;

            }

            let filename = match ent.file_name().into_string() {

                Ok(s) => s,

            let found: (u32,) =

                self.conn

                    .query_row(COUNT_EXTDOC_BY_PATH, params![&filename], |row| {

                        row.try_into()

                    })?;

            if found == (0,) {

                warn!("Removing unreferenced file '{}' from blob store", &filename);

                self.remove_blob_or_warn(ent.file_name());

        Ok(())

    }

    fn remove_entries_for_vanished_blobs<'a>(

        blob_dir: &CheckedDir,

        tx: &Transaction<'a>,

    ) -> Result<usize> {

        let in_directory: HashSet<PathBuf> = blob_dir

            .read_directory(".")?

            .flatten()

            .map(|dir_entry| PathBuf::from(dir_entry.file_name()))

            .collect();

        let in_db: Vec<String> = tx

            .prepare(FIND_ALL_EXTDOC_FILENAMES)?

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

            .collect::<StdResult<Vec<String>, _>>()?;

        let mut n_removed = 0;

        for fname in in_db {

            if in_directory.contains(Path::new(&fname)) {

                continue;

            }

            n_removed += tx.execute(DELETE_EXTDOC_BY_FILENAME, [fname])?;

        Ok(n_removed)

    }

    fn is_readonly(&self) -> bool {

        match &self.lockfile {

            Some(f) => !f.owns_lock(),

            None => false,

    }

    fn upgrade_to_readwrite(&mut self) -> Result<bool> {

        if self.is_readonly() && self.sql_path.is_some() {

            let lf = self

                .lockfile

                .as_mut()

                .expect("No lockfile open; cannot upgrade to read-write storage");

            if !lf.try_lock().map_err(Error::from_lockfile)? {

            }

            // Unwrap should be safe due to parent `.is_some()` check

            #[allow(clippy::unwrap_used)]

            match rusqlite::Connection::open(self.sql_path.as_ref().unwrap()) {

                Ok(conn) => {

                    self.conn = conn;

                }

        }

        Ok(true)

    }

    fn expire_all(&mut self, expiration: &ExpirationConfig) -> Result<()> {

        let tx = self.conn.transaction()?;

        let expired_blobs: Vec<String> = {

            let mut stmt = tx.prepare(FIND_EXPIRED_EXTDOCS)?;

            let names: Vec<String> = stmt

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

                .collect::<StdResult<Vec<String>, _>>()?;

            names

        };

        let now = OffsetDateTime::now_utc();

        tx.execute(DROP_OLD_EXTDOCS, [])?;

        tx.execute(DROP_OLD_MICRODESCS, [now - expiration.microdescs])?;

        tx.execute(DROP_OLD_AUTHCERTS, [now - expiration.authcerts])?;

        tx.execute(DROP_OLD_CONSENSUSES, [now - expiration.consensuses])?;

        tx.execute(DROP_OLD_ROUTERDESCS, [now - expiration.router_descs])?;

        tx.execute(DROP_OLD_BRIDGEDESCS, [now, now])?;

        let remove_consensus_blobs = {

            let mut stmt = tx.prepare(FIND_UNREFERENCED_CONSENSUS_EXTDOCS)?;

            let filenames: Vec<String> = stmt

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

                .collect::<StdResult<Vec<String>, _>>()?;

            drop(stmt);

            let mut stmt = tx.prepare(DELETE_EXTDOC_BY_FILENAME)?;

            for fname in filenames.iter() {

                stmt.execute([fname])?;

            filenames

        };

        tx.commit()?;

        let mut remove_blob_files: HashSet<_> = expired_blobs.iter().collect();

        remove_blob_files.extend(remove_consensus_blobs.iter());

        for name in remove_blob_files {

            let fname = self.blob_dir.join(name);

            if let Ok(fname) = fname {

                if let Err(e) = std::fs::remove_file(&fname) {

                }

        self.remove_unreferenced_blobs(now, expiration)?;

        Ok(())

    }

    fn latest_consensus(

        &self,

        flavor: ConsensusFlavor,

        pending: Option<bool>,

    ) -> Result<Option<InputString>> {

        match self.latest_consensus_internal(flavor, pending)? {

            Ok(s) => return Ok(Some(s)),

            Err(AbsentBlob::NothingToRead) => return Ok(None),

    }

    fn latest_consensus_meta(&self, flavor: ConsensusFlavor) -> Result<Option<ConsensusMeta>> {

        let mut stmt = self.conn.prepare(FIND_LATEST_CONSENSUS_META)?;

        let mut rows = stmt.query(params![flavor.name()])?;

        if let Some(row) = rows.next()? {

            Ok(Some(cmeta_from_row(row)?))

            Ok(None)

    }

    fn consensus_by_meta(&self, cmeta: &ConsensusMeta) -> Result<InputString> {

        if let Some((text, _)) =

            self.consensus_by_sha3_digest_of_signed_part(cmeta.sha3_256_of_signed())?

            Ok(text)

            Err(Error::CacheCorruption(

                "couldn't find a consensus we thought we had.",

            ))

    }

    fn consensus_by_sha3_digest_of_signed_part(

        &self,

        d: &[u8; 32],

    ) -> Result<Option<(InputString, ConsensusMeta)>> {

        let digest = hex::encode(d);

        let mut stmt = self

            .conn

            .prepare(FIND_CONSENSUS_AND_META_BY_DIGEST_OF_SIGNED)?;

        let mut rows = stmt.query(params![digest])?;

        if let Some(row) = rows.next()? {

            let meta = cmeta_from_row(row)?;

            let fname: String = row.get(5)?;

            if let Ok(text) = self.read_blob(&fname)? {

                return Ok(Some((text, meta)));

        }

        Ok(None)

    }

    fn store_consensus(

        &mut self,

        cmeta: &ConsensusMeta,

        flavor: ConsensusFlavor,

        pending: bool,

        contents: &str,

    ) -> Result<()> {

        let lifetime = cmeta.lifetime();

        let sha3_of_signed = cmeta.sha3_256_of_signed();

        let sha3_of_whole = cmeta.sha3_256_of_whole();

        let valid_after: OffsetDateTime = lifetime.valid_after().into();

        let fresh_until: OffsetDateTime = lifetime.fresh_until().into();

        let valid_until: OffsetDateTime = lifetime.valid_until().into();

        let expires = valid_until + CONSENSUS_LIFETIME;

        let doctype = format!("con_{}", flavor.name());

        let h = self.save_blob_internal(

            contents.as_bytes(),

            &doctype,

            "sha3-256",

            &sha3_of_whole[..],

            expires,

        )?;

        h.tx().execute(

            INSERT_CONSENSUS,

            params![

                valid_after,

                fresh_until,

                valid_until,

                flavor.name(),

                pending,

                hex::encode(sha3_of_signed),

                h.digest_string()

            ],

        )?;

        h.commit()?;

        Ok(())

    }

    fn mark_consensus_usable(&mut self, cmeta: &ConsensusMeta) -> Result<()> {

        let d = hex::encode(cmeta.sha3_256_of_whole());

        let digest = format!("sha3-256-{}", d);

        let tx = self.conn.transaction()?;

        let n = tx.execute(MARK_CONSENSUS_NON_PENDING, params![digest])?;

        trace!("Marked {} consensuses usable", n);

        tx.commit()?;

        Ok(())

    }

    fn delete_consensus(&mut self, cmeta: &ConsensusMeta) -> Result<()> {

        let d = hex::encode(cmeta.sha3_256_of_whole());

        let digest = format!("sha3-256-{}", d);

        let tx = self.conn.transaction()?;

        tx.execute(REMOVE_CONSENSUS, params![digest])?;

        tx.commit()?;

        Ok(())

    }

    fn authcerts(&self, certs: &[AuthCertKeyIds]) -> Result<HashMap<AuthCertKeyIds, String>> {

        let mut result = HashMap::new();

        let mut stmt = self.conn.prepare(FIND_AUTHCERT)?;

        for ids in certs {

            let id_digest = hex::encode(ids.id_fingerprint.as_bytes());

            let sk_digest = hex::encode(ids.sk_fingerprint.as_bytes());

            if let Some(contents) = stmt

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

                .optional()?

            {

                result.insert(*ids, contents);

            }

        Ok(result)

    }

    fn store_authcerts(&mut self, certs: &[(AuthCertMeta, &str)]) -> Result<()> {

        let tx = self.conn.transaction()?;

        let mut stmt = tx.prepare(INSERT_AUTHCERT)?;

        for (meta, content) in certs {

            let ids = meta.key_ids();

            let id_digest = hex::encode(ids.id_fingerprint.as_bytes());

            let sk_digest = hex::encode(ids.sk_fingerprint.as_bytes());

            let published: OffsetDateTime = meta.published().into();

            let expires: OffsetDateTime = meta.expires().into();

            stmt.execute(params![id_digest, sk_digest, published, expires, content])?;

        stmt.finalize()?;

        tx.commit()?;

        Ok(())

    }

    fn microdescs(&self, digests: &[MdDigest]) -> Result<HashMap<MdDigest, String>> {

        let mut result = HashMap::new();

        let mut stmt = self.conn.prepare(FIND_MD)?;

        for md_digest in digests {

            let h_digest = hex::encode(md_digest);

            if let Some(contents) = stmt

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

                .optional()?

            {

                result.insert(*md_digest, contents);

            }

        Ok(result)

    }

    fn store_microdescs(&mut self, digests: &[(&str, &MdDigest)], when: SystemTime) -> Result<()> {

        let when: OffsetDateTime = when.into();

        let tx = self.conn.transaction()?;

        let mut stmt = tx.prepare(INSERT_MD)?;

        for (content, md_digest) in digests {

            let h_digest = hex::encode(md_digest);

            stmt.execute(params![h_digest, when, content])?;

        stmt.finalize()?;

        tx.commit()?;

        Ok(())

    }

    fn update_microdescs_listed(&mut self, digests: &[MdDigest], when: SystemTime) -> Result<()> {

        let tx = self.conn.transaction()?;

        let mut stmt = tx.prepare(UPDATE_MD_LISTED)?;

        let when: OffsetDateTime = when.into();

        for md_digest in digests {

            let h_digest = hex::encode(md_digest);

            stmt.execute(params![when, h_digest])?;

        stmt.finalize()?;

        tx.commit()?;

        Ok(())

    }

    fn routerdescs(&self, digests: &[RdDigest]) -> Result<HashMap<RdDigest, String>> {

        let mut result = HashMap::new();

        let mut stmt = self.conn.prepare(FIND_RD)?;

        for rd_digest in digests {

            let h_digest = hex::encode(rd_digest);

            if let Some(contents) = stmt

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

                .optional()?

            {

                result.insert(*rd_digest, contents);

            }

        Ok(result)

    }

    fn store_routerdescs(&mut self, digests: &[(&str, SystemTime, &RdDigest)]) -> Result<()> {

        let tx = self.conn.transaction()?;

        let mut stmt = tx.prepare(INSERT_RD)?;

        for (content, when, rd_digest) in digests {

            let when: OffsetDateTime = (*when).into();

            let h_digest = hex::encode(rd_digest);

            stmt.execute(params![h_digest, when, content])?;

        stmt.finalize()?;

        tx.commit()?;

        Ok(())

    }

    fn lookup_bridgedesc(&self, bridge: &BridgeConfig) -> Result<Option<CachedBridgeDescriptor>> {

        let bridge_line = bridge.to_string();

        Ok(self

            .conn

            .query_row(FIND_BRIDGEDESC, params![bridge_line], |row| {

                let (fetched, document): (OffsetDateTime, _) = row.try_into()?;

                let fetched = fetched.into();

                Ok(CachedBridgeDescriptor { fetched, document })

            })

            .optional()?)

    }

    fn store_bridgedesc(

        &mut self,

        bridge: &BridgeConfig,

        entry: CachedBridgeDescriptor,

        until: SystemTime,

    ) -> Result<()> {

        if self.is_readonly() {

        }

        let bridge_line = bridge.to_string();

        let row = params![

            bridge_line,

            OffsetDateTime::from(entry.fetched),

            OffsetDateTime::from(until),

            entry.document,

        ];

        self.conn.execute(INSERT_BRIDGEDESC, row)?;

        Ok(())

    }

    fn update_protocol_recommendations(

        &mut self,

        valid_after: SystemTime,

        protocols: &tor_netdoc::doc::netstatus::ProtoStatuses,

    ) -> Result<()> {

        let json =

            serde_json::to_string(&protocols).map_err(into_internal!("Cannot encode protocols"))?;

        let params = params![OffsetDateTime::from(valid_after), json];

        self.conn.execute(UPDATE_PROTOCOL_STATUS, params)?;

        Ok(())

    }

    fn cached_protocol_recommendations(

        &self,

    ) -> Result<Option<(SystemTime, tor_netdoc::doc::netstatus::ProtoStatuses)>> {

        let opt_row: Option<(OffsetDateTime, String)> = self

            .conn

            .query_row(FIND_LATEST_PROTOCOL_STATUS, [], |row| {

                Ok((row.get(0)?, row.get(1)?))

            })

            .optional()?;

        let (date, json) = match opt_row {

            Some(v) => v,

            None => return Ok(None),

        let date = date.into();

        let statuses: tor_netdoc::doc::netstatus::ProtoStatuses =

            serde_json::from_str(json.as_str()).map_err(|e| Error::BadJsonInCache(Arc::new(e)))?;

        Ok(Some((date, statuses)))

    }

        pub(super) fn new(

            tx: Transaction<'a>,

            fname: String,

            digeststr: String,

            unlinker: Unlinker,

        ) -> Self {

            Self {

                tx,

                fname,

                digeststr,

                unlinker,

            }

        }

        pub(super) fn tx(&self) -> &Transaction<'a> {

            &self.tx

        }

        pub(super) fn digest_string(&self) -> &str {

            self.digeststr.as_ref()

        }

        pub(super) fn fname(&self) -> &str {

            self.fname.as_ref()

        }

        pub(super) fn commit(self) -> Result<()> {

            // The blob has been written to disk, so it is safe to

            // commit the transaction.

            // If the commit returns an error, self.unlinker will remove the blob.

            // (This could result in a vanished blob if the commit reports an error,

            // but the transaction is still visible in the database.)

            self.tx.commit()?;

            self.unlinker.forget();

            Ok(())

        }

        pub(super) fn new<P: AsRef<Path>>(p: P) -> Self {

            Unlinker {

                p: Some(p.as_ref().to_path_buf()),

            }

        }

        fn forget(mut self) {

            self.p = None;

        }

        fn drop(&mut self) {

            if let Some(p) = self.p.take() {

            }

        }

fn digest_from_hex(s: &str) -> Result<[u8; 32]> {

    let mut bytes = [0_u8; 32];

    hex::decode_to_slice(s, &mut bytes[..]).map_err(Error::BadHexInCache)?;

    Ok(bytes)

}

fn digest_from_dstr(s: &str) -> Result<[u8; 32]> {

    if let Some(stripped) = s.strip_prefix("sha3-256-") {

        digest_from_hex(stripped)

}

fn cmeta_from_row(row: &rusqlite::Row<'_>) -> Result<ConsensusMeta> {

    let va: OffsetDateTime = row.get(0)?;

    let fu: OffsetDateTime = row.get(1)?;

    let vu: OffsetDateTime = row.get(2)?;

    let d_signed: String = row.get(3)?;

    let d_all: String = row.get(4)?;

    let lifetime = Lifetime::new(va.into(), fu.into(), vu.into())

        .map_err(|_| Error::CacheCorruption("inconsistent lifetime in database"))?;

        lifetime,

        digest_from_hex(&d_signed)?,

        digest_from_dstr(&d_all)?,

}