sightingdb 0.0.5

use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex, PoisonError, RwLock};

use chrono::{DateTime, Utc};
use serde::ser::SerializeStruct;
use serde::{Deserialize, Serialize, Serializer};

use crate::attribute::{Attribute, AttributeView};
use crate::db_log::log_attribute;

/// Namespace holding every value ever written, used to derive consensus.
pub const ALL_NAMESPACE: &str = "_all";
/// Prefix under which reads are recorded ("shadow sightings").
pub const SHADOW_PREFIX: &str = "_shadow/";
/// Prefix holding the server's own configuration, including API keys.
pub const CONFIG_PREFIX: &str = "_config/";
/// Namespace under which API keys live.
pub const APIKEYS_NAMESPACE: &str = "_config/acl/apikeys/";
/// API key seeded on a fresh database, unless `-k` supplies one.
pub const DEFAULT_APIKEY: &str = "changeme";
/// Bumped whenever the on-disk snapshot layout changes incompatibly.
pub const SNAPSHOT_VERSION: u32 = 1;

/// A lookup that did not resolve, rendered as-is into the JSON body.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct NotFound {
    pub error: &'static str,
    pub namespace: String,
    pub value: String,
}

impl NotFound {
    pub fn namespace(namespace: &str, value: &str) -> Self {
        Self {
            error: "Path not found",
            namespace: namespace.to_string(),
            value: value.to_string(),
        }
    }

    pub fn value(namespace: &str, value: &str) -> Self {
        Self {
            error: "Value not found",
            namespace: namespace.to_string(),
            value: value.to_string(),
        }
    }
}

/// Retention rules applied to every write. Both default to "keep everything",
/// so an existing deployment does not start discarding data on upgrade.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct DatabasePolicy {
    /// Hourly statistics buckets kept per attribute; 0 keeps all of them.
    pub stats_retention: usize,
    /// TTL applied to shadow sightings; 0 means they never expire.
    pub shadow_ttl: u64,
}

/// How a single write should behave.
#[derive(Debug, Clone, Copy, Default)]
pub struct WriteOpts {
    /// Count this value towards consensus in [`ALL_NAMESPACE`].
    pub consensus: bool,
    /// Set the attribute's TTL. `None` leaves whatever it already had.
    pub ttl: Option<u64>,
}

/// What a sweep reclaimed.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct SweepReport {
    pub values_removed: usize,
    pub namespaces_removed: usize,
}

impl SweepReport {
    pub fn is_empty(&self) -> bool {
        self.values_removed == 0 && self.namespaces_removed == 0
    }
}

/// One namespace's values.
///
/// Values are behind their own mutex so that concurrent writes to *different*
/// values in the same namespace do not contend: the map lock is only taken for
/// writing when a value is seen for the first time.
#[derive(Default)]
struct Namespace {
    values: RwLock<HashMap<String, Mutex<Attribute>>>,
    /// Set once any attribute here is given a TTL, so that sweeps can skip
    /// namespaces that can never expire — which is all of them by default.
    has_ttl: AtomicBool,
}

impl Namespace {
    fn from_values(values: HashMap<String, Attribute>) -> Self {
        let has_ttl = values.values().any(|attr| attr.ttl > 0);
        Self {
            values: RwLock::new(
                values
                    .into_iter()
                    .map(|(value, attr)| (value, Mutex::new(attr)))
                    .collect(),
            ),
            has_ttl: AtomicBool::new(has_ttl),
        }
    }

