//! What an `include:` scope names, and what the permission set it names
//! grants.
//!
//! A permission set is a lexicon: a `permission-set` definition published
//! under the set's own NSID, whose `permissions` are `repo` and `rpc`
//! permissions in their JSON form. [`Include::grants`] reads that list the
//! way the permission spec's "Permission Sets" section says an
//! authorization server must:
//!
//! - It ignores a permission for any other resource, and a permission with a
//!   field or a value it cannot read. It never grants a permission in part.
//! - Every collection and method must sit under the set's own NSID group.
//!   `app.example.authFull` may grant `app.example.post` and
//!   `app.example.feed.like`, and nothing under `app.other`. A wildcard is
//!   never one of them.
//! - An `rpc` permission names `*` as its audience, or sets `inheritAud` and
//!   takes the audience the `include:` scope names. With `inheritAud` and no
//!   audience to take, it grants nothing.
//!
//! Finding the set is the caller's job: the DNS record, the authority's
//! repository, and the proof the record comes with.

use std::collections::BTreeSet;

use didbot_lexicon::nsid_syntax;
use serde_json::{Map, Value};

use crate::scope::ANY_AUD;
use crate::{Action, ActionSet, NsidPattern, Scope};

/// An `include:` scope, read: the permission set it names, and the audience
/// that set's `rpc` permissions may inherit.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Include {
    nsid: String,
    aud: Option<String>,
}

/// Why an `include:` scope names no permission set.
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum IncludeError {
    /// The set's name is not an NSID.
    #[error("`{0}` is not an NSID")]
    NotAnNsid(String),
    /// `aud` is not a `did:plc` or a `did:web` with a service fragment.
    #[error("`{0}` is not a DID with a service fragment")]
    NotAServiceReference(String),
}

impl Include {
    /// The set an `include:` scope names, and the audience it passes on,
    /// both percent-decoded as [`Scope::Include`] holds them.
    ///
    /// `nsid` must be an NSID, and `aud` must name one service:
    /// `did:web:api.example.com#svc_appview`. `*` is refused.
    pub fn new(nsid: &str, aud: Option<&str>) -> Result<Self, IncludeError> {
        if nsid_syntax::validate(nsid).is_err() {
            return Err(IncludeError::NotAnNsid(nsid.to_owned()));
        }
        if let Some(aud) = aud.filter(|aud| !is_service_reference(aud)) {
            return Err(IncludeError::NotAServiceReference(aud.to_owned()));
        }
        Ok(Self {
            nsid: nsid.to_owned(),
            aud: aud.map(str::to_owned),
        })
    }

    /// The permission set's NSID.
    pub fn nsid(&self) -> &str {
        &self.nsid
    }

    /// The audience the set's `inheritAud` permissions take, percent-decoded.
    pub fn aud(&self) -> Option<&str> {
        self.aud.as_deref()
    }

    /// The atoms a set's published `permissions` grant through this
    /// `include:`: one per collection and one per method.
    pub fn grants(&self, permissions: &[Value]) -> Vec<Scope> {
        permissions
            .iter()
            .filter_map(|permission| self.grant(permission.as_object()?))
            .flatten()
            .collect()
    }

    /// One permission's atoms, or `None` for a permission this ignores.
    fn grant(&self, permission: &Map<String, Value>) -> Option<Vec<Scope>> {
        if permission.get("type")?.as_str()? != "permission" {
            return None;
        }
        if permission
            .get("description")
            .is_some_and(|description| !description.is_string())
        {
            return None;
        }
        match permission.get("resource")?.as_str()? {
            "repo" => self.repo(permission),
            "rpc" => self.rpc(permission),
            _ => None,
        }
    }

    fn repo(&self, permission: &Map<String, Value>) -> Option<Vec<Scope>> {
        has_only(permission, &["collection", "action"])?;
        let collections = self.nsids(permission.get("collection")?, true)?;
        let actions = match permission.get("action") {
            None => ActionSet::All,
            Some(actions) => ActionSet::Only(repo_actions(actions)?),
        };
        Some(
            collections
                .into_iter()
                .map(|collection| Scope::Repo {
                    collection: NsidPattern::Exact(collection),
                    actions: actions.clone(),
                })
                .collect(),
        )
    }

