sightingdb 0.7.0

//! The admin interface served at `/_management/`.
//!
//! This is a browser over the database — namespaces as folders, the values
//! inside them, and what each value has been seen doing — plus a view of what
//! the server is configured to do. Namespaces can be created and values added
//! from it; the configuration itself is read-only, since it comes from a file
//! read at startup.
//!
//! It is deliberately *not* served under `_config`, even though there would be
//! no routing conflict: `_config` already names the database namespace holding
//! API keys, and the data paths refuse it outright. Serving an interface under
//! the same word would leave `/_config/` and `/r/_config/` looking alike while
//! doing opposite things.
//!
//! Access requires a key holding the `admin` grant, *regardless* of the
//! `authenticate` setting. Turning authentication off is a decision about the
//! sighting API — it should not hand the admin interface to anyone who can
//! reach the port.

use std::path::PathBuf;

use actix_web::{HttpRequest, HttpResponse, Responder, web};
use serde::{Deserialize, Serialize};

use crate::acl::{Grant, validate_key, validate_namespace};

use crate::error::Message;
use crate::handlers::{SharedState, State};

/// The single-page interface, compiled in so there is nothing to deploy.
const UI: &str = include_str!("ui.html");
/// Vendored so the interface works without internet access. See assets/README.md.
const ECHARTS: &str = include_str!("../../assets/echarts.min.js");
/// The logo, compiled in for the same reason: an interface that fetches its own
/// branding from somewhere else does not work on an air-gapped host. 64px, for
/// a header that shows it at about a third of that.
const LOGO: &[u8] = include_bytes!("../../doc/sightingdb-logo3_64.png");

/// Paging is capped so one request cannot ask the server to sort and serialize
/// an entire large namespace.
const MAX_LIMIT: usize = 500;
const DEFAULT_LIMIT: usize = 50;

#[derive(Debug, Deserialize)]
pub struct BrowseQuery {
    /// Substring filter, case-insensitive.
    #[serde(default)]
    q: String,
    #[serde(default)]
    offset: usize,
    limit: Option<usize>,
}

impl BrowseQuery {
    fn limit(&self) -> usize {
        self.limit.unwrap_or(DEFAULT_LIMIT).clamp(1, MAX_LIMIT)
    }
}

/// Which rejections to list.
#[derive(Debug, Deserialize)]
pub struct RejectionsQuery {
    /// Only rejections under this namespace, matched as a subtree the way an
    /// ACL grant is. Absent lists every namespace the key may read.
    namespace: Option<String>,
    limit: Option<usize>,
}

impl RejectionsQuery {
    fn limit(&self) -> usize {
        self.limit.unwrap_or(DEFAULT_LIMIT).clamp(1, MAX_LIMIT)
    }
}

/// A page of rejections, and how many are held in total.
#[derive(Debug, Serialize)]
pub struct RejectionPage {
    rejections: Vec<crate::rejections::Rejection>,
    /// Everything currently kept, not just this page, so the interface can say
    /// "showing 50 of 1000".
    total: usize,
    /// The cap. Once `total` reaches it, the oldest rejections are being lost.
    capacity: usize,
}

#[derive(Debug, Deserialize)]
pub struct ValuesQuery {
    namespace: String,
    #[serde(default)]
    q: String,
    #[serde(default)]
    offset: usize,
    limit: Option<usize>,
}

impl ValuesQuery {
    fn limit(&self) -> usize {
        self.limit.unwrap_or(DEFAULT_LIMIT).clamp(1, MAX_LIMIT)
    }
}

/// One level of the namespace tree, as the browser walks it.
#[derive(Debug, Deserialize)]
pub struct TreeQuery {
    /// The folder being opened. Empty is the root.
    #[serde(default)]
    path: String,
    /// Substring filter over the child names at this level, case-insensitive.
    #[serde(default)]
    q: String,
    #[serde(default)]
    offset: usize,
    limit: Option<usize>,
}

impl TreeQuery {
    fn limit(&self) -> usize {
        self.limit.unwrap_or(DEFAULT_LIMIT).clamp(1, MAX_LIMIT)
    }
}

/// A namespace to create, which may name a whole path at once.
#[derive(Debug, Deserialize)]
pub struct NewNamespace {
    namespace: String,
}

/// Values to record, one namespace at a time.
///
/// A single value and a bulk paste are the same request with a list of one:
/// the interface has one code path, and so does this.
#[derive(Debug, Deserialize)]
pub struct NewValues {
    namespace: String,
    values: Vec<String>,
    /// Unix seconds. Absent means "now", as on the write API.
    #[serde(default)]
    timestamp: Option<i64>,
    /// Absent leaves whatever TTL an existing value had; 0 clears it.
    #[serde(default)]
    ttl: Option<u64>,
    /// Comma-separated tags applied to every value in this request, merged
    /// with whatever each already carried.
    #[serde(default)]
    tags: String,
}

/// What a create or an add did, per value, so a paste of a thousand lines can
/// report the eight that were rejected without losing the rest.
#[derive(Debug, Serialize)]
pub struct WriteReport {
    namespace: String,
    written: usize,
    /// Each value that was recorded, with its running total afterwards.
    ///
    /// The count comes back from the write itself, taken under the lock that
    /// incremented it, so reporting it costs nothing and saves the interface a
    /// read to find out what a paste actually did.
    #[serde(skip_serializing_if = "Vec::is_empty")]
    counts: Vec<ValueCount>,
    #[serde(skip_serializing_if = "Vec::is_empty")]
    errors: Vec<ValueError>,
}

/// One recorded value and how often it has now been seen.
#[derive(Debug, Serialize)]
pub struct ValueCount {
    value: String,
    count: u64,
    /// Whether this was the first sighting of the value in this namespace.
    new: bool,
}

#[derive(Debug, Serialize)]
pub struct ValueError {
    value: String,
    error: String,
}

/// A value's tags, replaced outright.
#[derive(Debug, Deserialize)]
pub struct TagChange {
    namespace: String,
    value: String,
    /// The whole set, comma-separated. Empty clears it.
    tags: String,
}

/// A storage change from the interface or an automation.
///
/// Name any namespace: the setting lands on its top-level namespace, which is
/// the unit that is paged in and out, and so covers everything under it. Both
/// halves are optional, and `"default"` (or null) on either means "stop saying
/// anything here and take the configured default".
#[derive(Debug, Deserialize)]
pub struct TierChange {
    /// Any namespace in the shard to change. `shard` is the older spelling.
    #[serde(default, alias = "shard")]
    namespace: String,
    #[serde(default)]
    tier: Option<String>,
    /// Seconds a warm shard may sit untouched. Accepts a number or a string,
    /// since a form field hands over text.
    #[serde(default)]
    warm_idle: Option<serde_json::Value>,
}

impl TierChange {
    /// The settings to store, or an explanation of what was unreadable.
    fn entry(&self) -> Result<crate::tier::Entry, String> {
        let tier = match self.tier.as_deref().map(str::trim) {
            None | Some("") | Some("default") | Some("inherit") => None,
            Some(tier) => Some(crate::tier::Tier::parse(tier).map_err(|e| e.to_string())?),
        };

        let warm_idle = match &self.warm_idle {
            None | Some(serde_json::Value::Null) => None,
            Some(serde_json::Value::Number(number)) => match number.as_u64() {
                Some(seconds) => Some(seconds),
                None => return Err("warm_idle cannot be negative".to_string()),
            },
            Some(serde_json::Value::String(text)) => {
                let text = text.trim();
                match text {
                    "" | "default" | "inherit" => None,
                    _ => Some(
                        text.parse::<u64>()
                            .map_err(|_| format!("'{text}' is not a number of seconds"))?,
                    ),
                }
            }
            Some(other) => return Err(format!("warm_idle should be a number, found {other}")),
        };

        Ok(crate::tier::Entry { tier, warm_idle })
    }
}

#[derive(Debug, Deserialize)]
pub struct ValueQuery {
    namespace: String,
    value: String,
}

/// Where one value has been seen, for the relationship graph.
#[derive(Debug, Deserialize)]
pub struct SightingsQuery {
    value: String,
    limit: Option<usize>,
}

impl SightingsQuery {
    /// A graph is read by eye, so the cap is what stays legible rather than
    /// what the server could serialize.
    fn limit(&self) -> usize {
        self.limit.unwrap_or(200).clamp(1, MAX_LIMIT)
    }
}

/// What the server is doing, for the configuration view.
///
/// Configuration is read from a file at startup, so this reports rather than
/// edits: changing it means editing the file and restarting.
#[derive(Debug, Clone, Default, Serialize)]
pub struct ServerInfo {
    pub version: &'static str,
    pub authenticate: bool,
    pub http_enabled: bool,
    pub config_path: String,
    pub dbdir: Option<String>,
    pub snapshot_interval: u64,
    pub sweep_interval: u64,
    pub stats_retention: usize,
    pub shadow_ttl: u64,
    pub dns: Option<DnsInfo>,
    pub zmq: Option<ZmqInfo>,
    pub namespaces: usize,
    pub apikeys: usize,
    pub default_tier: String,
    pub warm_idle: u64,
    /// Shards with a tier of their own.
    pub tiers: Vec<(String, String)>,
    /// What this server stores and who it knows about: the two things that
    /// decide its place in a galaxy.
    pub role: RoleInfo,
    /// When this server's TLS certificate runs out. `None` when it serves
    /// plain HTTP, or when the certificate cannot be read.
    pub tls: Option<crate::tls::Expiry>,
    /// Days of certificate life below which the interface calls it urgent, so
    /// the page and the log agree on what "soon" means.
    pub expiring_soon_days: i64,
}

/// This server's place in a galaxy.
///
/// Derived from `[storage] namespaces` and `[galaxy]`, and reported so that a
/// topology can be described from the outside — which is what a cluster view
/// has to read.
#[derive(Debug, Clone, Serialize)]
pub struct RoleInfo {
    /// `"node"`, `"router"`, or `"both"`.
    ///
    /// A node stores namespaces and forwards nothing; a router stores none of
    /// its own and exists to forward; both does each. The words describe a
    /// configuration rather than a type — any server can be any of them.
    pub kind: &'static str,
    /// Whether this server stores every namespace. A full mirror.
    pub mirrors_everything: bool,
    /// The namespace prefixes stored, when it is not everything. Empty on a
    /// router, which stores none.
    pub namespaces: Vec<String>,
    /// Peers, without their keys — a key is a credential and does not belong
    /// in a response the interface renders.
    pub peers: Vec<String>,
    pub max_hops: u8,
}

impl Default for RoleInfo {
    /// A server that stores everything and knows no peers, which is what every
    /// release before galaxies was.
    fn default() -> Self {
        RoleInfo {
            kind: "node",
            mirrors_everything: true,
            namespaces: Vec::new(),
            peers: Vec::new(),
            max_hops: 0,
        }
    }
}

#[derive(Debug, Clone, Serialize)]
pub struct DnsInfo {
    pub listen: String,
    pub zone: String,
    pub ttl: u32,
    pub rate_limit: u32,
    pub shadow: bool,
    pub exposed: Vec<ExposedInfo>,
}

#[derive(Debug, Clone, Serialize)]
pub struct ExposedInfo {
    pub label: String,
    pub namespace: String,
    pub encoding: String,
}

/// One key as the interface sees it.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct KeyEntry {
    pub key: String,
    pub admin: bool,
    /// Namespace prefixes this key may read. An empty string means all.
    pub read: Vec<String>,
    pub write: Vec<String>,
}

impl KeyEntry {
    pub fn from_grants(key: &str, grants: &[Grant]) -> Self {
        let mut entry = KeyEntry {
            key: key.to_string(),
            admin: false,
            read: Vec::new(),
            write: Vec::new(),
        };
        for grant in grants {
            if grant.admin {
                entry.admin = true;
            }
            if grant.read {
                entry.read.push(grant.prefix.clone());
            }
            if grant.write {
                entry.write.push(grant.prefix.clone());
            }
        }
        entry
    }