    /// Record a sighting, reporting the new count, whether this was the first
    /// time the value appeared here, and a snapshot for the write log.
    fn record(
        &self,
        value: &str,
        when: DateTime<Utc>,
        ttl: Option<u64>,
        retention: usize,
    ) -> (u64, bool, AttributeView) {
        if ttl.is_some_and(|ttl| ttl > 0) {
            self.has_ttl.store(true, Ordering::Relaxed);
        }

        // Fast path: the value already exists, so a read lock is enough and
        // other values in this namespace stay writable.
        {
            let values = self.values.read().unwrap_or_else(PoisonError::into_inner);
            if let Some(cell) = values.get(value) {
                let mut attr = cell.lock().unwrap_or_else(PoisonError::into_inner);
                if let Some(ttl) = ttl {
                    attr.set_ttl(ttl);
                }
                attr.increment(when, retention);
                return (attr.count(), false, attr.view(0, false));
            }
        }

        // Slow path: first sighting of this value here. Deciding "is this new?"
        // under the write lock is what keeps consensus from being double
        // counted when two writers race.
        let mut values = self.values.write().unwrap_or_else(PoisonError::into_inner);
        let is_new = !values.contains_key(value);
        let cell = values
            .entry(value.to_string())
            .or_insert_with(|| Mutex::new(Attribute::new(value)));
        // We hold the map's write lock, so the mutex needs no locking here.
        let attr = cell.get_mut().unwrap_or_else(PoisonError::into_inner);
        if let Some(ttl) = ttl {
            attr.set_ttl(ttl);
        }
        attr.increment(when, retention);
        (attr.count(), is_new, attr.view(0, false))
    }

    /// An expired attribute is invisible to readers even before the sweeper
    /// gets round to reclaiming it.
    fn view(
        &self,
        value: &str,
        consensus: u64,
        with_stats: bool,
        now: DateTime<Utc>,
    ) -> Option<AttributeView> {
        let values = self.values.read().unwrap_or_else(PoisonError::into_inner);
        let cell = values.get(value)?;
        let attr = cell.lock().unwrap_or_else(PoisonError::into_inner);
        (!attr.is_expired(now)).then(|| attr.view(consensus, with_stats))
    }

    fn count(&self, value: &str, now: DateTime<Utc>) -> u64 {
        let values = self.values.read().unwrap_or_else(PoisonError::into_inner);
        values.get(value).map_or(0, |cell| {
            let attr = cell.lock().unwrap_or_else(PoisonError::into_inner);
            if attr.is_expired(now) {
                0
            } else {
                attr.count()
            }
        })
    }

    /// Every live value here, with a placeholder consensus the caller fills in
    /// afterwards — see the lock-ordering note on [`Database`].
    fn all_views(&self, with_stats: bool, now: DateTime<Utc>) -> Vec<AttributeView> {
        let values = self.values.read().unwrap_or_else(PoisonError::into_inner);
        values
            .values()
            .filter_map(|cell| {
                let attr = cell.lock().unwrap_or_else(PoisonError::into_inner);
                (!attr.is_expired(now)).then(|| attr.view(0, with_stats))
            })
            .collect()
    }

    /// Drop expired attributes, returning the values that went.
    fn remove_expired(&self, now: DateTime<Utc>) -> Vec<String> {
        // Nothing here has ever had a TTL, so nothing here can expire.
        if !self.has_ttl.load(Ordering::Relaxed) {
            return Vec::new();
        }

        // Check under a read lock first: sweeps usually find nothing, and
        // taking the write lock would block every reader of this namespace.
        {
            let values = self.values.read().unwrap_or_else(PoisonError::into_inner);
            let any_expired = values.values().any(|cell| {
                cell.lock()
                    .unwrap_or_else(PoisonError::into_inner)
                    .is_expired(now)
            });
            if !any_expired {
                return Vec::new();
            }
        }

        let mut values = self.values.write().unwrap_or_else(PoisonError::into_inner);
        let mut removed = Vec::new();
        values.retain(|value, cell| {
            let expired = cell
                .get_mut()
                .unwrap_or_else(PoisonError::into_inner)
                .is_expired(now);
            if expired {
                removed.push(value.clone());
            }
            !expired
        });
        removed
    }

    /// Give back one consensus count, dropping the entry when it reaches zero.
    /// Done under the write lock so a concurrent write cannot resurrect a value
    /// between the decrement and the removal.
    fn release(&self, value: &str) {
        let mut values = self.values.write().unwrap_or_else(PoisonError::into_inner);
        let Some(cell) = values.get_mut(value) else {
            return;
        };
        let remaining = cell
            .get_mut()
            .unwrap_or_else(PoisonError::into_inner)
            .decrement();
        if remaining == 0 {
            values.remove(value);
        }
    }