    fn rpc(&self, permission: &Map<String, Value>) -> Option<Vec<Scope>> {
        has_only(permission, &["lxm", "aud", "inheritAud"])?;
        let methods = self.nsids(permission.get("lxm")?, false)?;
        let inherits = match permission.get("inheritAud") {
            None => false,
            Some(inherits) => inherits.as_bool()?,
        };
        let aud = match (inherits, permission.get("aud")) {
            (true, None) => self.aud.clone()?,
            (false, Some(aud)) if aud.as_str() == Some(ANY_AUD) => ANY_AUD.to_owned(),
            _ => return None,
        };
        Some(
            methods
                .into_iter()
                .map(|method| Scope::Rpc {
                    method: NsidPattern::Exact(method),
                    aud: Some(aud.clone()),
                })
                .collect(),
        )
    }

    /// A non-empty list of NSIDs under this set's group, or `None` if any
    /// entry is not one. `unique` refuses a list that names one twice.
    fn nsids(&self, list: &Value, unique: bool) -> Option<BTreeSet<String>> {
        let list = list.as_array().filter(|list| !list.is_empty())?;
        let mut nsids = BTreeSet::new();
        for nsid in list {
            let nsid = nsid.as_str()?;
            if nsid_syntax::validate(nsid).is_err() || !self.authorizes(nsid) {
                return None;
            }
            if !nsids.insert(nsid.to_owned()) && unique {
                return None;
            }
        }
        Some(nsids)
    }

    /// Whether `nsid` sits under this set's group: the set's own NSID less
    /// its name. The group's parent and its siblings are outside it.
    fn authorizes(&self, nsid: &str) -> bool {
        let Some((group, _)) = self.nsid.rsplit_once('.') else {
            return false;
        };
        nsid.strip_prefix(group)
            .and_then(|rest| rest.strip_prefix('.'))
            .is_some_and(|rest| !rest.is_empty())
    }
}

/// `Some` when every field of `permission` is `type`, `resource`,
/// `description`, or one of `fields`.
fn has_only(permission: &Map<String, Value>, fields: &[&str]) -> Option<()> {
    permission
        .keys()
        .all(|key| {
            matches!(key.as_str(), "type" | "resource" | "description")
                || fields.contains(&key.as_str())
        })
        .then_some(())
}

/// A non-empty list of distinct `repo` actions, or `None`.
fn repo_actions(list: &Value) -> Option<BTreeSet<Action>> {
    let list = list.as_array().filter(|list| !list.is_empty())?;
    let mut actions = BTreeSet::new();
    for action in list {
        let action = match action.as_str()? {
            "create" => Action::Create,
            "update" => Action::Update,
            "delete" => Action::Delete,
            _ => return None,
        };
        if !actions.insert(action) {
            return None;
        }
    }
    Some(actions)
}

/// Whether `aud` names one service: a `did:plc` or a `did:web`, then `#`
/// and a fragment.
pub(crate) fn is_service_reference(aud: &str) -> bool {
    let Some((did, fragment)) = aud.split_once('#') else {
        return false;
    };
    (did.starts_with("did:plc:") || did.starts_with("did:web:"))
        && didbot_identity::validate_did(did).is_ok()
        && is_fragment(fragment)
}

/// Whether `fragment` is a non-empty RFC 3986 fragment: `pchar`, `/` and
/// `?`, with `%` only as an escape.
fn is_fragment(fragment: &str) -> bool {
    let bytes = fragment.as_bytes();
    let mut at = 0;
    while at < bytes.len() {
        match bytes[at] {
            b'%' if bytes
                .get(at + 1..at + 3)
                .is_some_and(|hex| hex.iter().all(u8::is_ascii_hexdigit)) =>
            {
                at += 3;
            }
            byte if byte.is_ascii_alphanumeric() || b"-._~!$&'()*+,;=:@/?".contains(&byte) => {
                at += 1;
            }
            _ => return false,
        }
    }
    !bytes.is_empty()
}

#[cfg(test)]
mod tests {
    use serde_json::json;

    use super::*;
    use crate::ScopeSet;

    const APPVIEW: &str = "did:web:api.example.com#svc_appview";

    /// What `include:<name>` reads as, or why it names no set.
    fn read(name: &str) -> Result<Include, IncludeError> {
        let Scope::Include { nsid, aud } = Scope::parse(&format!("include:{name}")).unwrap() else {
            panic!("`include:{name}` is not an include");
        };
        Include::new(&nsid, aud.as_deref())
    }

    fn include(name: &str) -> Include {
        read(name).unwrap_or_else(|error| panic!("`{name}`: {error}"))
    }

    /// What `permissions` grants through `include:<name>`, printed.
    fn granted(name: &str, permissions: Value) -> String {
        let permissions = permissions.as_array().expect("a list").clone();
        ScopeSet::new(include(name).grants(&permissions)).to_string()
    }