    /// Back to grants, collapsing a prefix granted for both into one `rw`.
    fn to_grants(&self) -> Vec<Grant> {
        let mut grants = Vec::new();
        for prefix in &self.read {
            grants.push(Grant {
                prefix: prefix.clone(),
                read: true,
                write: self.write.contains(prefix),
                admin: false,
            });
        }
        for prefix in &self.write {
            if !self.read.contains(prefix) {
                grants.push(Grant {
                    prefix: prefix.clone(),
                    read: false,
                    write: true,
                    admin: false,
                });
            }
        }
        if self.admin {
            grants.push(Grant::admin());
        }
        grants
    }

    fn validate(&self) -> anyhow::Result<()> {
        validate_key(&self.key)?;
        for prefix in self.read.iter().chain(self.write.iter()) {
            validate_namespace(prefix)?;
        }
        if !self.admin && self.read.is_empty() && self.write.is_empty() {
            anyhow::bail!("a key with no grants at all cannot do anything");
        }
        Ok(())
    }
}

#[derive(Debug, Clone, Serialize)]
pub struct ZmqInfo {
    pub endpoint: String,
    pub topics: Vec<String>,
    pub format: String,
    pub require_to_ids: bool,
    pub mapped_types: usize,
}

/// Check the caller holds an admin grant, handing back the key so the caller
/// can go on to check its rights over a particular namespace.
fn require_admin<'a>(state: &SharedState, req: &'a HttpRequest) -> Result<&'a str, HttpResponse> {
    let Some(header) = req.headers().get("Authorization") else {
        log::debug!(
            "Refused {}: no API key, asked for the management interface",
            crate::handlers::peer_of(req)
        );
        return Err(
            HttpResponse::Unauthorized().json(Message::new("An admin API key is required."))
        );
    };
    let Ok(key) = header.to_str() else {
        return Err(HttpResponse::BadRequest()
            .json(Message::new("Authorization header is not valid UTF-8.")));
    };
    if state.acl().is_admin(key) {
        Ok(key)
    } else {
        // Louder than a data refusal: this is someone at the door of the
        // interface that hands out keys.
        log::warn!(
            "{}",
            crate::handlers::refusal(
                &crate::handlers::peer_of(req),
                key,
                state.acl().contains(key),
                "use",
                "the management interface",
            )
        );
        Err(HttpResponse::Forbidden().json(Message::new("That key is not an admin key.")))
    }
}

/// Reaching the interface is not the same as being allowed to read the data in
/// it: an `admin, r:feeds` key browses `feeds/*` and nothing else.
fn require_read(state: &SharedState, key: &str, namespace: &str) -> Result<(), HttpResponse> {
    if state.acl().can_read(key, namespace) {
        Ok(())
    } else {
        log::warn!(
            "{}",
            crate::handlers::refusal("an admin key", key, true, "read", namespace)
        );
        // Same answer as a namespace that does not exist, so browsing cannot
        // be used to enumerate what is out of reach.
        Err(HttpResponse::NotFound().json(Message::new("No such namespace.")))
    }
}

/// The same rule as [`require_read`], for the paths that change data: an
/// `admin, r:feeds` key browses `feeds/*` but does not add to it.
///
/// Unlike the sighting API this does not care whether `authenticate` is off.
/// That setting is about the sighting API; the management interface has always
/// demanded a key, and a key that says what it may write is the only thing
/// that makes an admin grant safe to hand out.
fn require_write(state: &SharedState, key: &str, namespace: &str) -> Result<(), HttpResponse> {
    if state.acl().can_write(key, namespace) {
        Ok(())
    } else {
        log::warn!(
            "{}",
            crate::handlers::refusal("an admin key", key, true, "write", namespace)
        );
        Err(HttpResponse::Forbidden().json(Message::new(format!(
            "That key is not permitted to write to '{namespace}'."
        ))))
    }
}

/// Tidy a namespace typed by a person into the name the database will store.
///
/// Browsing is by path segment, so stray or doubled slashes would otherwise
/// create a namespace that looks like a folder someone already made but sorts
/// and links as something else.
fn clean_namespace(namespace: &str) -> Result<String, HttpResponse> {
    let cleaned = namespace
        .split('/')
        .map(str::trim)
        .filter(|segment| !segment.is_empty())
        .collect::<Vec<_>>()
        .join("/");

    if cleaned.is_empty() {
        return Err(HttpResponse::BadRequest().json(Message::new("A namespace needs a name.")));
    }
    // The same rules as an ACL prefix, so that anything created here can also
    // be granted to a key later.
    if let Err(e) = validate_namespace(&cleaned) {
        return Err(HttpResponse::BadRequest().json(Message::new(e.to_string())));
    }
    Ok(cleaned)
}

pub async fn index() -> impl Responder {
    HttpResponse::Ok()
        .content_type("text/html; charset=utf-8")
        .body(UI)
}

pub async fn logo() -> impl Responder {
    HttpResponse::Ok()
        .content_type("image/png")
        // Immutable: it only changes when the binary does.
        .insert_header(("Cache-Control", "public, max-age=86400"))
        .body(LOGO)
}

pub async fn echarts() -> impl Responder {
    HttpResponse::Ok()
        .content_type("application/javascript; charset=utf-8")
        // Immutable: it only changes when the binary does.
        .insert_header(("Cache-Control", "public, max-age=86400"))
        .body(ECHARTS)
}

/// Confirms a key is an admin key, so the interface can show its login result.
pub async fn session(state: State, req: HttpRequest) -> HttpResponse {
    if let Err(resp) = require_admin(&state, &req) {
        return resp;
    }
    HttpResponse::Ok().json(Message::new("ok"))
}

pub async fn info(state: State, req: HttpRequest) -> HttpResponse {
    if let Err(resp) = require_admin(&state, &req) {
        return resp;
    }
    HttpResponse::Ok().json(state.info.clone())
}

pub async fn namespaces(
    state: State,
    query: web::Query<BrowseQuery>,
    req: HttpRequest,
) -> HttpResponse {
    let key = match require_admin(&state, &req) {
        Ok(key) => key.to_string(),
        Err(resp) => return resp,
    };

    let acl = state.acl();
    let page = state
        .db
        .namespace_page(&query.q, query.offset, query.limit(), |name| {
            acl.can_read(&key, name)
        });
    drop(acl);
    HttpResponse::Ok().json(page)
}

/// `GET /_management/api/galaxy` — this server and the peers it knows.
///
/// What a topology view reads. Reports this server's own role and each peer's
/// health, so the picture can be drawn from one request against whichever
/// server the operator happens to be connected to.
///
/// Peer **keys are never included**: they are credentials, and this response is
/// rendered in a browser.
///
/// Nothing is forwarded to peers yet, so this describes the configured galaxy
/// and whether its members answer — not a traversal of it. A cascade is
/// reported one level deep: each peer appears as a peer, and asking *it* for
/// its own galaxy is how the next level is reached.
pub async fn galaxy(state: State, req: HttpRequest) -> HttpResponse {
    if let Err(resp) = require_admin(&state, &req) {
        return resp;
    }

    let Some(galaxy) = state.galaxy.as_ref() else {
        return HttpResponse::Ok().json(serde_json::json!({
            "self": describe(&state),
            "peers": [],
            "below": {},
        }));
    };

    // One level per hop. Asked for only when this request was not itself a
    // walk that has run out of budget, so a cascade is described all the way
    // down and a cycle still terminates.
    let hops = crate::galaxy::hops_left(&req, galaxy.max_hops()).unwrap_or(0);
    let below: serde_json::Map<String, serde_json::Value> = galaxy
        .walk(hops)
        .await
        .into_iter()
        .collect::<std::collections::BTreeMap<_, _>>()
        .into_iter()
        .collect();

    HttpResponse::Ok().json(serde_json::json!({
        // The server answering, so a view has a root to draw from without
        // being told separately which one it asked.
        "self": describe(&state),
        "peers": galaxy.health(),
        // Each peer's own answer, keyed by its url. A peer that is itself a
        // router has peers of its own in here, which is how a cascade is
        // drawn from one request.
        "below": below,
    }))
}

/// This server, as a topology view needs it.
fn describe(state: &SharedState) -> serde_json::Value {
    serde_json::json!({
        "node_id": state.db.node(),
        "version": env!("CARGO_PKG_VERSION"),
        "role": state.info.role.clone(),
        "uptime_seconds": state.started.elapsed().as_secs(),
    })
}

/// `GET /_management/api/rejections` — values that were not written.
///
/// Newest first, because the question is nearly always "what has just started
/// failing?". Every write path feeds this, including the ZMQ ingest, which has
/// no caller of its own to tell.
///
/// Filtered by what the key may *read*: a rejection names a namespace and a
/// value someone tried to put in it, which is not something to hand to a key
/// that could not have read that namespace anyway.
pub async fn rejections(
    state: State,
    query: web::Query<RejectionsQuery>,
    req: HttpRequest,
) -> HttpResponse {
    let key = match require_admin(&state, &req) {
        Ok(key) => key.to_string(),
        Err(resp) => return resp,
    };

    if let Some(namespace) = &query.namespace
        && let Err(resp) = require_read(&state, &key, namespace)
    {
        return resp;
    }

    let acl = state.acl();
    // Over-fetch, then drop what this key may not read, so that filtering does
    // not silently return a short page.
    let mut found: Vec<_> = state
        .rejections
        .recent(query.namespace.as_deref(), MAX_LIMIT)
        .into_iter()
        .filter(|entry| acl.can_read(&key, &entry.namespace))
        .collect();
    drop(acl);
    found.truncate(query.limit());

    HttpResponse::Ok().json(RejectionPage {
        rejections: found,
        total: state.rejections.len(),
        capacity: state.rejections.capacity(),
    })
}

/// `DELETE /_management/api/rejections` — forget them all.
///
/// How an operator marks a feed as dealt with, so that what shows up next is
/// new rather than the same thousand entries they have already read.
pub async fn clear_rejections(state: State, req: HttpRequest) -> HttpResponse {
    let key = match require_admin(&state, &req) {
        Ok(key) => key.to_string(),
        Err(resp) => return resp,
    };

    // Clearing is all-or-nothing, so it takes a key that could read all of it.
    if !state.acl().can_read(&key, "") {
        return HttpResponse::Forbidden().json(Message::new(
            "Clearing the rejection log needs a key with unscoped read access.",
        ));
    }

    let held = state.rejections.len();
    state.rejections.clear();
    log::info!("{held} rejection(s) cleared by '{key}'");
    HttpResponse::Ok().json(Message::new("ok"))
}

pub async fn values(
    state: State,
    query: web::Query<ValuesQuery>,
    req: HttpRequest,
) -> HttpResponse {
    let key = match require_admin(&state, &req) {
        Ok(key) => key,
        Err(resp) => return resp,
    };
    if let Err(resp) = require_read(&state, key, &query.namespace) {
        return resp;
    }

    // Statistics are per value and can be large, so the list omits them; the
    // detail view fetches them for one value at a time.
    match state.db.value_page(
        &query.namespace,
        &query.q,
        query.offset,
        query.limit(),
        false,
    ) {
        Some(page) => HttpResponse::Ok().json(page),
        None => HttpResponse::NotFound().json(Message::new("No such namespace.")),
    }
}

/// One level of the namespace tree, so the interface can browse namespaces the
/// way a file manager browses directories.
pub async fn tree(state: State, query: web::Query<TreeQuery>, req: HttpRequest) -> HttpResponse {
    let key = match require_admin(&state, &req) {
        Ok(key) => key.to_string(),
        Err(resp) => return resp,
    };

    let acl = state.acl();
    let page =
        state
            .db
            .namespace_children(&query.path, &query.q, query.offset, query.limit(), |name| {
                acl.can_read(&key, name)
            });
    drop(acl);
    HttpResponse::Ok().json(page)
}

