//! What the user of a request may do, from the user registry (`access`), checked on
//! every request ([`access.md`](../../docs/access.md)): the endpoints by `authorize`,
//! the store's paths by its handlers, through the `Access` it leaves on the request.

use axum::{
    body::Body,
    extract::State,
    http::{Method, Request, StatusCode},
    response::Response,
};
use serde_json::Value as JsonValue;
use tracing::{debug, warn};

use super::auth::AuthenticatedUser;
use super::error_response;
use crate::access::{Caps, Right, registry};
use crate::document::meta;
use crate::state::AppState;

/// What the user of a request may do (a request extension, set by `authorize`).
#[derive(Clone, Debug)]
pub(crate) enum Access {
    /// The swarm has no registry (`access.md`, A swarm without a registry): every user the
    /// node lets in may do what all could before; protocol data only those in
    /// `--protocol-writers` (`store::parse_write_path`).
    Everything,
    /// What its registry entry gives.
    Caps(Caps),
}

/// How much of a place a user may read (`Access::visible`).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum Visible {
    /// It and all below it.
    All,
    /// Only what lies below it on the way to what it may read (a map's keys leading
    /// there, as a directory one may pass through but not list): `prune`.
    Partly,
    Nothing,
}

impl Access {
    pub(crate) fn allows(&self, right: Right, path: &[String]) -> bool {
        match self {
            Self::Everything => true,
            Self::Caps(caps) => caps.allows(right, path),
        }
    }

    /// As capabilities: `Caps` as they are; in a swarm without a registry, what every
    /// user may do there (the swarm's data, protocol data to read, and to write for
    /// `--protocol-writers`), to narrow by a token's limit.
    pub(crate) fn caps(&self, state: &AppState, user: &str) -> Caps {
        use crate::access::Capability;
        match self {
            Self::Caps(caps) => caps.clone(),
            Self::Everything => {
                let protocol: &[&str] = if state.protocol_writers.contains(user) {
                    &["read", "write"]
                } else {
                    &["read"]
                };
                let cap = |path: Vec<String>, rights: &[&str]| Capability {
                    path,
                    rights: rights.iter().map(|right| (*right).to_owned()).collect(),
                    by: String::new(),
                    since: 0,
                };
                Caps::new(vec![
                    cap(Vec::new(), &["read", "write"]),
                    cap(meta::path(&[]), protocol),
                ])
            }
        }
    }

    pub(crate) fn visible(&self, path: &[String]) -> Visible {
        match self {
            Self::Everything => Visible::All,
            // The root holds protocol data too, which a capability on `[]` doesn't give.
            Self::Caps(caps)
                if caps.allows(Right::Read, path)
                    && (!path.is_empty() || caps.allows(Right::Read, &meta::path(&[]))) =>
            {
                Visible::All
            }
            Self::Caps(caps) if caps.reaches(Right::Read, path) => Visible::Partly,
            Self::Caps(_) => Visible::Nothing,
        }
    }
}

/// `value`, read at `path`, as much of it as `access` lets its user read: of a map it
/// may only partly read, the keys on the way to what it may (recursively); a list, a
/// text or a scalar it may only partly read is left out, as a list's items have no path
/// of their own to give rights to. `typed`: `value` is typed JSON
/// (`typed::JsonForm::typed`), whose maps may be tagged (`{"$map": {…}}`) and whose
/// tagged values (`{"$uint": 5}`) are scalars. None if nothing of it is readable.
pub(crate) fn prune(
    value: JsonValue,
    path: &mut Vec<String>,
    access: &Access,
    typed: bool,
) -> Option<JsonValue> {
    match access.visible(path) {
        Visible::All => return Some(value),
        Visible::Nothing => return None,
        Visible::Partly => {}
    }
    let JsonValue::Object(map) = value else {
        return None;
    };
    let tag = |map: &serde_json::Map<String, JsonValue>| {
        (typed && map.len() == 1)
            .then(|| {
                map.keys()
                    .next()
                    .filter(|key| key.starts_with('$'))
                    .cloned()
            })
            .flatten()
    };
    let map = match tag(&map) {
        None => map,
        Some(tag) if tag == "$map" => match map.into_iter().next() {
            Some((_, JsonValue::Object(inner))) => inner,
            _ => return None,
        },
        // A tagged scalar or text.
        Some(_) => return None,
    };
    let mut out = serde_json::Map::new();
    for (key, child) in map {
        path.push(key.clone());
        let child = prune(child, path, access, typed);
        path.pop();
        if let Some(child) = child {
            out.insert(key, child);
        }
    }
    // A map whose only key looks like a tag is tagged itself.
    Some(match tag(&out) {
        Some(_) => serde_json::json!({ "$map": out }),
        None => JsonValue::Object(out),
    })
}