    /// The values stored here, live or not, for consensus bookkeeping on delete.
    fn value_names(&self) -> Vec<String> {
        self.values
            .read()
            .unwrap_or_else(PoisonError::into_inner)
            .keys()
            .cloned()
            .collect()
    }

    fn is_empty(&self) -> bool {
        self.values
            .read()
            .unwrap_or_else(PoisonError::into_inner)
            .is_empty()
    }
}

/// In-memory store: namespace -> value -> attribute.
///
/// Every method takes `&self`; there is no global lock. Namespaces are handed
/// out as `Arc`s so the outer map's lock is released before any value is
/// touched.
///
/// **Lock ordering:** outer map, then a namespace's value map, then a single
/// attribute — and never two namespaces at once. Anything needing a second
/// namespace (consensus lives in `_all`) must finish with the first one before
/// reaching for it, or two writers can deadlock.
#[derive(Default)]
pub struct Database {
    namespaces: RwLock<HashMap<String, Arc<Namespace>>>,
    policy: DatabasePolicy,
}

impl Database {
    /// A database with the default (keep-everything) policy. Production code
    /// always has a policy to hand and calls [`Database::with_policy`].
    #[cfg(test)]
    pub fn new() -> Database {
        Database::with_policy(DatabasePolicy::default())
    }

    pub fn with_policy(policy: DatabasePolicy) -> Database {
        Database {
            namespaces: RwLock::new(HashMap::new()),
            policy,
        }
    }

    /// Rebuild a database from a snapshot. No API key is seeded here: the
    /// snapshot carries whatever keys were registered when it was written.
    pub fn from_snapshot(data: SnapshotData, policy: DatabasePolicy) -> Database {
        let namespaces = data
            .namespaces
            .into_iter()
            .map(|(name, values)| (name, Arc::new(Namespace::from_values(values))))
            .collect();

        Database {
            namespaces: RwLock::new(namespaces),
            policy,
        }
    }

    /// API keys found in a snapshot written by an older build, which stored
    /// them as `_config/acl/apikeys/<key>` namespaces.
    ///
    /// Permissions now come from the configuration instead; this exists only so
    /// that upgrading does not lock an existing deployment out of its own data.
    pub fn legacy_apikeys(&self) -> Vec<String> {
        self.namespaces
            .read()
            .unwrap_or_else(PoisonError::into_inner)
            .keys()
            .filter_map(|name| name.strip_prefix(APIKEYS_NAMESPACE))
            .filter(|key| !key.is_empty())
            .map(String::from)
            .collect()
    }

    /// Record one sighting of `value` in `path` at `when`, returning the new count.
    ///
    /// When `opts.consensus` is set, the value is also counted in
    /// [`ALL_NAMESPACE`] — but only the *first* time it appears in this
    /// namespace, since consensus means "how many namespaces have seen this
    /// value", not "how many times was it written".
    pub fn write(&self, path: &str, value: &str, when: DateTime<Utc>, opts: WriteOpts) -> u64 {
        // Shadow sightings get their retention from policy rather than from the
        // caller, which is what bounds `_shadow/*` growth.
        let ttl = match opts.ttl {
            Some(ttl) => Some(ttl),
            None if path.starts_with(SHADOW_PREFIX) && self.policy.shadow_ttl > 0 => {
                Some(self.policy.shadow_ttl)
            }
            None => None,
        };

        let namespace = self.namespace_or_create(path);
        let (count, is_new, mut view) =
            namespace.record(value, when, ttl, self.policy.stats_retention);

        // The namespace's locks are released by now, so reaching into `_all`
        // here respects the ordering rule above.
        if opts.consensus && is_new {
            self.write(ALL_NAMESPACE, value, when, WriteOpts::default());
        }

        view.consensus = self.count(ALL_NAMESPACE, value);
        log_attribute(path, &view);

        count
    }