/// Create a namespace that holds nothing yet.
///
/// Nothing else in SightingDB needs this — writing a value brings its namespace
/// into being — but a browser wants somewhere to put things before it has them,
/// and a folder made in advance is how anyone expects that to work.
pub async fn create_namespace(
    state: State,
    body: web::Json<NewNamespace>,
    req: HttpRequest,
) -> HttpResponse {
    let caller = match require_admin(&state, &req) {
        Ok(key) => key.to_string(),
        Err(resp) => return resp,
    };

    let namespace = match clean_namespace(&body.namespace) {
        Ok(namespace) => namespace,
        Err(resp) => return resp,
    };
    if let Err(resp) = require_write(&state, &caller, &namespace) {
        return resp;
    }

    if !state.db.create_namespace(&namespace) {
        return HttpResponse::Conflict()
            .json(Message::new(format!("'{namespace}' already exists.")));
    }

    log::info!("Namespace '{namespace}' created by '{caller}'");
    HttpResponse::Ok().json(serde_json::json!({ "namespace": namespace }))
}

/// Record one value or a pasted list of them, in one namespace.
///
/// The namespace does not have to exist: writing is what creates it, here as
/// everywhere else. Values are counted towards consensus exactly as a `/w/`
/// write would be, so nothing added here is a second class of sighting.
pub async fn add_values(
    state: State,
    body: web::Json<NewValues>,
    req: HttpRequest,
) -> HttpResponse {
    let caller = match require_admin(&state, &req) {
        Ok(key) => key.to_string(),
        Err(resp) => return resp,
    };

    let body = body.into_inner();
    let namespace = match clean_namespace(&body.namespace) {
        Ok(namespace) => namespace,
        Err(resp) => return resp,
    };
    if let Err(resp) = require_write(&state, &caller, &namespace) {
        return resp;
    }

    let when = match body
        .timestamp
        .map(crate::sighting_writer::timestamp_to_instant)
    {
        Some(Ok(when)) => Some(when),
        Some(Err(e)) => return HttpResponse::BadRequest().json(Message::new(e.to_string())),
        None => None,
    };

    // Whitespace-only lines are what a paste ends with, not something someone
    // meant to record.
    let values: Vec<&str> = body
        .values
        .iter()
        .map(|value| value.trim())
        .filter(|value| !value.is_empty())
        .collect();
    if values.is_empty() {
        return HttpResponse::BadRequest().json(Message::new("No values to add."));
    }

    let mut report = WriteReport {
        namespace: namespace.clone(),
        written: 0,
        counts: Vec::with_capacity(values.len()),
        errors: Vec::new(),
    };
    for value in values {
        match crate::sighting_writer::write_tagged(
            &state.db, &namespace, value, when, body.ttl, &body.tags,
        ) {
            Ok(written) => {
                report.written += 1;
                report.counts.push(ValueCount {
                    value: value.to_string(),
                    count: written.count,
                    new: written.new,
                });
            }
            Err(e) => {
                state.rejections.record(
                    &namespace,
                    value,
                    &e.to_string(),
                    crate::rejections::Source::Management,
                );
                report.errors.push(ValueError {
                    value: value.to_string(),
                    error: e.to_string(),
                });
            }
        }
    }

    log::info!(
        "{} value(s) added to '{namespace}' by '{caller}'{}",
        report.written,
        if report.errors.is_empty() {
            String::new()
        } else {
            format!(", {} rejected", report.errors.len())
        }
    );

    // Every value failing is a client error; a mix still reports what landed,
    // the same way `/wb` does.
    if report.written == 0 {
        HttpResponse::BadRequest().json(report)
    } else {
        HttpResponse::Ok().json(report)
    }
}

/// One value with its hourly statistics, which is what the histogram draws.
pub async fn value(state: State, query: web::Query<ValueQuery>, req: HttpRequest) -> HttpResponse {
    let key = match require_admin(&state, &req) {
        Ok(key) => key,
        Err(resp) => return resp,
    };
    if let Err(resp) = require_read(&state, key, &query.namespace) {
        return resp;
    }

    let consensus = state.db.count(crate::db::ALL_NAMESPACE, &query.value);
    match state
        .db
        .view(&query.namespace, &query.value, consensus, true)
    {
        Some(view) => HttpResponse::Ok().json(view),
        None => HttpResponse::NotFound().json(Message::new("No such value.")),
    }
}

/// Every namespace one value appears in, which is what the graph draws.
///
/// Only namespaces this key may read are returned. The count in `_all` — shown
/// beside the graph as consensus — still reflects every namespace, so a scoped
/// key can tell that it is not seeing all of them without being told their
/// names.
pub async fn sightings(
    state: State,
    query: web::Query<SightingsQuery>,
    req: HttpRequest,
) -> HttpResponse {
    let key = match require_admin(&state, &req) {
        Ok(key) => key.to_string(),
        Err(resp) => return resp,
    };
    if query.value.is_empty() {
        return HttpResponse::BadRequest().json(Message::new("No value to look for."));
    }

    let acl = state.acl();
    let found = state
        .db
        .sightings_of(&query.value, query.limit(), |name| acl.can_read(&key, name));
    drop(acl);

    HttpResponse::Ok().json(serde_json::json!({
        "value": query.value,
        "consensus": state.db.count(crate::db::ALL_NAMESPACE, &query.value),
        "items": found.items,
        "truncated": found.truncated,
        "paged_in": found.paged_in,
    }))
}

/// Replace one value's tags.
///
/// Adding tags is a write like any other and goes through `/w`; this exists for
/// the other direction, since a wrong tag can only come off by replacing the
/// set. It is not a sighting: nothing is counted and no timestamp moves.
pub async fn set_tags(state: State, body: web::Json<TagChange>, req: HttpRequest) -> HttpResponse {
    let caller = match require_admin(&state, &req) {
        Ok(key) => key.to_string(),
        Err(resp) => return resp,
    };

    let change = body.into_inner();
    let namespace = match clean_namespace(&change.namespace) {
        Ok(namespace) => namespace,
        Err(resp) => return resp,
    };
    if let Err(resp) = require_write(&state, &caller, &namespace) {
        return resp;
    }

    if !state.db.set_tags(&namespace, &change.value, &change.tags) {
        return HttpResponse::NotFound().json(Message::new("No such value."));
    }

    log::info!(
        "Tags of '{}' in '{namespace}' set by '{caller}'",
        change.value
    );
    let consensus = state.db.count(crate::db::ALL_NAMESPACE, &change.value);
    match state.db.view(&namespace, &change.value, consensus, false) {
        Some(view) => HttpResponse::Ok().json(view),
        None => HttpResponse::Ok().json(Message::new("ok")),
    }
}

// ---------------------------------------------------------------------------
// Key management
// ---------------------------------------------------------------------------

/// Where the ACL is written. Without one configured, keys are read-only:
/// rewriting the daemon configuration in place is not something this does.
fn acl_file(state: &SharedState) -> Result<&PathBuf, HttpResponse> {
    state.acl_file.as_ref().ok_or_else(|| {
        HttpResponse::Conflict().json(Message::new(
            "No acl_file is configured, so keys cannot be edited here. Set acl_file in \
             [daemon] and restart.",
        ))
    })
}

/// Persist the ACL, then adopt it. Written to a temporary file and renamed, so
/// a crash mid-write cannot leave a truncated file that locks everyone out.
fn save_acl(state: &SharedState, acl: crate::acl::Acl) -> Result<(), HttpResponse> {
    let path = acl_file(state)?;

    let write = || -> std::io::Result<()> {
        if let Some(parent) = path.parent() {
            std::fs::create_dir_all(parent)?;
        }
        let temp = path.with_extension("tmp");
        std::fs::write(&temp, acl.to_toml())?;
        std::fs::rename(&temp, path)
    };

    if let Err(e) = write() {
        log::error!("Could not write {}: {e}", path.display());
        return Err(HttpResponse::InternalServerError()
            .json(Message::new(format!("Could not write the ACL file: {e}"))));
    }

    *state
        .acl
        .write()
        .unwrap_or_else(std::sync::PoisonError::into_inner) = acl;
    Ok(())
}

/// `GET /_management/api/keys/drift` — where this server's keys and its peers'
/// disagree.
///
/// Worth a view of its own because of one deliberate limitation: the periodic
/// offer of keys to peers is *additive*, so a key revoked while a peer was down
/// stays live on that peer. That is a security hole if it is invisible, and
/// merely a chore if it is not — this is what makes it visible.
///
/// Reports, per peer:
///
///   * `revoked_but_present` — keys this server **revoked** and the peer still
///     accepts. A revocation that did not land, and the reason this exists.
///   * `only_on_peer` — keys the peer has that this server never knew about.
///     Ordinary: a peer has its own keys, including the one this server
///     authenticates with. Kept separate so it does not drown the case above.
///   * `missing` — keys this server has and the peer does not. Usually a peer
///     that has not had the periodic offer yet.
///
/// The record of revocations is in memory, so it is lost on restart: after one,
/// a revocation that never landed moves from `revoked_but_present` into
/// `only_on_peer` and stops being flagged. Checking the peer's own key list is
/// then the way to find it.
///
/// Asked through each peer's own management interface, so a peer this server
/// does not administer reports why rather than appearing to agree.
pub async fn key_drift(state: State, req: HttpRequest) -> HttpResponse {
    if let Err(resp) = require_admin(&state, &req) {
        return resp;
    }

    let Some(galaxy) = state.galaxy.as_ref() else {
        return HttpResponse::Ok().json(serde_json::json!({ "peers": [] }));
    };

    let mine: std::collections::BTreeSet<String> = state
        .acl()
        .entries()
        .iter()
        .map(|(key, _)| key.clone())
        .collect();
    let revoked_here: std::collections::BTreeSet<String> =
        state.revoked_keys().into_iter().collect();

    let mut rows = Vec::new();
    for peer in galaxy.peer_keys().await {
        let theirs: std::collections::BTreeSet<String> = peer.keys.iter().cloned().collect();
        let unknown: Vec<&String> = theirs.difference(&mine).collect();

        // The dangerous case: this server revoked it and the peer still takes
        // it. Separated from the rest, because a peer having keys this server
        // never knew about is ordinary — it has its own, including the one
        // this server authenticates with, which would otherwise be reported as
        // stale for ever.
        let (stale, theirs_alone): (Vec<&String>, Vec<&String>) = unknown
            .into_iter()
            .partition(|key| revoked_here.contains(*key));

        let missing: Vec<&String> = mine.difference(&theirs).collect();

        rows.push(serde_json::json!({
            "url": peer.url,
            "readable": peer.readable,
            "error": peer.error,
            "revoked_but_present": stale,
            "only_on_peer": theirs_alone,
            "missing": missing,
            "agrees": peer.readable && stale.is_empty() && missing.is_empty(),
        }));
    }

    HttpResponse::Ok().json(serde_json::json!({ "peers": rows }))
}