/// A shallow read of `path` (`store::read_shallow`), its `entries` only those `access`
/// lets its user read at least partly.
pub(crate) fn prune_shallow(mut value: JsonValue, path: &[String], access: &Access) -> JsonValue {
    if access.visible(path) == Visible::All {
        return value;
    }
    if let Some(entries) = value.get_mut("entries").and_then(JsonValue::as_object_mut) {
        let mut child = path.to_vec();
        entries.retain(|key, _| {
            child.push(key.clone());
            let visible = access.visible(&child) != Visible::Nothing;
            child.pop();
            visible
        });
        let size = entries.len();
        value["size"] = size.into();
    }
    value
}

/// The registry as a write found it (`registry_kept`): whether it existed, and whether
/// the writing user could manage users (`registry::managers`).
#[derive(Clone, Copy)]
pub(crate) struct Before {
    existed: bool,
    managed: bool,
}

impl Before {
    pub(crate) fn of(doc: &impl crate::document::ReadDoc, user: &str) -> Self {
        use crate::access::registry;
        let existed = registry::exists(doc);
        Self {
            existed,
            managed: existed
                && registry::caps_of(doc, user)
                    .allows(crate::access::Right::Write, &registry::users_dir()),
        }
    }
}

/// `409` for a write to `path` by `user` that would lock that user out of managing
/// users: one that could and no longer may (deleting, disabling or narrowing its own
/// entry), and the write that makes the registry, if it doesn't make `user` one who may
/// (from then on only registered users are served). And for one that would remove
/// the registry, back to a swarm without one. Checked in the write's transaction,
/// after it: the write is dropped whole. Others are not kept from removing the last who
/// may manage users: no node could keep two such writes at once on two nodes from
/// leaving nobody once they merge; then the users in `--protocol-writers` may do
/// everything again (`access::authorize`).
pub(crate) fn registry_kept(
    doc: &impl crate::document::ReadDoc,
    path: &[String],
    before: Before,
    user: &str,
) -> Result<(), super::ApiError> {
    use crate::access::registry;
    if !crate::document::store::paths_related(path, &registry::users_dir()) {
        return Ok(());
    }
    let refuse = |why: &str| {
        Err(super::ApiError(
            StatusCode::CONFLICT,
            Some(format!("{why} (docs/access.md)")),
        ))
    };
    if !registry::exists(doc) {
        return if before.existed {
            refuse("this would remove the user registry")
        } else {
            Ok(())
        };
    }
    let manages =
        registry::caps_of(doc, user).allows(crate::access::Right::Write, &registry::users_dir());
    if !before.existed && !manages {
        return refuse("the registry's first users must include you, with write on _continuum");
    }
    if before.managed && !manages {
        return refuse("this would keep you from managing users");
    }
    Ok(())
}

/// What an endpoint needs (`needed`).
#[derive(Debug, PartialEq, Eq)]
enum Need {
    /// The store and its streams: their handlers check each path.
    Store,
    /// Looking at this node (its figures, peers, dashboard): any capability.
    Observe,
    /// Logging in, refreshing, asking who one is: any registered user.
    Auth,
    /// A right on protocol data.
    Protocol(Right, Vec<String>),
}

/// What a request to `path` by `method` needs. Remakes' records and forget records
/// are protocol data of their own: proposing or aborting a remake needs `write` on
/// `_continuum/remakes`, forgetting a peer on `_continuum/forgotten`; everything else
/// this node's operators do (its log filter, exports, quarantine, the changes) needs
/// protocol data itself, reading it to look, writing it to act.
fn needed(method: &Method, path: &str, node: &str) -> Need {
    let looking = method == Method::GET || method == Method::HEAD;
    let protocol = |right, parts: &[&str]| Need::Protocol(right, meta::path(parts));
    let under = |prefix: &str| path == prefix || path.starts_with(&format!("{prefix}/"));
    // What this node holds alone: `_continuum/local/<node>/<name>`, never stored.
    let local = |right, name: &str| protocol(right, &[meta::LOCAL, node, name]);
    if under("/store") || under("/stream") {
        Need::Store
    } else if under("/auth") {
        Need::Auth
    } else if (path == "/peers" && looking) || path == "/admin/stats" || under("/admin/dashboard") {
        Need::Observe
    } else if under("/peers") {
        protocol(Right::Write, &[meta::FORGOTTEN])
    } else if under("/admin/users") {
        // Changing them is checked by their handlers too (`users::may_manage`): in a
        // swarm without a registry, `--protocol-writers`.
        if looking {
            protocol(Right::Read, &[meta::USERS])
        } else {
            protocol(Right::Write, &[meta::USERS])
        }
    } else if under("/admin/remakes") {
        if looking {
            protocol(Right::Read, &[meta::REMAKES])
        } else {
            protocol(Right::Write, &[meta::REMAKES])
        }
    } else if under("/admin/log") || under("/admin/config") {
        let name = if under("/admin/log") { "log" } else { "config" };
        local(if looking { Right::Read } else { Right::Write }, name)
    } else if under("/debug") {
        local(Right::Write, "debug")
    } else if looking {
        protocol(Right::Read, &[])
    } else {
        protocol(Right::Write, &[])
    }
}