    #[test]
    fn a_set_grants_each_collection_and_method_it_names() {
        let permissions = json!([
            {
                "type": "permission",
                "resource": "repo",
                "collection": ["com.example.post", "com.example.feed.like"],
                "action": ["create", "delete"],
            },
            {
                "type": "permission",
                "resource": "repo",
                "collection": ["com.example.profile"],
            },
            {
                "type": "permission",
                "resource": "rpc",
                "lxm": ["com.example.getFeedSkeleton"],
                "aud": "*",
            },
        ]);
        assert_eq!(
            granted("com.example.authFull", permissions),
            "repo:com.example.feed.like?action=create&action=delete \
             repo:com.example.post?action=create&action=delete \
             repo:com.example.profile \
             rpc:com.example.getFeedSkeleton?aud=*"
        );
    }

    /// The spec's own example: `inheritAud` takes the audience the
    /// `include:` names, percent-decoded, and it prints the way a client
    /// would have written it.
    #[test]
    fn the_include_aud_reaches_the_rpc_permissions_that_inherit_it() {
        let permissions = json!([{
            "type": "permission",
            "resource": "rpc",
            "inheritAud": true,
            "lxm": ["com.example.getFeed", "com.example.getProfile"],
        }]);
        let name = "com.example.authBasicFeatures?aud=did:web:api.example.com%23svc_appview";
        assert_eq!(include(name).aud(), Some(APPVIEW));
        let grant = granted(name, permissions);
        assert_eq!(
            grant,
            "rpc:com.example.getFeed?aud=did:web:api.example.com%23svc_appview \
             rpc:com.example.getProfile?aud=did:web:api.example.com%23svc_appview"
        );
        assert!(ScopeSet::parse(&grant)
            .expect("a grant parses")
            .permits_rpc(Some("com.example.getFeed"), APPVIEW));
    }

    /// `inheritAud` with nothing to inherit, or beside an `aud` of its own,
    /// is invalid: the permission grants nothing, and never an `rpc:` scope
    /// with no audience, which would reach every service.
    #[test]
    fn inherit_aud_grants_nothing_without_one_audience_to_take() {
        let inherits = json!([{
            "type": "permission",
            "resource": "rpc",
            "inheritAud": true,
            "lxm": ["com.example.getFeed"],
        }]);
        assert_eq!(granted("com.example.authFull", inherits), "");

        let both = json!([{
            "type": "permission",
            "resource": "rpc",
            "inheritAud": true,
            "aud": "*",
            "lxm": ["com.example.getFeed"],
        }]);
        assert_eq!(
            granted(
                "com.example.authFull?aud=did:web:api.example.com%23svc_appview",
                both
            ),
            ""
        );
    }

    /// A set grants lexicons under its own group and the groups below it,
    /// never its group's parent, a sibling, or a name that only shares the
    /// group's text.
    #[test]
    fn a_set_grants_only_under_its_own_group() {
        let repo = |collection: &str| {
            json!([{
                "type": "permission",
                "resource": "repo",
                "collection": [collection],
            }])
        };
        for (collection, granted_it) in [
            ("app.example.feed.post", true),
            ("app.example.feed.thread.gate", true),
            ("app.example.feed", false),
            ("app.example.actor.profile", false),
            ("app.example.feedx.post", false),
            ("app.other.feed.post", false),
            ("*", false),
            ("app.example.feed.*", false),
        ] {
            let grant = granted("app.example.feed.authOnlyPost", repo(collection));
            assert_eq!(!grant.is_empty(), granted_it, "`{collection}`: `{grant}`");
        }
    }