/// `PUT /_management/api/keys` — hold exactly these keys and no others.
///
/// What makes a galaxy have one place to manage keys, and the only thing that
/// makes a revocation reach a server that was offline for it: the periodic
/// offer of individual keys is additive and cannot delete.
///
/// **Refused unless this server has said it may be replaced.** `acl_replaceable`
/// in `[galaxy]` is off by default, because a server quietly having its keys
/// rewritten is not a state to arrive at by accident. With it off, the offer
/// falls back to being additive and nothing is lost.
///
/// Two guards, both the same shape as the ones on a single-key change:
///
///   * the set must contain an admin key, or nobody could use the interface
///     again;
///   * it must contain the key making the request, or the server doing the
///     replacing locks itself out of the server it just took over.
///
/// Keys removed are noted as revoked, so this server's own drift view can
/// point out any peer that still holds them.
pub async fn replace_keys(
    state: State,
    body: web::Json<KeyList>,
    req: HttpRequest,
) -> HttpResponse {
    let caller = match require_admin(&state, &req) {
        Ok(key) => key.to_string(),
        Err(resp) => return resp,
    };

    if !state.acl_replaceable {
        return HttpResponse::Forbidden().json(Message::new(
            "This server does not let another replace its keys. Set acl_replaceable              in [galaxy] if it should.",
        ));
    }

    let wanted = body.into_inner().keys;
    for entry in &wanted {
        if let Err(e) = entry.validate() {
            return HttpResponse::BadRequest().json(Message::new(e.to_string()));
        }
    }

    let mut acl = crate::acl::Acl::new();
    for entry in &wanted {
        acl.set(&entry.key, entry.to_grants());
    }

    if acl.admin_count() == 0 {
        return HttpResponse::Conflict().json(Message::new(
            "That set has no admin key, and would lock everyone out of this interface.",
        ));
    }
    if !acl.contains(&caller) {
        return HttpResponse::Conflict().json(Message::new(
            "That set does not include the key making this request, which would lock              the caller out of the server it is replacing.",
        ));
    }

    // What is going away, before it does, so the drift view can say which
    // peers still take it.
    let going: Vec<String> = state
        .acl()
        .entries()
        .iter()
        .map(|(key, _)| key.clone())
        .filter(|key| !acl.contains(key))
        .collect();

    if let Err(resp) = save_acl(&state, acl) {
        return resp;
    }
    for key in &going {
        state.note_revoked(key);
    }
    for entry in &wanted {
        state.note_unrevoked(&entry.key);
    }

    log::warn!(
        "Key list replaced by '{caller}': {} key(s) held, {} revoked",
        wanted.len(),
        going.len()
    );

    HttpResponse::Ok().json(serde_json::json!({
        "held": wanted.len(),
        "revoked": going,
    }))
}

/// A whole key list, for [`replace_keys`].
#[derive(Debug, Deserialize)]
pub struct KeyList {
    pub keys: Vec<KeyEntry>,
}

pub async fn list_keys(state: State, req: HttpRequest) -> HttpResponse {
    if let Err(resp) = require_admin(&state, &req) {
        return resp;
    }
    let entries: Vec<KeyEntry> = state
        .acl()
        .entries()
        .iter()
        .map(|(key, grants)| KeyEntry::from_grants(key, grants))
        .collect();
    HttpResponse::Ok().json(entries)
}

/// Create or replace one key.
pub async fn save_key(state: State, body: web::Json<KeyEntry>, req: HttpRequest) -> HttpResponse {
    let caller = match require_admin(&state, &req) {
        Ok(key) => key.to_string(),
        Err(resp) => return resp,
    };

    let entry = body.into_inner();
    if let Err(e) = entry.validate() {
        return HttpResponse::BadRequest().json(Message::new(e.to_string()));
    }

    let mut acl = state.acl().clone();
    let was_admin = acl.is_admin(&entry.key);
    acl.set(&entry.key, entry.to_grants());

    // Removing the admin grant from the last admin would lock everyone out of
    // the interface with no way back in short of editing the file by hand.
    if was_admin && !entry.admin && acl.admin_count() == 0 {
        return HttpResponse::Conflict().json(Message::new(
            "That would leave no admin key. Grant admin to another key first.",
        ));
    }

    if let Err(resp) = save_acl(&state, acl) {
        return resp;
    }
    log::info!("Key '{}' saved by '{caller}'", entry.key);
    // A key put back is not a stale revocation any more.
    state.note_unrevoked(&entry.key);
    gossip_key(&state, &req, &entry).await;
    HttpResponse::Ok().json(entry)
}

/// Pass a key change on to the peers this server administers.
///
/// After the change has been saved here, so a peer never hears about something
/// this server then failed to keep. Best effort: a peer that is down misses it
/// and picks it up from the periodic offer — see [`crate::galaxy::gossip`].
///
/// A change that *arrived* as gossip is passed along with one hop less, so a
/// cascade propagates and a cycle dies on the budget. A change made by a person
/// here starts with the full budget.
async fn gossip_key(state: &State, req: &HttpRequest, entry: &KeyEntry) {
    let Some(galaxy) = state.galaxy.as_ref() else {
        return;
    };
    let Some(hops) = crate::galaxy::hops_left(req, galaxy.max_hops()) else {
        return;
    };
    let report = galaxy.push_key(&serde_json::json!(entry), hops).await;
    if report.accepted > 0 || report.failed > 0 {
        log::info!(
            "Key '{}' passed to {} peer(s); {} could not take it, {} do not let this \
             server administer them",
            entry.key,
            report.accepted,
            report.failed,
            report.refused
        );
    }
}

pub async fn delete_key(state: State, path: web::Path<String>, req: HttpRequest) -> HttpResponse {
    let caller = match require_admin(&state, &req) {
        Ok(key) => key.to_string(),
        Err(resp) => return resp,
    };
    let key = path.into_inner();

    let mut acl = state.acl().clone();
    if !acl.contains(&key) {
        return HttpResponse::NotFound().json(Message::new("No such key."));
    }
    acl.remove(&key);

    if acl.admin_count() == 0 {
        return HttpResponse::Conflict().json(Message::new(
            "That would revoke the last admin key. Grant admin to another key first.",
        ));
    }

    if let Err(resp) = save_acl(&state, acl) {
        return resp;
    }
    log::warn!("Key '{key}' revoked by '{caller}'");
    // Noted so the interface can point out a peer that still holds it: the
    // periodic offer never deletes, so a revocation missed while a peer was
    // down is the one thing here that is genuinely dangerous.
    state.note_revoked(&key);

    if let Some(galaxy) = state.galaxy.as_ref()
        && let Some(hops) = crate::galaxy::hops_left(&req, galaxy.max_hops())
    {
        let report = galaxy.push_key_removal(&key, hops).await;
        if report.accepted > 0 || report.failed > 0 {
            log::warn!(
                "Revocation of '{key}' passed to {} peer(s); {} could not take it, {} do \
                 not let this server administer them",
                report.accepted,
                report.failed,
                report.refused
            );
        }
    }

    HttpResponse::Ok().json(Message::new("ok"))
}

/// Suggest a strong key, so nobody has to invent one.
pub async fn generate_key(state: State, req: HttpRequest) -> HttpResponse {
    if let Err(resp) = require_admin(&state, &req) {
        return resp;
    }
    HttpResponse::Ok().json(serde_json::json!({ "key": random_key() }))
}

fn random_key() -> String {
    use rand::RngExt;
    const ALPHABET: &[u8] = b"abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789";
    let mut rng = rand::rng();
    (0..40)
        .map(|_| ALPHABET[rng.random_range(0..ALPHABET.len())] as char)
        .collect()
}

/// Change a shard's tier and write it back, so it survives a restart.
/// Set, change or clear the storage settings of the shard a namespace is in.
///
/// The tier decides whether the shard is kept in memory, and `warm_idle` how
/// long a warm one waits before being written out — the second being the thing
/// that could only be changed in the configuration file until now.
///
/// It applies to the whole top-level namespace, and the reply says so: a change
/// made from a row deep in a tree is a change to everything beside it, which is
/// better said than discovered.
pub async fn set_tier(state: State, body: web::Json<TierChange>, req: HttpRequest) -> HttpResponse {
    let caller = match require_admin(&state, &req) {
        Ok(key) => key.to_string(),
        Err(resp) => return resp,
    };

    let change = body.into_inner();
    let namespace = change.namespace.trim().trim_matches('/');
    if namespace.is_empty() {
        return HttpResponse::BadRequest().json(Message::new("Name the namespace to change."));
    }
    if namespace.starts_with('_') {
        return HttpResponse::BadRequest().json(Message::new(
            "Internal namespaces are always hot and cannot be retiered.",
        ));
    }
    let entry = match change.entry() {
        Ok(entry) => entry,
        Err(e) => return HttpResponse::BadRequest().json(Message::new(e)),
    };

    // Changing what a namespace costs to keep is a change to it.
    if let Err(resp) = require_write(&state, &caller, namespace) {
        return resp;
    }

    let Some(path) = state.tiers_file.as_ref() else {
        return HttpResponse::Conflict().json(Message::new(
            "No tiers_file is configured, so tiers cannot be changed here. Set tiers_file in \
             [storage] and restart.",
        ));
    };

    let shard = crate::persistence::shard_of(namespace).to_string();
    state.db.set_policy(&shard, entry);
    if let Err(e) = state.db.tier_policy().save(path) {
        log::error!("Could not write {}: {e:#}", path.display());
        return HttpResponse::InternalServerError()
            .json(Message::new(format!("Could not write the tier file: {e}")));
    }

    let resolved = state.db.resolved_policy(&shard);
    log::info!(
        "Storage of '{shard}' set to {} (warm_idle {}s) by '{caller}'",
        resolved.tier.as_str(),
        resolved.warm_idle,
    );

    // Passed on to the peers this server administers, under the same rule as a
    // key change: the peer's own ACL decides whether to accept it. A tier only
    // ever takes a value — there is no unset — so this needs no deletion to
    // chase, which is the one thing that made keys awkward.
    if let Some(galaxy) = state.galaxy.as_ref()
        && let Some(hops) = crate::galaxy::hops_left(&req, galaxy.max_hops())
    {
        let report = galaxy
            .push_tier(
                &serde_json::json!({
                    "namespace": shard,
                    "tier": resolved.tier.as_str(),
                    "warm_idle": resolved.warm_idle,
                }),
                hops,
            )
            .await;
        if report.accepted > 0 || report.failed > 0 {
            log::info!(
                "Storage of '{shard}' passed to {} peer(s); {} could not take it, {} do \
                 not let this server administer them",
                report.accepted,
                report.failed,
                report.refused
            );
        }
    }

    HttpResponse::Ok().json(serde_json::json!({
        "shard": shard,
        "tier": resolved.tier.as_str(),
        "warm_idle": resolved.warm_idle,
        "own_tier": resolved.own_tier.is_some(),
        "own_warm_idle": resolved.own_warm_idle.is_some(),
        "effect": state.db.shard_effect(&shard),
    }))
}