    pub fn view(
        &self,
        path: &str,
        value: &str,
        consensus: u64,
        with_stats: bool,
    ) -> Option<AttributeView> {
        self.namespace(path)?
            .view(value, consensus, with_stats, Utc::now())
    }

    pub fn count(&self, path: &str, value: &str) -> u64 {
        let now = Utc::now();
        self.namespace(path)
            .map_or(0, |namespace| namespace.count(value, now))
    }

    pub fn namespace_exists(&self, namespace: &str) -> bool {
        self.namespaces
            .read()
            .unwrap_or_else(PoisonError::into_inner)
            .contains_key(namespace)
    }

    /// Every live attribute stored in `namespace`, or `None` if it does not exist.
    ///
    /// Consensus is filled in only after the namespace's lock has been dropped,
    /// so that this never holds two namespaces at once.
    pub fn namespace_views(&self, namespace: &str) -> Option<Vec<AttributeView>> {
        let mut views = self.namespace(namespace)?.all_views(false, Utc::now());
        for view in &mut views {
            view.consensus = self.count(ALL_NAMESPACE, &view.value);
        }
        Some(views)
    }

    /// Drop a namespace, giving back the consensus its values were holding.
    pub fn delete(&self, name: &str) -> bool {
        let Some(namespace) = self.namespace(name) else {
            return false;
        };
        let values = namespace.value_names();
        drop(namespace);

        let removed = self
            .namespaces
            .write()
            .unwrap_or_else(PoisonError::into_inner)
            .remove(name)
            .is_some();

        if removed && counts_towards_consensus(name) {
            for value in values {
                self.release_consensus(&value);
            }
        }
        removed
    }

    /// Reclaim expired attributes and the namespaces left empty by them.
    pub fn sweep(&self, now: DateTime<Utc>) -> SweepReport {
        let entries: Vec<(String, Arc<Namespace>)> = {
            let map = self
                .namespaces
                .read()
                .unwrap_or_else(PoisonError::into_inner);
            map.iter()
                .map(|(name, namespace)| (name.clone(), Arc::clone(namespace)))
                .collect()
        };

        let mut report = SweepReport::default();
        for (name, namespace) in &entries {
            // API keys have no TTL and must never be swept out from under the ACL.
            if name.starts_with(CONFIG_PREFIX) {
                continue;
            }

            let expired = namespace.remove_expired(now);
            report.values_removed += expired.len();

            if counts_towards_consensus(name) {
                for value in expired {
                    self.release_consensus(&value);
                }
            }
        }

        // Our own handles must go before pruning, or `strong_count` below would
        // see them and conclude every namespace is still in use.
        drop(entries);
        report.namespaces_removed = self.prune_empty();
        report
    }

    /// A borrowed, streaming view for serialization. Namespaces are locked one
    /// at a time as they are written, so this never copies the whole database.
    pub fn snapshot(&self) -> Snapshot<'_> {
        Snapshot(self)
    }

    pub fn namespace_count(&self) -> usize {
        self.namespaces
            .read()
            .unwrap_or_else(PoisonError::into_inner)
            .len()
    }

    fn release_consensus(&self, value: &str) {
        if let Some(all) = self.namespace(ALL_NAMESPACE) {
            all.release(value);
        }
    }

    fn prune_empty(&self) -> usize {
        let mut map = self
            .namespaces
            .write()
            .unwrap_or_else(PoisonError::into_inner);
        let before = map.len();
        map.retain(|name, namespace| {
            if name.starts_with(CONFIG_PREFIX) {
                return true;
            }
            // Only drop a namespace nobody else is holding: a writer that
            // already took an `Arc` would otherwise record its sighting into an
            // orphaned namespace and lose it.
            Arc::strong_count(namespace) > 1 || !namespace.is_empty()
        });
        before - map.len()
    }

    fn namespace(&self, name: &str) -> Option<Arc<Namespace>> {
        self.namespaces
            .read()
            .unwrap_or_else(PoisonError::into_inner)
            .get(name)
            .cloned()
    }

    fn namespace_or_create(&self, name: &str) -> Arc<Namespace> {
        if let Some(namespace) = self.namespace(name) {
            return namespace;
        }
        self.namespaces
            .write()
            .unwrap_or_else(PoisonError::into_inner)
            .entry(name.to_string())
            .or_default()
            .clone()
    }
}