/// Finds what the request's user may do, refuses the request if its endpoint needs
/// more (`needed`), and leaves `Access` on it for the store's handlers. Inside
/// `auth::basic_auth`, which found the user. A user the registry doesn't have may do
/// nothing: `403`.
pub(crate) async fn authorize(
    State(state): State<AppState>,
    mut request: Request<Body>,
    next: axum::middleware::Next,
) -> Response {
    let Some(AuthenticatedUser(user)) = request.extensions().get::<AuthenticatedUser>().cloned()
    else {
        return error_response(StatusCode::UNAUTHORIZED, "not authenticated", None);
    };
    let access = {
        let replica = state.doc.read().await;
        if !registry::exists(&replica.doc) {
            Access::Everything
        } else if state.protocol_writers.contains(&user)
            && registry::managers(&replica.doc).is_empty()
        {
            // Nobody may manage users any more (two removals at once, merged; or an
            // entry damaged): this node's protocol writers may set it right.
            warn!(user = %user, "The user registry has nobody who may manage users; a protocol writer may do everything until it has");
            Access::Everything
        } else {
            match registry::user(&replica.doc, &user) {
                Some(entry) => Access::Caps(entry.caps()),
                None => {
                    debug!(user = %user, "Refused a user the registry doesn't have");
                    return error_response(
                        StatusCode::FORBIDDEN,
                        format!("{user} is not a registered user (docs/access.md)"),
                        None,
                    );
                }
            }
        }
    };
    let allowed = match needed(request.method(), request.uri().path(), state.local_id()) {
        Need::Store | Need::Auth => true,
        Need::Observe => match &access {
            Access::Everything => true,
            Access::Caps(caps) => !caps.is_empty(),
        },
        Need::Protocol(right, path) => access.allows(right, &path),
    };
    if !allowed {
        debug!(user = %user, method = %request.method(), path = %request.uri().path(), "Refused a request the user's capabilities don't cover");
        return error_response(
            StatusCode::FORBIDDEN,
            format!("{user} may not do that here (docs/access.md)"),
            None,
        );
    }
    // A token that says less: only what it allows too.
    let access = match request
        .extensions()
        .get::<super::login::Token>()
        .and_then(|token| token.claims.limit.clone())
    {
        None => access,
        Some(limit) => Access::Caps(access.caps(&state, &user).within(limit)),
    };
    request.extensions_mut().insert(access);
    next.run(request).await
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::access::Capability;
    use serde_json::json;

    fn path(text: &str) -> Vec<String> {
        text.split('/')
            .filter(|segment| !segment.is_empty())
            .map(str::to_owned)
            .collect()
    }

    fn caps(list: &[(&str, &[&str])]) -> Access {
        Access::Caps(Caps::new(
            list.iter()
                .map(|(at, rights)| Capability {
                    path: path(at),
                    rights: rights.iter().map(|right| (*right).to_owned()).collect(),
                    by: String::new(),
                    since: 0,
                })
                .collect(),
        ))
    }

    #[test]
    fn endpoints_and_what_they_need() {
        let read = |parts: &[&str]| Need::Protocol(Right::Read, meta::path(parts));
        let write = |parts: &[&str]| Need::Protocol(Right::Write, meta::path(parts));
        assert_eq!(needed(&Method::GET, "/store", "n1"), Need::Store);
        assert_eq!(needed(&Method::POST, "/auth/login", "n1"), Need::Auth);
        assert_eq!(needed(&Method::GET, "/auth/whoami", "n1"), Need::Auth);
        assert_eq!(needed(&Method::PUT, "/store/a/b", "n1"), Need::Store);
        assert_eq!(needed(&Method::GET, "/stream/a", "n1"), Need::Store);
        assert_eq!(needed(&Method::GET, "/storehouse", "n1"), read(&[]));
        assert_eq!(needed(&Method::GET, "/peers", "n1"), Need::Observe);
        assert_eq!(needed(&Method::GET, "/admin/stats", "n1"), Need::Observe);
        assert_eq!(
            needed(&Method::GET, "/admin/dashboard/state", "n1"),
            Need::Observe
        );
        assert_eq!(
            needed(&Method::POST, "/admin/dashboard/refresh", "n1"),
            Need::Observe
        );
        assert_eq!(
            needed(&Method::DELETE, "/peers/continuum:00", "n1"),
            write(&[meta::FORGOTTEN])
        );
        assert_eq!(
            needed(&Method::GET, "/admin/remakes/r1", "n1"),
            read(&[meta::REMAKES])
        );
        assert_eq!(
            needed(&Method::POST, "/admin/remakes", "n1"),
            write(&[meta::REMAKES])
        );
        assert_eq!(
            needed(&Method::DELETE, "/admin/remakes/r1", "n1"),
            write(&[meta::REMAKES])
        );
        assert_eq!(
            needed(&Method::GET, "/admin/config", "n1"),
            read(&[meta::LOCAL, "n1", "config"])
        );
        assert_eq!(
            needed(&Method::GET, "/admin/users/alice", "n1"),
            read(&[meta::USERS])
        );
        assert_eq!(
            needed(&Method::POST, "/admin/users", "n1"),
            write(&[meta::USERS])
        );
        assert_eq!(needed(&Method::GET, "/admin/quarantine/q", "n1"), read(&[]));
        assert_eq!(
            needed(&Method::POST, "/admin/quarantine/q/replay", "n1"),
            write(&[])
        );
        assert_eq!(
            needed(&Method::GET, "/admin/log", "n1"),
            read(&[meta::LOCAL, "n1", "log"])
        );
        assert_eq!(
            needed(&Method::PUT, "/admin/log", "n1"),
            write(&[meta::LOCAL, "n1", "log"])
        );
        assert_eq!(
            needed(&Method::GET, "/debug/firewall", "n1"),
            write(&[meta::LOCAL, "n1", "debug"])
        );
        // Every node's log filter, by one capability.
        let logs = Caps::new(vec![Capability {
            path: path("_continuum/local/*/log"),
            rights: ["write".to_owned()].into(),
            by: String::new(),
            since: 0,
        }]);
        let Need::Protocol(right, at) = needed(&Method::PUT, "/admin/log", "n2") else {
            unreachable!()
        };
        assert!(logs.allows(right, &at));
        let Need::Protocol(right, at) = needed(&Method::GET, "/admin/config", "n2") else {
            unreachable!()
        };
        assert!(!logs.allows(right, &at));
    }

    #[test]
    fn what_a_user_sees_of_a_tree() {
        let tree = json!({
            "servers": { "eu": { "a": 1 }, "us": { "b": 2 } },
            "list": [1, 2],
            "_continuum": { "users": {} }
        });
        let reader = caps(&[("servers/eu", &["read"]), ("list/0", &["read"])]);
        assert_eq!(
            prune(tree.clone(), &mut Vec::new(), &reader, false),
            Some(json!({ "servers": { "eu": { "a": 1 } } }))
        );
        let data = caps(&[("", &["read", "write"])]);
        assert_eq!(
            prune(tree.clone(), &mut Vec::new(), &data, false),
            Some(json!({ "servers": { "eu": { "a": 1 }, "us": { "b": 2 } }, "list": [1, 2] }))
        );
        assert_eq!(
            prune(tree.clone(), &mut Vec::new(), &Access::Everything, false),
            Some(tree.clone())
        );
        assert_eq!(
            prune(tree, &mut Vec::new(), &caps(&[("x", &["read"])]), false),
            Some(json!({}))
        );
        assert_eq!(
            prune(json!(1), &mut path("y"), &caps(&[("x", &["read"])]), false),
            None
        );
    }

    #[test]
    fn typed_json_keeps_its_tags() {
        let reader = caps(&[("m/$x", &["read"]), ("n/k", &["read"])]);
        let tree = json!({
            "m": { "$map": { "$x": 1, "$y": 2 } },
            "n": { "k": { "$uint": 5 }, "$z": 1 }
        });
        assert_eq!(
            prune(tree, &mut Vec::new(), &reader, true),
            Some(json!({ "m": { "$map": { "$x": 1 } }, "n": { "k": { "$uint": 5 } } }))
        );
        // A tagged scalar on the way is no map.
        let tagged = json!({ "t": { "$text": "hello" } });
        assert_eq!(
            prune(tagged, &mut Vec::new(), &caps(&[("t/x", &["read"])]), true),
            Some(json!({}))
        );
    }

    #[test]
    fn a_shallow_read_lists_the_way() {
        let read = json!({ "type": "map", "size": 3,
                           "entries": { "a": {}, "b": {}, "_continuum": {} } });
        let reader = caps(&[("a/x", &["read"])]);
        assert_eq!(
            prune_shallow(read.clone(), &[], &reader),
            json!({ "type": "map", "size": 1, "entries": { "a": {} } })
        );
        assert_eq!(prune_shallow(read.clone(), &[], &Access::Everything), read);
    }
}