/// Register the admin routes.
pub fn routes(cfg: &mut web::ServiceConfig) {
    // Order matters: the catch-all must come last or it would swallow the API.
    cfg.route("/_management/echarts.min.js", web::get().to(echarts))
        .route("/_management/logo.png", web::get().to(logo))
        .route("/_management/api/session", web::get().to(session))
        .route("/_management/api/info", web::get().to(info))
        .route("/_management/api/namespaces", web::get().to(namespaces))
        .route(
            "/_management/api/namespaces",
            web::post().to(create_namespace),
        )
        .route("/_management/api/tree", web::get().to(tree))
        .route("/_management/api/galaxy", web::get().to(galaxy))
        .route("/_management/api/rejections", web::get().to(rejections))
        .route(
            "/_management/api/rejections",
            web::delete().to(clear_rejections),
        )
        .route("/_management/api/values", web::get().to(values))
        .route("/_management/api/values", web::post().to(add_values))
        .route("/_management/api/tags", web::post().to(set_tags))
        .route("/_management/api/value", web::get().to(value))
        .route("/_management/api/sightings", web::get().to(sightings))
        .route("/_management/api/keys", web::get().to(list_keys))
        .route("/_management/api/keys", web::post().to(save_key))
        .route("/_management/api/keys", web::put().to(replace_keys))
        .route(
            "/_management/api/keys/generate",
            web::get().to(generate_key),
        )
        // Before the `{key}` route, or "drift" would be taken for a key name.
        .route("/_management/api/keys/drift", web::get().to(key_drift))
        .route("/_management/api/keys/{key}", web::delete().to(delete_key))
        .route("/_management/api/tier", web::post().to(set_tier))
        .route("/_management", web::get().to(index))
        .route("/_management/{namespace:.*}", web::get().to(index));
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::acl::parse_grants;
    use crate::db::WriteOpts;
    use actix_web::http::StatusCode;
    use actix_web::{App, test};
    use chrono::Utc;
    use serde_json::{Value as Json, json};

    const ADMIN: &str = "changeme";
    /// Spelled out so `None` has a type at the call sites.
    const NO_KEY: Option<&str> = None;

    fn state() -> State {
        let mut inner = SharedState::new(false);
        inner.acl.get_mut().unwrap().grant_full(ADMIN);
        inner
            .acl
            .get_mut()
            .unwrap()
            .set("plain", parse_grants("rw").unwrap());

        for value in ["1.2.3.4", "5.6.7.8", "9.9.9.9"] {
            inner.db.write(
                "feeds/ips",
                value,
                Utc::now(),
                WriteOpts {
                    consensus: true,
                    ttl: None,
                },
            );
        }
        web::Data::new(inner)
    }

    macro_rules! app {
        ($state:expr) => {
            test::init_service(App::new().app_data($state.clone()).configure(routes)).await
        };
    }

    /// The management routes *and* the data routes, for tests that drive a
    /// write and then ask the management interface what it made of it.
    macro_rules! app_with_data {
        ($state:expr) => {
            test::init_service(
                App::new()
                    .app_data($state.clone())
                    .configure(routes)
                    .configure(crate::handlers::routes),
            )
            .await
        };
    }

    /// A GET with an optional key; a macro rather than a function so the
    /// service type does not have to be named.
    macro_rules! get {
        ($app:expr, $uri:expr, $key:expr $(,)?) => {{
            let mut req = test::TestRequest::get().uri($uri);
            if let Some(key) = $key {
                req = req.insert_header(("Authorization", key));
            }
            test::call_service(&$app, req.to_request()).await
        }};
    }

    #[actix_web::test]
    async fn the_interface_is_served_without_a_key() {
        let st = state();
        let app = app!(st);

        // The page itself is just markup; everything it shows needs the key.
        let resp = get!(app, "/_management/", NO_KEY);
        assert_eq!(resp.status(), StatusCode::OK);
        let body = test::read_body(resp).await;
        assert!(String::from_utf8_lossy(&body).contains("<title>"));
    }

    #[actix_web::test]
    async fn the_data_endpoints_need_an_admin_key() {
        let st = state();
        let app = app!(st);

        for uri in [
            "/_management/api/info",
            "/_management/api/namespaces",
            "/_management/api/values?namespace=feeds/ips",
            "/_management/api/value?namespace=feeds/ips&value=1.2.3.4",
            "/_management/api/session",
        ] {
            assert_eq!(
                get!(app, uri, NO_KEY).status(),
                StatusCode::UNAUTHORIZED,
                "{uri}"
            );
            // A valid key that is not an admin key is not enough either.
            assert_eq!(
                get!(app, uri, Some("plain")).status(),
                StatusCode::FORBIDDEN,
                "{uri}"
            );
            assert_eq!(
                get!(app, uri, Some(ADMIN)).status(),
                StatusCode::OK,
                "{uri}"
            );
        }
    }

    /// A paste reports what each value now stands at, so the interface does not
    /// have to read the values back to say what it just did.
    #[actix_web::test]
    async fn adding_values_reports_each_running_count() {
        let st = state();
        let app = app!(st);

        // `1.2.3.4` is already in `feeds/ips` once, from `state()`.
        let resp = test::call_service(
            &app,
            test::TestRequest::post()
                .uri("/_management/api/values")
                .insert_header(("Authorization", ADMIN))
                .set_json(serde_json::json!({
                    "namespace": "feeds/ips",
                    "values": ["1.2.3.4", "fresh", "   "],
                }))
                .to_request(),
        )
        .await;

        assert_eq!(resp.status(), StatusCode::OK);
        let body: serde_json::Value = test::read_body_json(resp).await;

        // The blank line was dropped before anything was tried.
        assert_eq!(body["written"], 2, "{body}");

        let counts = body["counts"].as_array().unwrap();
        assert_eq!(counts.len(), 2, "{body}");
        assert_eq!(counts[0]["value"], "1.2.3.4");
        assert_eq!(counts[0]["count"], 2, "an existing value was not added to");
        assert_eq!(counts[1]["value"], "fresh");
        assert_eq!(counts[1]["count"], 1, "{body}");

        // And the counts are what the database actually holds.
        assert_eq!(st.db.count("feeds/ips", "1.2.3.4"), 2);
        assert_eq!(st.db.count("feeds/ips", "fresh"), 1);
    }

    /// Every write path feeds the rejection log, and the log is what answers
    /// "which values errored" after the response has gone.
    #[actix_web::test]
    async fn rejections_are_listed_newest_first() {
        let st = state();
        let app = app_with_data!(st);

        // A rejection from the single-value route, and two from a bulk write.
        test::call_service(
            &app,
            test::TestRequest::get()
                .uri("/w/feeds/ips?val=x&timestamp=99999999999999")
                .to_request(),
        )
        .await;
        test::call_service(
            &app,
            test::TestRequest::post()
                .uri("/wb")
                .set_json(serde_json::json!({"items": [
                    {"namespace": "feeds/ips", "value": ""},
                    {"namespace": "_config/acl/apikeys/mine", "value": "sneaky"}
                ]}))
                .to_request(),
        )
        .await;

        let resp = get!(app, "/_management/api/rejections", Some(ADMIN));
        assert_eq!(resp.status(), StatusCode::OK);
        let body: serde_json::Value = test::read_body_json(resp).await;

        let found = body["rejections"].as_array().unwrap();
        assert_eq!(found.len(), 3, "{body}");
        assert_eq!(body["total"], 3, "{body}");
        assert_eq!(
            body["capacity"],
            crate::rejections::DEFAULT_CAPACITY,
            "{body}"
        );

        // Newest first.
        assert_eq!(found[0]["value"], "sneaky", "{body}");
        assert_eq!(found[0]["source"], "bulkwrite", "{body}");
        assert!(
            found[0]["reason"].as_str().unwrap().contains("_config"),
            "{body}"
        );

        // The empty value is kept as itself, which is the whole reason this is
        // not a namespace of sightings.
        assert_eq!(found[1]["value"], "", "{body}");

        assert_eq!(found[2]["value"], "x", "{body}");
        assert_eq!(found[2]["source"], "write", "{body}");
        assert!(found[2]["when"].as_i64().unwrap() > 0, "{body}");
    }

    /// A dry run writes nothing, so it must leave no rejections behind either
    /// — otherwise checking a batch would pollute the record of real failures,
    /// and anyone could flood it without writing a thing.
    #[actix_web::test]
    async fn a_dry_run_records_no_rejections() {
        let st = state();
        let app = app_with_data!(st);

        let resp = test::call_service(
            &app,
            test::TestRequest::post()
                .uri("/vwb")
                .set_json(serde_json::json!({"items": [
                    {"namespace": "feeds/ips", "value": ""}
                ]}))
                .to_request(),
        )
        .await;
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);

        assert_eq!(
            st.rejections.len(),
            0,
            "the dry run left a rejection behind"
        );
    }

    /// The management add path guards its namespace before it writes, so it
    /// cannot currently reach a rejection at all.
    ///
    /// Pinned because that is load-bearing for the claim above it: the
    /// recording wired into `add_values` is unreachable today and kept only so
    /// that loosening the guard does not silently lose the value. If this test
    /// starts failing, that path has become live and wants a test of its own.
    #[actix_web::test]
    async fn the_management_add_path_rejects_before_it_writes() {
        let st = state();
        let app = app!(st);

        let resp = test::call_service(
            &app,
            test::TestRequest::post()
                .uri("/_management/api/values")
                .insert_header(("Authorization", ADMIN))
                .set_json(serde_json::json!({
                    "namespace": "_config/acl/apikeys/mine",
                    "values": ["sneaky"],
                }))
                .to_request(),
        )
        .await;

        // Turned away by `clean_namespace`, not by the writer.
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
        assert_eq!(st.rejections.len(), 0, "a rejection was recorded after all");
        assert!(!st.db.namespace_exists("_config/acl/apikeys/mine"));
    }

    /// A rejection names a namespace and what someone tried to put in it, so a
    /// key that could not read that namespace must not see it here.
    #[actix_web::test]
    async fn rejections_are_filtered_by_what_the_key_may_read() {
        let st = state();
        st.rejections
            .record("feeds/ips", "a", "nope", crate::rejections::Source::Ingest);
        st.rejections
            .record("secrets", "b", "nope", crate::rejections::Source::Ingest);

        // An admin key scoped to `feeds` only.
        st.acl
            .write()
            .unwrap()
            .set("scoped", parse_grants("admin, rw:feeds").unwrap());
        let app = app!(st);

        let resp = get!(app, "/_management/api/rejections", Some("scoped"));
        assert_eq!(resp.status(), StatusCode::OK);
        let body: serde_json::Value = test::read_body_json(resp).await;
        let found = body["rejections"].as_array().unwrap();

        assert_eq!(found.len(), 1, "a scoped key saw another subtree: {body}");
        assert_eq!(found[0]["namespace"], "feeds/ips", "{body}");

        // The full-access key sees both.
        let resp = get!(app, "/_management/api/rejections", Some(ADMIN));
        let body: serde_json::Value = test::read_body_json(resp).await;
        assert_eq!(body["rejections"].as_array().unwrap().len(), 2, "{body}");
    }

    /// Asking for one subtree must be authorized like any other read of it —
    /// including the answer, which is `404` rather than `403` throughout the
    /// management interface so that browsing cannot enumerate what is out of
    /// reach. See [`require_read`].
    #[actix_web::test]
    async fn filtering_by_a_namespace_needs_read_access_to_it() {
        let st = state();
        st.acl
            .write()
            .unwrap()
            .set("scoped", parse_grants("admin, rw:feeds").unwrap());
        let app = app!(st);

        let resp = get!(
            app,
            "/_management/api/rejections?namespace=secrets",
            Some("scoped")
        );
        assert_eq!(resp.status(), StatusCode::NOT_FOUND);
    }

    #[actix_web::test]
    async fn rejections_can_be_cleared() {
        let st = state();
        st.rejections
            .record("feeds/ips", "a", "nope", crate::rejections::Source::Ingest);
        st.acl
            .write()
            .unwrap()
            .set("scoped", parse_grants("admin, rw:feeds").unwrap());
        let app = app!(st);

        // A scoped key cannot clear what it cannot wholly see.
        let resp = test::call_service(
            &app,
            test::TestRequest::delete()
                .uri("/_management/api/rejections")
                .insert_header(("Authorization", "scoped"))
                .to_request(),
        )
        .await;
        assert_eq!(resp.status(), StatusCode::FORBIDDEN);
        assert_eq!(st.rejections.len(), 1);

        let resp = test::call_service(
            &app,
            test::TestRequest::delete()
                .uri("/_management/api/rejections")
                .insert_header(("Authorization", ADMIN))
                .to_request(),
        )
        .await;
        assert_eq!(resp.status(), StatusCode::OK);
        assert_eq!(st.rejections.len(), 0);
    }

    /// A server that has not opted in keeps its own keys, whoever asks.
    ///
    /// The default, because a server quietly having its key list rewritten is
    /// not a state to arrive at by accident.
    #[actix_web::test]
    async fn keys_cannot_be_replaced_unless_the_server_allows_it() {
        let st = state();
        let app = app!(st);

        let resp = test::call_service(
            &app,
            test::TestRequest::put()
                .uri("/_management/api/keys")
                .insert_header(("Authorization", ADMIN))
                .set_json(serde_json::json!({"keys": [
                    {"key": ADMIN, "admin": true, "read": [""], "write": [""]}
                ]}))
                .to_request(),
        )
        .await;

        assert_eq!(resp.status(), StatusCode::FORBIDDEN);
        let body: serde_json::Value = test::read_body_json(resp).await;
        assert!(
            body["message"]
                .as_str()
                .unwrap()
                .contains("acl_replaceable"),
            "the refusal does not say how to allow it: {body}"
        );
    }

    /// With it allowed, a replace removes what is not in the set — which is
    /// the whole reason it exists, since offering keys one at a time cannot
    /// delete and so cannot carry a revocation to a server that was offline.
    #[actix_web::test]
    async fn an_allowed_replace_removes_what_is_not_offered() {
        let mut inner = SharedState::new(false);
        inner.acl.get_mut().unwrap().grant_full(ADMIN);
        inner
            .acl
            .get_mut()
            .unwrap()
            .set("doomed", parse_grants("rw:feeds").unwrap());
        let dir = TempDir::new("replace");
        inner.acl_file = Some(dir.0.join("acl.toml"));
        inner.acl_replaceable = true;
        let st: State = web::Data::new(inner);
        let app = app!(st);

        let resp = test::call_service(
            &app,
            test::TestRequest::put()
                .uri("/_management/api/keys")
                .insert_header(("Authorization", ADMIN))
                .set_json(serde_json::json!({"keys": [
                    {"key": ADMIN, "admin": true, "read": [""], "write": [""]},
                    {"key": "kept", "admin": false, "read": ["feeds"], "write": []}
                ]}))
                .to_request(),
        )
        .await;

        assert_eq!(resp.status(), StatusCode::OK);
        let body: serde_json::Value = test::read_body_json(resp).await;
        assert_eq!(body["held"], 2, "{body}");
        assert_eq!(body["revoked"][0], "doomed", "{body}");

        let acl = st.acl();
        assert!(acl.contains("kept"));
        assert!(!acl.contains("doomed"), "the revocation did not land");
        drop(acl);

        // Noted, so this server's drift view can point out a peer that still
        // takes it.
        assert!(st.revoked_keys().contains(&"doomed".to_string()));
    }

    /// A replace that would lock the interface or the caller out is refused,
    /// the same way a single-key change is.
    #[actix_web::test]
    async fn a_replace_cannot_lock_anyone_out() {
        let dir = TempDir::new("lockout");
        let fresh = || {
            let mut inner = SharedState::new(false);
            inner.acl.get_mut().unwrap().grant_full(ADMIN);
            inner.acl_file = Some(dir.0.join("acl.toml"));
            inner.acl_replaceable = true;
            web::Data::new(inner) as State
        };

        // No admin at all.
        let st = fresh();
        let app = app!(st);
        let resp = test::call_service(
            &app,
            test::TestRequest::put()
                .uri("/_management/api/keys")
                .insert_header(("Authorization", ADMIN))
                .set_json(serde_json::json!({"keys": [
                    {"key": "reader", "admin": false, "read": ["feeds"], "write": []}
                ]}))
                .to_request(),
        )
        .await;
        assert_eq!(resp.status(), StatusCode::CONFLICT);
        assert!(st.acl().contains(ADMIN), "the ACL was replaced anyway");

        // An admin, but not the caller: the server replacing it would lose its
        // own way in.
        let st = fresh();
        let app = app!(st);
        let resp = test::call_service(
            &app,
            test::TestRequest::put()
                .uri("/_management/api/keys")
                .insert_header(("Authorization", ADMIN))
                .set_json(serde_json::json!({"keys": [
                    {"key": "someone-else", "admin": true, "read": [""], "write": [""]}
                ]}))
                .to_request(),
        )
        .await;
        assert_eq!(resp.status(), StatusCode::CONFLICT);
        let body: serde_json::Value = test::read_body_json(resp).await;
        assert!(
            body["message"]
                .as_str()
                .unwrap()
                .contains("making this request"),
            "{body}"
        );
        assert!(st.acl().contains(ADMIN));
    }

    #[actix_web::test]
    async fn namespaces_are_paged_and_filtered() {
        let st = state();
        let app = app!(st);

        let resp = get!(app, "/_management/api/namespaces?limit=1", Some(ADMIN));
        let body: Json = test::read_body_json(resp).await;
        assert_eq!(body["items"].as_array().unwrap().len(), 1);
        assert_eq!(body["total"], 1);

        let resp = get!(app, "/_management/api/namespaces?q=nomatch", Some(ADMIN));
        let body: Json = test::read_body_json(resp).await;
        assert_eq!(body["total"], 0);
    }

    #[actix_web::test]
    async fn values_are_paged_and_omit_statistics() {
        let st = state();
        let app = app!(st);

        let resp = get!(
            app,
            "/_management/api/values?namespace=feeds/ips&offset=1&limit=1",
            Some(ADMIN),
        );
        let body: Json = test::read_body_json(resp).await;

        assert_eq!(body["total"], 3);
        assert_eq!(body["offset"], 1);
        let items = body["items"].as_array().unwrap();
        assert_eq!(items.len(), 1);
        // The list would be enormous with per-value histograms in it.
        assert!(items[0].get("stats").is_none(), "{items:?}");
    }

    #[actix_web::test]
    async fn a_single_value_carries_its_histogram() {
        let st = state();
        let app = app!(st);

        let resp = get!(
            app,
            "/_management/api/value?namespace=feeds/ips&value=1.2.3.4",
            Some(ADMIN),
        );
        let body: Json = test::read_body_json(resp).await;

        assert_eq!(body["value"], "1.2.3.4");
        assert!(body["stats"].is_object(), "{body}");
    }

    #[actix_web::test]
    async fn missing_things_are_404_not_500() {
        let st = state();
        let app = app!(st);

        assert_eq!(
            get!(app, "/_management/api/values?namespace=nope", Some(ADMIN)).status(),
            StatusCode::NOT_FOUND
        );
        assert_eq!(
            get!(
                app,
                "/_management/api/value?namespace=feeds/ips&value=nope",
                Some(ADMIN)
            )
            .status(),
            StatusCode::NOT_FOUND
        );
    }

    #[actix_web::test]
    async fn the_page_size_is_capped() {
        let st = state();
        let app = app!(st);

        let resp = get!(
            app,
            "/_management/api/values?namespace=feeds/ips&limit=100000",
            Some(ADMIN),
        );
        assert_eq!(resp.status(), StatusCode::OK);
        // Capped rather than refused, so a careless caller still gets an answer.
        let body: Json = test::read_body_json(resp).await;
        assert_eq!(body["items"].as_array().unwrap().len(), 3);
    }

    // -- key management ----------------------------------------------------

    /// State with a real acl_file, so saves actually hit the disk.
    fn writable_state(dir: &std::path::Path) -> State {
        let mut inner = SharedState::new(true);
        inner.acl.get_mut().unwrap().grant_full(ADMIN);
        inner.acl_file = Some(dir.join("acl.toml"));
        web::Data::new(inner)
    }

    struct TempDir(std::path::PathBuf);
    impl TempDir {
        fn new(tag: &str) -> Self {
            let path = std::env::temp_dir().join(format!("sightingdb-keys-{tag}"));
            let _ = std::fs::remove_dir_all(&path);
            std::fs::create_dir_all(&path).unwrap();
            TempDir(path)
        }
    }
    impl Drop for TempDir {
        fn drop(&mut self) {
            let _ = std::fs::remove_dir_all(&self.0);
        }
    }

    macro_rules! post_key {
        ($app:expr, $body:expr, $key:expr) => {
            test::call_service(
                &$app,
                test::TestRequest::post()
                    .uri("/_management/api/keys")
                    .insert_header(("Authorization", $key))
                    .set_json($body)
                    .to_request(),
            )
            .await
        };
    }

    #[actix_web::test]
    async fn a_saved_key_lands_on_disk_and_takes_effect_at_once() {
        let dir = TempDir::new("save");
        let st = writable_state(&dir.0);
        let app = app!(st);

        let resp = post_key!(
            app,
            json!({"key": "analyst", "admin": false, "read": ["feeds"], "write": []}),
            ADMIN
        );
        assert_eq!(resp.status(), StatusCode::OK);

        // In the running ACL, with no restart.
        assert!(st.acl().can_read("analyst", "feeds/misp"));
        assert!(!st.acl().can_write("analyst", "feeds/misp"));
        assert!(!st.acl().can_read("analyst", "secrets"));

        // And on disk, in a form that reads back.
        let written = std::fs::read_to_string(dir.0.join("acl.toml")).unwrap();
        assert!(written.contains("\"analyst\" = \"r:feeds\""), "{written}");
        assert!(written.contains("\"changeme\""), "{written}");
    }

    #[actix_web::test]
    async fn read_and_write_on_one_prefix_collapse_to_rw() {
        let dir = TempDir::new("rw");
        let st = writable_state(&dir.0);
        let app = app!(st);

        post_key!(
            app,
            json!({"key": "feed", "admin": false, "read": ["feeds"], "write": ["feeds"]}),
            ADMIN
        );

        let written = std::fs::read_to_string(dir.0.join("acl.toml")).unwrap();
        assert!(written.contains("\"feed\" = \"rw:feeds\""), "{written}");
    }

    #[actix_web::test]
    async fn a_key_can_be_revoked() {
        let dir = TempDir::new("revoke");
        let st = writable_state(&dir.0);
        let app = app!(st);
        post_key!(
            app,
            json!({"key": "temp", "admin": false, "read": [""], "write": []}),
            ADMIN
        );
        assert!(st.acl().contains("temp"));

        let resp = test::call_service(
            &app,
            test::TestRequest::delete()
                .uri("/_management/api/keys/temp")
                .insert_header(("Authorization", ADMIN))
                .to_request(),
        )
        .await;

        assert_eq!(resp.status(), StatusCode::OK);
        assert!(!st.acl().contains("temp"));
        assert!(
            !std::fs::read_to_string(dir.0.join("acl.toml"))
                .unwrap()
                .contains("temp")
        );
    }

    /// Locking every admin out would leave no way back in short of editing the
    /// file by hand, so both routes to it are refused.
    #[actix_web::test]
    async fn the_last_admin_cannot_be_removed() {
        let dir = TempDir::new("lastadmin");
        let st = writable_state(&dir.0);
        let app = app!(st);

        let demote = post_key!(
            app,
            json!({"key": ADMIN, "admin": false, "read": [""], "write": [""]}),
            ADMIN
        );
        assert_eq!(demote.status(), StatusCode::CONFLICT);

        let revoke = test::call_service(
            &app,
            test::TestRequest::delete()
                .uri(&format!("/_management/api/keys/{ADMIN}"))
                .insert_header(("Authorization", ADMIN))
                .to_request(),
        )
        .await;
        assert_eq!(revoke.status(), StatusCode::CONFLICT);

        // Still an admin, on disk and in memory.
        assert!(st.acl().is_admin(ADMIN));
    }

    #[actix_web::test]
    async fn demoting_an_admin_is_fine_once_another_exists() {
        let dir = TempDir::new("secondadmin");
        let st = writable_state(&dir.0);
        let app = app!(st);

        post_key!(
            app,
            json!({"key": "other", "admin": true, "read": [""], "write": [""]}),
            ADMIN
        );
        let resp = post_key!(
            app,
            json!({"key": ADMIN, "admin": false, "read": [""], "write": [""]}),
            ADMIN
        );

        assert_eq!(resp.status(), StatusCode::OK);
        assert!(!st.acl().is_admin(ADMIN));
        assert!(st.acl().is_admin("other"));
    }

    /// A key carrying `"` or a newline could rewrite the file as something
    /// else, so it never reaches the disk.
    #[actix_web::test]
    async fn keys_that_would_corrupt_the_file_are_refused() {
        let dir = TempDir::new("badkey");
        let st = writable_state(&dir.0);
        let app = app!(st);

        for bad in ["", "has space", "has\"quote", "has\nnewline", "has=equals"] {
            let resp = post_key!(
                app,
                json!({"key": bad, "admin": false, "read": [""], "write": []}),
                ADMIN
            );
            assert_eq!(resp.status(), StatusCode::BAD_REQUEST, "{bad:?}");
        }
    }

    #[actix_web::test]
    async fn granting_an_internal_namespace_is_refused() {
        let dir = TempDir::new("internal");
        let st = writable_state(&dir.0);
        let app = app!(st);

        let resp = post_key!(
            app,
            json!({"key": "sneaky", "admin": false, "read": ["_config/acl"], "write": []}),
            ADMIN
        );
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
        assert!(!st.acl().contains("sneaky"));
    }

    #[actix_web::test]
    async fn a_key_with_no_grants_is_refused() {
        let dir = TempDir::new("nogrants");
        let st = writable_state(&dir.0);
        let app = app!(st);

        let resp = post_key!(
            app,
            json!({"key": "useless", "admin": false, "read": [], "write": []}),
            ADMIN
        );
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
    }

    /// Without an acl_file there is nowhere to write, and rewriting the daemon
    /// configuration in place is not something this does.
    #[actix_web::test]
    async fn editing_without_an_acl_file_says_so() {
        let st = state(); // no acl_file
        let app = app!(st);

        let resp = post_key!(
            app,
            json!({"key": "x", "admin": false, "read": [""], "write": []}),
            ADMIN
        );
        assert_eq!(resp.status(), StatusCode::CONFLICT);
        let body: Json = test::read_body_json(resp).await;
        assert!(
            body["message"].as_str().unwrap().contains("acl_file"),
            "{body}"
        );
    }

    #[actix_web::test]
    async fn keys_are_listed_and_generated() {
        let dir = TempDir::new("list");
        let st = writable_state(&dir.0);
        let app = app!(st);

        let listed: Json =
            test::read_body_json(get!(app, "/_management/api/keys", Some(ADMIN))).await;
        let entries = listed.as_array().unwrap();
        assert_eq!(entries.len(), 1);
        assert_eq!(entries[0]["key"], ADMIN);
        assert_eq!(entries[0]["admin"], true);

        let generated: Json =
            test::read_body_json(get!(app, "/_management/api/keys/generate", Some(ADMIN))).await;
        let suggestion = generated["key"].as_str().unwrap();
        assert_eq!(suggestion.len(), 40);
        assert!(crate::acl::validate_key(suggestion).is_ok());
    }

    #[actix_web::test]
    async fn key_management_needs_an_admin_key() {
        let dir = TempDir::new("keyauth");
        let st = writable_state(&dir.0);
        let app = app!(st);

        assert_eq!(
            get!(app, "/_management/api/keys", NO_KEY).status(),
            StatusCode::UNAUTHORIZED
        );
        let resp = post_key!(
            app,
            json!({"key": "x", "admin": true, "read": [""], "write": [""]}),
            "not-a-key"
        );
        assert_eq!(resp.status(), StatusCode::FORBIDDEN);
    }

    // -- tiers -------------------------------------------------------------

    fn tiered_state(dir: &std::path::Path) -> State {
        let mut inner = SharedState::new(true);
        inner.acl.get_mut().unwrap().grant_full(ADMIN);
        inner.tiers_file = Some(dir.join("tiers.toml"));
        inner.db.write(
            "myorg/one",
            "v",
            Utc::now(),
            WriteOpts {
                consensus: true,
                ttl: None,
            },
        );
        web::Data::new(inner)
    }

    macro_rules! post_tier {
        ($app:expr, $body:expr, $key:expr) => {
            test::call_service(
                &$app,
                test::TestRequest::post()
                    .uri("/_management/api/tier")
                    .insert_header(("Authorization", $key))
                    .set_json($body)
                    .to_request(),
            )
            .await
        };
    }

    #[actix_web::test]
    async fn a_namespace_listing_carries_its_tier_and_residency() {
        let dir = TempDir::new("tierlist");
        let st = tiered_state(&dir.0);
        let app = app!(st);

        let body: Json =
            test::read_body_json(get!(app, "/_management/api/namespaces", Some(ADMIN))).await;
        let item = &body["items"][0];

        assert_eq!(item["namespace"], "myorg/one");
        // The tier belongs to the top-level namespace, which the row names so
        // the interface can say what a change will affect.
        assert_eq!(item["shard"], "myorg");
        assert_eq!(item["tier"], "hot");
        assert_eq!(item["resident"], true);
    }

    #[actix_web::test]
    async fn a_tier_change_takes_effect_and_is_written_out() {
        let dir = TempDir::new("tierset");
        let st = tiered_state(&dir.0);
        let app = app!(st);

        let resp = post_tier!(app, json!({"shard": "myorg", "tier": "cold"}), ADMIN);
        assert_eq!(resp.status(), StatusCode::OK);
        let body: Json =
            test::read_body_json(get!(app, "/_management/api/namespaces", Some(ADMIN))).await;
        assert_eq!(body["items"][0]["tier"], "cold");

        let written = std::fs::read_to_string(dir.0.join("tiers.toml")).unwrap();
        assert!(written.contains("\"myorg\" = \"cold\""), "{written}");
    }

    #[actix_web::test]
    async fn the_idle_window_can_be_changed_and_cleared() {
        let dir = TempDir::new("tieridle");
        let st = tiered_state(&dir.0);
        let app = app!(st);

        // The thing that could only be set in the configuration file before.
        let resp = post_tier!(
            app,
            json!({"namespace": "myorg/one", "tier": "warm", "warm_idle": 86400}),
            ADMIN
        );
        assert_eq!(resp.status(), StatusCode::OK);
        let body: Json = test::read_body_json(resp).await;
        // Named a namespace, changed its shard, and said so.
        assert_eq!(body["shard"], "myorg");
        assert_eq!(body["tier"], "warm");
        assert_eq!(body["warm_idle"], 86400);
        assert!(
            body["effect"]
                .as_str()
                .unwrap()
                .contains("everything under it"),
            "{body}"
        );

        let written = std::fs::read_to_string(dir.0.join("tiers.toml")).unwrap();
        assert!(
            written.contains(r#""myorg" = { tier = "warm", warm_idle = 86400 }"#),
            "{written}"
        );

        // A form hands over text, and "default" means stop saying anything.
        let resp = post_tier!(
            app,
            json!({"namespace": "myorg", "tier": "warm", "warm_idle": "default"}),
            ADMIN
        );
        let body: Json = test::read_body_json(resp).await;
        assert_eq!(body["own_warm_idle"], false);
        assert_eq!(body["warm_idle"], 3600, "back to the configured default");

        // Clearing both leaves nothing behind.
        let resp = post_tier!(app, json!({"namespace": "myorg", "tier": "default"}), ADMIN);
        assert_eq!(resp.status(), StatusCode::OK);
        let body: Json = test::read_body_json(resp).await;
        assert_eq!(body["own_tier"], false);
        let written = std::fs::read_to_string(dir.0.join("tiers.toml")).unwrap();
        assert!(!written.contains("myorg"), "{written}");
    }

    #[actix_web::test]
    async fn a_row_reports_the_window_its_shard_is_on() {
        let dir = TempDir::new("tierrow");
        let st = tiered_state(&dir.0);
        let app = app!(st);
        post_tier!(
            app,
            json!({"namespace": "myorg", "tier": "warm", "warm_idle": 900}),
            ADMIN
        );

        for uri in [
            "/_management/api/namespaces",
            "/_management/api/tree?path=myorg",
        ] {
            let body: Json = test::read_body_json(get!(app, uri, Some(ADMIN))).await;
            let item = &body["items"][0];
            assert_eq!(item["tier"], "warm", "{uri}");
            assert_eq!(item["warm_idle"], 900, "{uri}");
            assert_eq!(item["own_tier"], true, "{uri}");
            assert_eq!(item["own_warm_idle"], true, "{uri}");
        }
    }

    #[actix_web::test]
    async fn an_unreadable_idle_window_is_refused() {
        let dir = TempDir::new("tierjunk");
        let st = tiered_state(&dir.0);
        let app = app!(st);

        for body in [
            json!({"namespace": "myorg", "warm_idle": "soon"}),
            json!({"namespace": "myorg", "warm_idle": -5}),
            json!({"namespace": "", "tier": "warm"}),
        ] {
            assert_eq!(
                post_tier!(app, body.clone(), ADMIN).status(),
                StatusCode::BAD_REQUEST,
                "{body}"
            );
        }
        assert!(!dir.0.join("tiers.toml").exists());
    }

    #[actix_web::test]
    async fn an_unknown_tier_is_refused() {
        let dir = TempDir::new("tierbad");
        let st = tiered_state(&dir.0);
        let app = app!(st);

        let resp = post_tier!(app, json!({"shard": "myorg", "tier": "tepid"}), ADMIN);
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
        assert!(!dir.0.join("tiers.toml").exists());
    }

    /// Consensus and API keys are consulted constantly, so the internal shard
    /// is not something the interface may demote.
    #[actix_web::test]
    async fn internal_namespaces_cannot_be_retiered() {
        let dir = TempDir::new("tierinternal");
        let st = tiered_state(&dir.0);
        let app = app!(st);

        for shard in ["_all", "_config", ""] {
            let resp = post_tier!(app, json!({"shard": shard, "tier": "cold"}), ADMIN);
            assert_eq!(resp.status(), StatusCode::BAD_REQUEST, "{shard:?}");
        }
    }

    #[actix_web::test]
    async fn changing_a_tier_needs_an_admin_key() {
        let dir = TempDir::new("tierauth");
        let st = tiered_state(&dir.0);
        let app = app!(st);

        let resp = post_tier!(app, json!({"shard": "myorg", "tier": "cold"}), "plain");
        assert_eq!(resp.status(), StatusCode::FORBIDDEN);
    }

    /// Without a file there is nowhere to record the change, and a tier that
    /// silently reverted on restart would be worse than refusing.
    #[actix_web::test]
    async fn changing_a_tier_without_a_file_says_so() {
        let st = state();
        let app = app!(st);

        let resp = post_tier!(app, json!({"shard": "feeds", "tier": "cold"}), ADMIN);
        assert_eq!(resp.status(), StatusCode::CONFLICT);
        let body: Json = test::read_body_json(resp).await;
        assert!(
            body["message"].as_str().unwrap().contains("tiers_file"),
            "{body}"
        );
    }

    // -- browsing, creating and adding ------------------------------------

    macro_rules! post_json {
        ($app:expr, $uri:expr, $body:expr, $key:expr) => {
            test::call_service(
                &$app,
                test::TestRequest::post()
                    .uri($uri)
                    .insert_header(("Authorization", $key))
                    .set_json($body)
                    .to_request(),
            )
            .await
        };
    }

    #[actix_web::test]
    async fn the_tree_walks_one_level_at_a_time() {
        let st = state();
        let app = app!(st);

        // At the root, `feeds` is a folder: it holds `feeds/ips` but nothing
        // of its own.
        let body: Json =
            test::read_body_json(get!(app, "/_management/api/tree", Some(ADMIN))).await;
        assert_eq!(body["total"], 1);
        assert_eq!(body["items"][0]["name"], "feeds");
        assert_eq!(body["items"][0]["path"], "feeds");
        assert_eq!(body["items"][0]["is_namespace"], false);
        assert_eq!(body["items"][0]["descendants"], 1);

        // A level down, `ips` is the namespace holding the values.
        let body: Json =
            test::read_body_json(get!(app, "/_management/api/tree?path=feeds", Some(ADMIN))).await;
        assert_eq!(body["items"][0]["name"], "ips");
        assert_eq!(body["items"][0]["path"], "feeds/ips");
        assert_eq!(body["items"][0]["is_namespace"], true);
        assert_eq!(body["items"][0]["descendants"], 0);

        // And nothing below that.
        let body: Json = test::read_body_json(get!(
            app,
            "/_management/api/tree?path=feeds/ips",
            Some(ADMIN)
        ))
        .await;
        assert_eq!(body["total"], 0);
    }

    /// A key scoped to one subtree must not learn the names of the others,
    /// which for the tree means not even seeing the folder above them.
    #[actix_web::test]
    async fn the_tree_only_shows_what_the_key_may_read() {
        let mut inner = SharedState::new(false);
        inner.acl.get_mut().unwrap().grant_full(ADMIN);
        inner
            .acl
            .get_mut()
            .unwrap()
            .set("scoped", parse_grants("admin, r:feeds").unwrap());
        for namespace in ["feeds/ips", "private/ips"] {
            inner
                .db
                .write(namespace, "1.2.3.4", Utc::now(), WriteOpts::default());
        }
        let st = web::Data::new(inner);
        let app = app!(st);

        let body: Json =
            test::read_body_json(get!(app, "/_management/api/tree", Some("scoped"))).await;
        assert_eq!(body["total"], 1);
        assert_eq!(body["items"][0]["name"], "feeds");
    }

    #[actix_web::test]
    async fn a_created_namespace_is_empty_and_browsable() {
        let st = state();
        let app = app!(st);

        let resp = post_json!(
            app,
            "/_management/api/namespaces",
            json!({"namespace": "feeds/domains"}),
            ADMIN
        );
        assert_eq!(resp.status(), StatusCode::OK);

        // It exists as a folder under `feeds`...
        let body: Json =
            test::read_body_json(get!(app, "/_management/api/tree?path=feeds", Some(ADMIN))).await;
        let names: Vec<&str> = body["items"]
            .as_array()
            .unwrap()
            .iter()
            .map(|item| item["name"].as_str().unwrap())
            .collect();
        assert_eq!(names, ["domains", "ips"]);

        // ...and as a namespace holding nothing yet, rather than a 404.
        let body: Json = test::read_body_json(get!(
            app,
            "/_management/api/values?namespace=feeds/domains",
            Some(ADMIN)
        ))
        .await;
        assert_eq!(body["total"], 0);
    }

    #[actix_web::test]
    async fn creating_a_namespace_twice_is_refused() {
        let st = state();
        let app = app!(st);

        let body = json!({"namespace": "feeds/ips"});
        let resp = post_json!(app, "/_management/api/namespaces", body, ADMIN);
        assert_eq!(resp.status(), StatusCode::CONFLICT);
    }

    /// Slashes are how the interface nests folders, so a name doubling or
    /// trailing them must not create a second namespace beside the first.
    #[actix_web::test]
    async fn a_namespace_name_is_tidied_before_it_is_stored() {
        let st = state();
        let app = app!(st);

        let resp = post_json!(
            app,
            "/_management/api/namespaces",
            json!({"namespace": "/feeds//domains/ "}),
            ADMIN
        );
        assert_eq!(resp.status(), StatusCode::OK);
        let body: Json = test::read_body_json(resp).await;
        assert_eq!(body["namespace"], "feeds/domains");
        assert!(st.db.namespace_exists("feeds/domains"));
    }

    #[actix_web::test]
    async fn internal_and_empty_namespaces_are_refused() {
        let st = state();
        let app = app!(st);

        for name in ["_config/acl/apikeys/mine", "  ", "/", "with space"] {
            let resp = post_json!(
                app,
                "/_management/api/namespaces",
                json!({ "namespace": name }),
                ADMIN
            );
            assert_eq!(resp.status(), StatusCode::BAD_REQUEST, "{name}");
        }
    }

    #[actix_web::test]
    async fn values_can_be_added_one_at_a_time_or_in_bulk() {
        let st = state();
        let app = app!(st);

        let resp = post_json!(
            app,
            "/_management/api/values",
            json!({"namespace": "feeds/domains", "values": ["example.com"]}),
            ADMIN
        );
        assert_eq!(resp.status(), StatusCode::OK);
        let body: Json = test::read_body_json(resp).await;
        assert_eq!(body["written"], 1);
        // Writing is what creates a namespace, here as everywhere else.
        assert!(st.db.namespace_exists("feeds/domains"));

        // A pasted list, blank lines and all, lands as the values in it.
        let resp = post_json!(
            app,
            "/_management/api/values",
            json!({"namespace": "feeds/domains", "values": ["a.example", "", "  ", "b.example"]}),
            ADMIN
        );
        let body: Json = test::read_body_json(resp).await;
        assert_eq!(body["written"], 2);
        assert!(body["errors"].is_null(), "{body}");

        let page = st.db.value_page("feeds/domains", "", 0, 10, false).unwrap();
        assert_eq!(page.total, 3);
        // Counted towards consensus, exactly as a `/w/` write would be.
        assert_eq!(st.db.count(crate::db::ALL_NAMESPACE, "example.com"), 1);
    }

    #[actix_web::test]
    async fn added_values_take_a_ttl_and_a_time() {
        let st = state();
        let app = app!(st);

        // An hour ago with an hour to live: recorded then, and still here now.
        let seen = Utc::now().timestamp() - 60;
        let resp = post_json!(
            app,
            "/_management/api/values",
            json!({
                "namespace": "feeds/domains",
                "values": ["example.com"],
                "ttl": 3600,
                "timestamp": seen,
            }),
            ADMIN
        );
        assert_eq!(resp.status(), StatusCode::OK);

        let view = st
            .db
            .view("feeds/domains", "example.com", 0, false)
            .unwrap();
        assert_eq!(view.first_seen, seen);
        assert_eq!(view.ttl, 3600);

        // A time the calendar cannot hold is a client error, not a panic.
        let resp = post_json!(
            app,
            "/_management/api/values",
            json!({"namespace": "feeds/domains", "values": ["x"], "timestamp": i64::MAX}),
            ADMIN
        );
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
    }

    #[actix_web::test]
    async fn a_request_with_nothing_to_add_is_a_bad_request() {
        let st = state();
        let app = app!(st);

        let resp = post_json!(
            app,
            "/_management/api/values",
            json!({"namespace": "feeds/ips", "values": ["", " "]}),
            ADMIN
        );
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
    }

    /// Reaching the interface is not the same as being allowed to change the
    /// data in it: an `admin, r:feeds` key browses but does not write.
    #[actix_web::test]
    async fn writing_needs_a_write_grant_over_the_namespace() {
        let mut inner = SharedState::new(false);
        inner.acl.get_mut().unwrap().grant_full(ADMIN);
        inner
            .acl
            .get_mut()
            .unwrap()
            .set("reader", parse_grants("admin, r").unwrap());
        let st = web::Data::new(inner);
        let app = app!(st);

        for (uri, body) in [
            (
                "/_management/api/namespaces",
                json!({"namespace": "feeds/domains"}),
            ),
            (
                "/_management/api/values",
                json!({"namespace": "feeds/domains", "values": ["example.com"]}),
            ),
        ] {
            assert_eq!(
                post_json!(app, uri, body.clone(), "reader").status(),
                StatusCode::FORBIDDEN,
                "{uri}"
            );
            // A key that is not an admin key at all does not get further.
            assert_eq!(
                post_json!(app, uri, body, "nobody").status(),
                StatusCode::FORBIDDEN,
                "{uri}"
            );
        }
        assert!(!st.db.namespace_exists("feeds/domains"));
    }

    // -- the relationship graph -------------------------------------------

    #[actix_web::test]
    async fn a_value_reports_where_else_it_has_been_seen() {
        let mut inner = SharedState::new(false);
        inner.acl.get_mut().unwrap().grant_full(ADMIN);
        inner
            .acl
            .get_mut()
            .unwrap()
            .set("scoped", parse_grants("admin, r:feeds").unwrap());
        for namespace in ["feeds/misp/ips", "feeds/otx/ips", "private/ips"] {
            inner.db.write(
                namespace,
                "1.2.3.4",
                Utc::now(),
                WriteOpts {
                    consensus: true,
                    ttl: None,
                },
            );
        }
        let st = web::Data::new(inner);
        let app = app!(st);

        let body: Json = test::read_body_json(get!(
            app,
            "/_management/api/sightings?value=1.2.3.4",
            Some(ADMIN)
        ))
        .await;
        let namespaces: Vec<&str> = body["items"]
            .as_array()
            .unwrap()
            .iter()
            .map(|item| item["namespace"].as_str().unwrap())
            .collect();
        assert_eq!(
            namespaces,
            ["feeds/misp/ips", "feeds/otx/ips", "private/ips"]
        );
        assert_eq!(body["items"][0]["shard"], "feeds");
        assert_eq!(body["consensus"], 3);
        assert_eq!(body["truncated"], false);

        // A scoped key sees its own subtree; consensus still says how many
        // namespaces hold the value, so it can tell the graph is not all of it.
        let body: Json = test::read_body_json(get!(
            app,
            "/_management/api/sightings?value=1.2.3.4",
            Some("scoped")
        ))
        .await;
        assert_eq!(body["items"].as_array().unwrap().len(), 2);
        assert_eq!(body["consensus"], 3);
    }

    #[actix_web::test]
    async fn the_graph_needs_an_admin_key_and_a_value() {
        let st = state();
        let app = app!(st);

        assert_eq!(
            get!(app, "/_management/api/sightings?value=1.2.3.4", NO_KEY).status(),
            StatusCode::UNAUTHORIZED
        );
        assert_eq!(
            get!(
                app,
                "/_management/api/sightings?value=1.2.3.4",
                Some("plain")
            )
            .status(),
            StatusCode::FORBIDDEN
        );
        assert_eq!(
            get!(app, "/_management/api/sightings?value=", Some(ADMIN)).status(),
            StatusCode::BAD_REQUEST
        );
    }

    // -- tags and the STIX export -------------------------------------------

    #[actix_web::test]
    async fn values_can_be_added_with_tags_and_retagged_afterwards() {
        let st = state();
        let app = app!(st);

        let resp = post_json!(
            app,
            "/_management/api/values",
            json!({
                "namespace": "feeds/ips",
                "values": ["8.8.8.8"],
                "tags": "stix-type:ipv4-addr, tlp:green",
            }),
            ADMIN
        );
        assert_eq!(resp.status(), StatusCode::OK);
        assert_eq!(
            st.db.view("feeds/ips", "8.8.8.8", 0, false).unwrap().tags,
            "stix-type:ipv4-addr,tlp:green"
        );

        // Replacing is how a wrong tag comes off, and it is not a sighting:
        // the count stays where it was.
        let before = st.db.view("feeds/ips", "8.8.8.8", 0, false).unwrap().count;
        let resp = post_json!(
            app,
            "/_management/api/tags",
            json!({"namespace": "feeds/ips", "value": "8.8.8.8", "tags": "stix-type:ipv4-addr"}),
            ADMIN
        );
        assert_eq!(resp.status(), StatusCode::OK);
        let view = st.db.view("feeds/ips", "8.8.8.8", 0, false).unwrap();
        assert_eq!(view.tags, "stix-type:ipv4-addr");
        assert_eq!(view.count, before);
    }

    #[actix_web::test]
    async fn retagging_a_value_that_is_not_there_is_a_not_found() {
        let st = state();
        let app = app!(st);

        let resp = post_json!(
            app,
            "/_management/api/tags",
            json!({"namespace": "feeds/ips", "value": "203.0.113.9", "tags": "tlp:red"}),
            ADMIN
        );
        assert_eq!(resp.status(), StatusCode::NOT_FOUND);
    }

    #[actix_web::test]
    async fn tagging_needs_a_write_grant() {
        let mut inner = SharedState::new(false);
        inner.acl.get_mut().unwrap().grant_full(ADMIN);
        inner
            .acl
            .get_mut()
            .unwrap()
            .set("reader", parse_grants("admin, r").unwrap());
        inner
            .db
            .write("feeds/ips", "1.2.3.4", Utc::now(), WriteOpts::default());
        let st = web::Data::new(inner);
        let app = app!(st);

        let resp = post_json!(
            app,
            "/_management/api/tags",
            json!({"namespace": "feeds/ips", "value": "1.2.3.4", "tags": "tlp:red"}),
            "reader"
        );
        assert_eq!(resp.status(), StatusCode::FORBIDDEN);
        assert!(
            st.db
                .view("feeds/ips", "1.2.3.4", 0, false)
                .unwrap()
                .tags
                .is_empty()
        );
    }

    #[actix_web::test]
    async fn the_logo_is_served_from_the_binary() {
        let st = state();
        let app = app!(st);

        // No key: the page shows it before anyone has signed in.
        let resp = get!(app, "/_management/logo.png", NO_KEY);
        assert_eq!(resp.status(), StatusCode::OK);
        assert_eq!(resp.headers().get("content-type").unwrap(), "image/png");

        let body = test::read_body(resp).await;
        assert!(body.starts_with(b"\x89PNG"), "not a PNG");
    }

    #[actix_web::test]
    async fn echarts_is_served_from_the_binary() {
        let st = state();
        let app = app!(st);

        let resp = get!(app, "/_management/echarts.min.js", NO_KEY);
        assert_eq!(resp.status(), StatusCode::OK);
        let body = test::read_body(resp).await;
        assert!(body.len() > 100_000, "expected the real library");
    }
}