/// Namespaces whose values were counted towards consensus when written, and so
/// must give that count back when they go away.
fn counts_towards_consensus(name: &str) -> bool {
    name != ALL_NAMESPACE && !name.starts_with(SHADOW_PREFIX) && !name.starts_with(CONFIG_PREFIX)
}

// ---------------------------------------------------------------------------
// Snapshots
// ---------------------------------------------------------------------------

/// Owned form of a snapshot, used when loading from disk.
#[derive(Debug, Deserialize)]
pub struct SnapshotData {
    pub version: u32,
    pub namespaces: HashMap<String, HashMap<String, Attribute>>,
}

pub struct Snapshot<'a>(&'a Database);

impl Serialize for Snapshot<'_> {
    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
        let mut out = serializer.serialize_struct("Snapshot", 2)?;
        out.serialize_field("version", &SNAPSHOT_VERSION)?;
        out.serialize_field("namespaces", &NamespacesRef(self.0))?;
        out.end()
    }
}

struct NamespacesRef<'a>(&'a Database);

impl Serialize for NamespacesRef<'_> {
    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
        use serde::ser::SerializeMap;

        // Only the names are copied up front; each namespace is locked, written
        // and released in turn.
        let entries: Vec<(String, Arc<Namespace>)> = {
            let map = self
                .0
                .namespaces
                .read()
                .unwrap_or_else(PoisonError::into_inner);
            map.iter()
                .map(|(name, namespace)| (name.clone(), Arc::clone(namespace)))
                .collect()
        };

        let mut out = serializer.serialize_map(Some(entries.len()))?;
        for (name, namespace) in &entries {
            out.serialize_entry(name, &NamespaceRef(namespace))?;
        }
        out.end()
    }
}

struct NamespaceRef<'a>(&'a Namespace);