    /// The spec: "ignore any individual permission declarations within a
    /// permission set that describe an unknown resource, or include
    /// unexpected parameter names or values". Each of these sits beside one
    /// permission that is fine, which still grants.
    #[test]
    fn a_permission_this_cannot_read_is_ignored_whole() {
        let fine = json!({
            "type": "permission",
            "resource": "repo",
            "collection": ["com.example.post"],
        });
        for ignored in [
            json!({"type": "permission", "resource": "blob", "accept": ["image/*"]}),
            json!({"type": "permission", "resource": "account", "attr": "email"}),
            json!({"type": "permission", "resource": "identity", "attr": "handle"}),
            json!({"type": "permission", "resource": "scry", "collection": ["com.example.x"]}),
            json!({"type": "token", "resource": "repo", "collection": ["com.example.x"]}),
            json!({"resource": "repo", "collection": ["com.example.x"]}),
            json!({"type": "permission", "resource": "repo", "collection": ["com.example.x"],
                   "maxRecords": 1}),
            json!({"type": "permission", "resource": "repo", "collection": ["com.example.x"],
                   "action": ["manage"]}),
            json!({"type": "permission", "resource": "repo", "collection": ["com.example.x"],
                   "action": ["create", "create"]}),
            json!({"type": "permission", "resource": "repo", "collection": ["com.example.x"],
                   "action": []}),
            json!({"type": "permission", "resource": "repo",
                   "collection": ["com.example.x", "com.example.x"]}),
            json!({"type": "permission", "resource": "repo", "collection": []}),
            json!({"type": "permission", "resource": "repo", "collection": "com.example.x"}),
            json!({"type": "permission", "resource": "repo",
                   "collection": ["com.example.x", "com.other.x"]}),
            json!({"type": "permission", "resource": "repo", "collection": ["com.example.x"],
                   "description": 7}),
            json!({"type": "permission", "resource": "rpc", "lxm": ["com.example.get"],
                   "aud": APPVIEW}),
            json!({"type": "permission", "resource": "rpc", "lxm": ["com.example.get"]}),
            json!({"type": "permission", "resource": "rpc", "lxm": ["com.example.get"],
                   "inheritAud": "yes"}),
            json!({"type": "permission", "resource": "rpc", "lxm": ["*"], "aud": "*"}),
            json!("repo:com.example.x"),
        ] {
            assert_eq!(
                granted(
                    "com.example.authFull",
                    json!([ignored.clone(), fine.clone()])
                ),
                "repo:com.example.post",
                "{ignored}"
            );
        }
    }

    /// Fields the lexicon spec defines: `description` on every schema
    /// object, and `inheritAud` as a boolean that may be `false`.
    #[test]
    fn a_description_and_a_false_inherit_aud_are_read() {
        let permissions = json!([
            {
                "type": "permission",
                "resource": "repo",
                "collection": ["com.example.post"],
                "description": "Posts.",
            },
            {
                "type": "permission",
                "resource": "rpc",
                "lxm": ["com.example.getFeed"],
                "aud": "*",
                "inheritAud": false,
            },
        ]);
        assert_eq!(
            granted("com.example.authFull", permissions),
            "repo:com.example.post rpc:com.example.getFeed?aud=*"
        );
    }

    #[test]
    fn an_include_names_a_set_by_its_nsid_and_one_service() {
        assert_eq!(
            include("com.example.authFull").nsid(),
            "com.example.authFull"
        );
        assert_eq!(include("com.example.authFull").aud(), None);
        assert_eq!(
            include("com.example.authFull?aud=did:web:api.example.com#svc_appview").aud(),
            Some(APPVIEW)
        );
        assert_eq!(
            include("com.example.authFull?aud=did:plc:ewvi7nxzyoun6zhxrhs64oiz%23atproto_labeler")
                .aud(),
            Some("did:plc:ewvi7nxzyoun6zhxrhs64oiz#atproto_labeler")
        );
        for refused in [
            "authFull",
            "com.example",
            "com.example.*",
            "com.example.authFull?aud=*",
            "com.example.authFull?aud=did:web:api.example.com",
            "com.example.authFull?aud=did:web:api.example.com%23",
            "com.example.authFull?aud=did:key:zQ3shokFTS3brHcDQrn82RUDfCZESWL1ZdCEJwekUDPQiYBme%23x",
            "com.example.authFull?aud=did:web:a.example%23x%23y",
        ] {
            assert!(read(refused).is_err(), "`include:{refused}` names a set");
        }
        // Forms the spec does not define, read as well as they can be.
        for (spelled, aud) in [
            (
                "com.example.authFull?aud=did:web:a.example%23x&aud=did:web:b.example%23x",
                Some("did:web:a.example#x"),
            ),
            ("com.example.authFull?nsid=com.example.other", None),
            ("com.example.authFull?lang=en", None),
            ("?nsid=com.example.authFull", None),
        ] {
            let read = include(spelled);
            assert_eq!(read.nsid(), "com.example.authFull", "`include:{spelled}`");
            assert_eq!(read.aud(), aud, "`include:{spelled}`");
        }
        for unread in [
            "include:?aud=did:web:api.example.com%23svc",
            "include:com.example.authFull?aud",
        ] {
            assert!(
                Scope::parse(unread).is_err(),
                "`{unread}` read as an include"
            );
        }
    }
}