impl Serialize for NamespaceRef<'_> {
    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
        use serde::ser::SerializeMap;

        let values = self.0.values.read().unwrap_or_else(PoisonError::into_inner);

        let mut out = serializer.serialize_map(Some(values.len()))?;
        for (value, cell) in values.iter() {
            let attr = cell.lock().unwrap_or_else(PoisonError::into_inner);
            out.serialize_entry(value, &*attr)?;
        }
        out.end()
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn at(secs: i64) -> DateTime<Utc> {
        DateTime::from_timestamp(secs, 0).expect("timestamp in range")
    }

    fn consensus() -> WriteOpts {
        WriteOpts {
            consensus: true,
            ttl: None,
        }
    }

    fn with_ttl(ttl: u64) -> WriteOpts {
        WriteOpts {
            consensus: true,
            ttl: Some(ttl),
        }
    }

    #[test]
    fn write_returns_the_running_count() {
        let db = Database::default();

        assert_eq!(db.write("ns", "1.2.3.4", at(100), consensus()), 1);
        assert_eq!(db.write("ns", "1.2.3.4", at(200), consensus()), 2);
        assert_eq!(db.count("ns", "1.2.3.4"), 2);
    }

    #[test]
    fn consensus_counts_namespaces_not_writes() {
        let db = Database::default();

        db.write("my/namespace", "127.0.0.1", at(100), consensus());
        db.write("another/namespace", "127.0.0.1", at(200), consensus());
        db.write("another/namespace", "127.0.0.1", at(300), consensus());

        assert_eq!(db.count(ALL_NAMESPACE, "127.0.0.1"), 2);
    }

    #[test]
    fn a_new_value_in_an_existing_namespace_still_counts_for_consensus() {
        let db = Database::default();

        db.write("ns", "a", at(100), consensus());
        db.write("ns", "b", at(100), consensus());

        assert_eq!(db.count(ALL_NAMESPACE, "b"), 1);
    }

    #[test]
    fn writes_without_consensus_leave_all_alone() {
        let db = Database::default();

        db.write("ns", "a", at(100), WriteOpts::default());

        assert_eq!(db.count(ALL_NAMESPACE, "a"), 0);
    }

    #[test]
    fn missing_lookups_are_zero_and_none() {
        let db = Database::default();

        assert_eq!(db.count("nope", "nope"), 0);
        assert!(db.view("nope", "nope", 0, false).is_none());
        assert!(!db.namespace_exists("nope"));
        assert!(db.namespace_views("nope").is_none());
    }

    /// Older builds kept API keys as namespaces. We no longer write them, but
    /// we must still recognise them in a restored snapshot.
    #[test]
    fn legacy_apikeys_are_recovered_from_old_snapshots() {
        let db = Database::default();
        assert!(db.legacy_apikeys().is_empty());

        db.write(
            &format!("{APIKEYS_NAMESPACE}{DEFAULT_APIKEY}"),
            "",
            at(100),
            WriteOpts::default(),
        );
        db.write(
            &format!("{APIKEYS_NAMESPACE}secret"),
            "",
            at(100),
            WriteOpts::default(),
        );

        let mut keys = db.legacy_apikeys();
        keys.sort();
        assert_eq!(keys, [DEFAULT_APIKEY, "secret"]);
    }

    #[test]
    fn a_fresh_database_stores_no_keys() {
        let db = Database::new();
        assert!(db.legacy_apikeys().is_empty());
    }

    // -- delete ------------------------------------------------------------

    #[test]
    fn delete_removes_the_namespace_once() {
        let db = Database::default();
        db.write("ns", "a", at(100), consensus());

        assert!(db.delete("ns"));
        assert!(!db.delete("ns"));
        assert!(!db.namespace_exists("ns"));
    }

    #[test]
    fn delete_gives_back_the_consensus_it_was_holding() {
        let db = Database::default();
        db.write("a/ns", "v", at(100), consensus());
        db.write("b/ns", "v", at(100), consensus());
        assert_eq!(db.count(ALL_NAMESPACE, "v"), 2);

        db.delete("a/ns");
        assert_eq!(db.count(ALL_NAMESPACE, "v"), 1);

        // The last holder going away retires the `_all` entry entirely.
        db.delete("b/ns");
        assert_eq!(db.count(ALL_NAMESPACE, "v"), 0);
    }

    // -- TTL ---------------------------------------------------------------

    #[test]
    fn an_expired_attribute_is_invisible_before_it_is_swept() {
        let db = Database::default();
        // Written in 1970 with a one minute TTL, so it is long expired by now.
        db.write("ns", "v", at(1000), with_ttl(60));

        assert!(db.view("ns", "v", 0, false).is_none());
        assert_eq!(db.count("ns", "v"), 0);
        assert_eq!(db.namespace_views("ns").unwrap().len(), 0);
    }

    #[test]
    fn a_live_attribute_reports_its_ttl() {
        let db = Database::default();
        db.write("ns", "v", Utc::now(), with_ttl(3600));

        let view = db.view("ns", "v", 0, false).unwrap();
        assert_eq!(view.ttl, 3600);
    }

    #[test]
    fn writing_again_without_a_ttl_keeps_the_existing_one() {
        let db = Database::default();
        db.write("ns", "v", Utc::now(), with_ttl(3600));
        db.write("ns", "v", Utc::now(), consensus());

        assert_eq!(db.view("ns", "v", 0, false).unwrap().ttl, 3600);
    }

    #[test]
    fn sweeping_reclaims_expired_values_and_their_consensus() {
        let db = Database::default();
        db.write("a/ns", "v", at(1000), with_ttl(60));
        db.write("b/ns", "v", at(1000), consensus());
        assert_eq!(db.count(ALL_NAMESPACE, "v"), 2);

        let report = db.sweep(Utc::now());

        assert_eq!(report.values_removed, 1);
        assert_eq!(report.namespaces_removed, 1); // a/ns is now empty
        assert!(!db.namespace_exists("a/ns"));
        assert!(db.namespace_exists("b/ns"));
        // b/ns still holds the value, so consensus drops to one rather than zero.
        assert_eq!(db.count(ALL_NAMESPACE, "v"), 1);
    }

    #[test]
    fn sweeping_leaves_live_data_alone() {
        let db = Database::default();
        db.write("ns", "forever", at(1000), consensus());
        db.write("ns", "later", Utc::now(), with_ttl(3600));

        assert_eq!(db.sweep(Utc::now()), SweepReport::default());
        assert_eq!(db.namespace_views("ns").unwrap().len(), 2);
    }

    /// A legacy key namespace has no TTL, but the sweeper skips the whole
    /// `_config` tree anyway rather than relying on that.
    #[test]
    fn sweeping_never_touches_api_keys() {
        let db = Database::default();
        let namespace = format!("{APIKEYS_NAMESPACE}{DEFAULT_APIKEY}");
        db.write(&namespace, "", at(100), WriteOpts::default());

        db.sweep(Utc::now());

        assert!(db.namespace_exists(&namespace));
        assert_eq!(db.legacy_apikeys(), [DEFAULT_APIKEY]);
    }

    #[test]
    fn shadow_sightings_inherit_the_policy_ttl() {
        let db = Database::with_policy(DatabasePolicy {
            stats_retention: 0,
            shadow_ttl: 60,
        });
        db.write("_shadow/ns", "v", at(1000), WriteOpts::default());

        // Expired by policy, without the caller asking for a TTL.
        assert_eq!(db.count("_shadow/ns", "v"), 0);
        assert_eq!(db.sweep(Utc::now()).values_removed, 1);
    }

    #[test]
    fn the_policy_ttl_does_not_leak_into_ordinary_namespaces() {
        let db = Database::with_policy(DatabasePolicy {
            stats_retention: 0,
            shadow_ttl: 60,
        });
        db.write("ns", "v", at(1000), consensus());

        assert_eq!(db.count("ns", "v"), 1);
    }

    #[test]
    fn stats_retention_is_applied_on_write() {
        let db = Database::with_policy(DatabasePolicy {
            stats_retention: 2,
            shadow_ttl: 0,
        });
        for hour in 0..5 {
            db.write("ns", "v", at(hour * 3600), consensus());
        }

        let view = db.view("ns", "v", 0, true).unwrap();
        assert_eq!(view.stats.unwrap().len(), 2);
        assert_eq!(view.count, 5);
    }

    // -- snapshots ---------------------------------------------------------

    #[test]
    fn a_snapshot_round_trips() {
        let db = Database::new();
        db.write("my/ns", "1.2.3.4", at(1_600_000_000), consensus());
        db.write("my/ns", "1.2.3.4", at(1_600_003_600), consensus());
        db.write("other/ns", "1.2.3.4", at(1_600_000_000), with_ttl(99));

        let json = serde_json::to_string(&db.snapshot()).unwrap();
        let data: SnapshotData = serde_json::from_str(&json).unwrap();
        assert_eq!(data.version, SNAPSHOT_VERSION);

        let restored = Database::from_snapshot(data, DatabasePolicy::default());

        assert_eq!(restored.count("my/ns", "1.2.3.4"), 2);
        assert_eq!(restored.count(ALL_NAMESPACE, "1.2.3.4"), 2);

        let view = restored.view("my/ns", "1.2.3.4", 0, true).unwrap();
        assert_eq!(view.first_seen, 1_600_000_000);
        assert_eq!(view.last_seen, 1_600_003_600);
        assert_eq!(view.stats.unwrap().len(), 2);
    }

    #[test]
    fn a_restored_database_still_knows_about_ttls() {
        let db = Database::new();
        db.write("ns", "v", at(1000), with_ttl(60));

        let json = serde_json::to_string(&db.snapshot()).unwrap();
        let restored = Database::from_snapshot(
            serde_json::from_str(&json).unwrap(),
            DatabasePolicy::default(),
        );

        // `has_ttl` must survive the round trip, or the sweeper would skip this.
        assert_eq!(restored.sweep(Utc::now()).values_removed, 1);
    }

    #[test]
    fn an_empty_database_snapshots_cleanly() {
        let db = Database::default();
        let json = serde_json::to_string(&db.snapshot()).unwrap();

        assert_eq!(json, r#"{"version":1,"namespaces":{}}"#);
    }

    // -- concurrency -------------------------------------------------------

    #[test]
    fn concurrent_writes_to_one_value_are_all_counted() {
        const THREADS: usize = 8;
        const PER_THREAD: usize = 500;

        let db = Arc::new(Database::default());
        std::thread::scope(|scope| {
            for _ in 0..THREADS {
                let db = Arc::clone(&db);
                scope.spawn(move || {
                    for i in 0..PER_THREAD {
                        db.write("ns", "shared", at(i as i64), consensus());
                    }
                });
            }
        });

        assert_eq!(db.count("ns", "shared"), (THREADS * PER_THREAD) as u64);
        // Every writer raced on the same first sighting; consensus must still
        // have counted the namespace exactly once.
        assert_eq!(db.count(ALL_NAMESPACE, "shared"), 1);
    }

    #[test]
    fn concurrent_writes_across_namespaces_agree_on_consensus() {
        const THREADS: usize = 8;

        let db = Arc::new(Database::default());
        std::thread::scope(|scope| {
            for t in 0..THREADS {
                let db = Arc::clone(&db);
                scope.spawn(move || {
                    for i in 0..200 {
                        db.write(&format!("ns/{t}"), "shared", at(i), consensus());
                    }
                });
            }
        });

        assert_eq!(db.count(ALL_NAMESPACE, "shared"), THREADS as u64);
        for t in 0..THREADS {
            assert_eq!(db.count(&format!("ns/{t}"), "shared"), 200);
        }
    }

    /// Readers and writers hitting `_all` and a namespace from both directions
    /// at once: the lock-ordering rule is what keeps this from deadlocking.
    #[test]
    fn readers_and_writers_do_not_deadlock() {
        let db = Arc::new(Database::default());
        std::thread::scope(|scope| {
            for t in 0..8 {
                let db = Arc::clone(&db);
                scope.spawn(move || {
                    for i in 0..500 {
                        // 3 and 20 are coprime, so every value really does land
                        // in all three namespaces rather than sticking to one.
                        let value = format!("v{}", i % 20);
                        db.write(&format!("ns/{}", i % 3), &value, at(i), consensus());
                        db.count(ALL_NAMESPACE, &value);
                        db.view(&format!("ns/{}", t % 3), &value, 0, true);
                        db.namespace_views(&format!("ns/{}", i % 3));
                    }
                });
            }
        });

        for v in 0..20 {
            assert_eq!(db.count(ALL_NAMESPACE, &format!("v{v}")), 3);
        }
    }

    /// A sweep running against live writers must never lose a sighting to the
    /// empty-namespace pruning race.
    #[test]
    fn sweeping_concurrently_with_writers_loses_nothing() {
        let db = Arc::new(Database::default());
        let stop = Arc::new(AtomicBool::new(false));

        std::thread::scope(|scope| {
            let sweeper_db = Arc::clone(&db);
            let sweeper_stop = Arc::clone(&stop);
            scope.spawn(move || {
                while !sweeper_stop.load(Ordering::Relaxed) {
                    sweeper_db.sweep(Utc::now());
                }
            });

            for t in 0..4 {
                let db = Arc::clone(&db);
                scope.spawn(move || {
                    for _ in 0..500 {
                        db.write(&format!("ns/{t}"), "v", Utc::now(), consensus());
                    }
                });
            }

            // Writers finish inside the scope; stop the sweeper afterwards.
            scope.spawn({
                let stop = Arc::clone(&stop);
                move || {
                    std::thread::sleep(std::time::Duration::from_millis(300));
                    stop.store(true, Ordering::Relaxed);
                }
            });
        });

        for t in 0..4 {
            assert_eq!(db.count(&format!("ns/{t}"), "v"), 500, "namespace ns/{t}");
        }
    }
}