//! What an admitted client is allowed to do.
//!
//! [`crate::introspect`] answers *whether* to admit and *as whom*. This
//! answers *what that identity may touch* — the two are deliberately separate
//! decisions: admission is the `IdP`'s, authorisation is the swarm's.
//!
//! # Deny is the default, and that is a choice with a measurement behind it
//!
//! A client whose id matches no rule gets **no grant at all**, not an
//! unrestricted one. Every other shape here fails open: an unmatched client
//! that kept today's unscoped grant would make the whole policy advisory, and
//! the failure would be silent. Denying costs a loud refusal the first time a
//! new consumer appears, which is a config line to fix.
//!
//! Each subject set below is *minimal by removal*, and carries its own note
//! saying what breaks without it — the reasoning lives next to the list it
//! constrains rather than in one block here, because that is where a reader
//! about to edit the list will meet it.

/// The subjects an admitted client may publish to.
///
/// **Publish only — subscription is left unrestricted.** The queue's
/// confidentiality boundary is the account, and a hive reading another hive's
/// *published* status is not the problem this solves; writing it is. Scoping
/// `sub` is a separate change with its own measurement, and claiming it here
/// without one would repeat the `$JS.API.>` mistake described on
/// `Policy::hive_subjects`.
#[derive(Debug, PartialEq, Eq)]
pub struct Permissions {
    /// Subjects allowed for publish. Never empty: an empty allow-list is a
    /// grant that can do nothing, which is a denial wearing a grant's shape.
    pub publish: Vec<String>,
}

/// Which client ids get which permissions.
///
/// Constructed from configuration so that adding a subject a hive may publish
/// — a second event stream, say — is a deployment change rather than a change
/// to this responder.
#[derive(Debug, Clone)]
pub struct Policy {
    hive_prefix: String,
    bucket: String,
    readers: Vec<String>,
    extra_hive_subjects: Vec<String>,
}

/// Placeholder replaced with the hive's own name in `extra_hive_subjects`.
const HIVE_PLACEHOLDER: &str = "{hive}";

impl Policy {
    /// `hive_prefix` is the client-id prefix that marks a hive, `bucket` the
    /// KV bucket hives report status in, `readers` the client ids allowed to
    /// read every hive's key, and `extra_hive_subjects` additional subjects a
    /// hive may publish to (with `{hive}` standing for its own name).
    ///
    /// # Errors
    ///
    /// An extra subject with no `{hive}` in it is refused. Such a template
    /// expands to the **same** subject for every hive, which is not a per-hive
    /// namespace — it is the absence of one, arrived at through the option
    /// whose only purpose is to provide one. Returning a `Result` rather than
    /// checking at the call site is deliberate: it makes an unscoped policy
    /// unconstructible instead of merely unlikely.
    pub fn new(
        hive_prefix: String,
        bucket: String,
        readers: Vec<String>,
        extra_hive_subjects: Vec<String>,
    ) -> anyhow::Result<Self> {
        if let Some(bad) = extra_hive_subjects
            .iter()
            .find(|s| !s.contains(HIVE_PLACEHOLDER))
        {
            anyhow::bail!(
                "--hive-publish-subject {bad:?} contains no {HIVE_PLACEHOLDER}: every hive \
                 would be granted that exact subject, so it is a shared channel rather than \
                 a per-hive namespace"
            );
        }
        Ok(Self {
            hive_prefix,
            bucket,
            readers,
            extra_hive_subjects,
        })
    }

    /// The permissions for `client_id`, or `None` when no rule matches.
    ///
    /// `None` is a denial. It is not "grant nothing and let them connect":
    /// a connected client with no permissions still holds a slot and still
    /// looks admitted in the logs, which is a worse answer than a refusal.
    pub fn permissions(&self, client_id: &str) -> Option<Permissions> {
        if let Some(hive) = self.hive_name(client_id) {
            return Some(Permissions {
                publish: self.hive_subjects(hive),
            });
        }
        if self.readers.iter().any(|r| r == client_id) {
            return Some(Permissions {
                publish: self.reader_subjects(),
            });
        }
        None
    }

    /// The hive a client id names, when it names one.
    ///
    /// The prefix is configuration, not a pattern this module guesses at: the
    /// authelia client is `hive-<name>` while the KV key is the bare `<name>`,
    /// and stripping by eye is how a client id that merely *looks*
    /// hive-shaped ends up with a hive's permissions. An id that is the prefix
    /// and nothing else names no hive and is refused — `$KV.<bucket>.` is not
    /// a narrower subject than `$KV.<bucket>.<name>`, it is a different one.
    ///
    /// # This matches the prefix; it does not verify the roster
    ///
    /// A client id is a hive here because it *starts with the prefix*, not
    /// because it appears in `services.hyperhive.swarm.hives`. This responder
    /// runs inside a container and has no view of the roster, so an
    /// operator-declared client called `hive-anything` would be granted
    /// `$KV.<bucket>.anything` — a key no real hive owns, which the controller
    /// renders as a hive it does not recognise.
    ///
    /// Passing the roster in would close that, and would also be a **second
    /// place deciding who may connect as what**, which `crate::introspect`'s
    /// docs argue against for the same reason admission lives in one place.
    /// The prefix is a contract with `swarm-authelia.nix`, and this is the end
    /// of it that can be checked from in here.
    fn hive_name<'a>(&self, client_id: &'a str) -> Option<&'a str> {
        client_id
            .strip_prefix(&self.hive_prefix)
            .filter(|name| !name.is_empty())
    }

    fn stream(&self) -> String {
        format!("KV_{}", self.bucket)
    }

    /// What *any* `JetStream` client must be able to ask before it can do
    /// anything at all, bucket-specific or not.
    ///
    /// Both were measured from the server's own refusals, not reasoned about:
    /// a grant carrying every bucket-specific subject and neither of these
    /// cannot even create the bucket — the client times out on `$JS.API.INFO`
    /// long before it reaches a subject that was granted.
    ///
    /// - `$JS.API.INFO` — account-level `JetStream` info, requested on connect.
    /// - `$JS.API.STREAM.NAMES` — how a client finds the stream backing a
    ///   bucket. It lets a client enumerate stream names in the account, which
    ///   in an account holding one bucket discloses a name both ends already
    ///   share.
    ///
    /// 🩸 Earlier measurements missed both, because they either granted
    /// `$JS.API.>` wholesale or ran against a bucket the *setup* had already
    /// created while unscoped. A minimum established against an existing
    /// bucket is not the minimum for making one.
    fn jetstream_minimum() -> [String; 2] {
        ["$JS.API.INFO".to_owned(), "$JS.API.STREAM.NAMES".to_owned()]
    }

    /// Creating the bucket, which **both** ends need.
    ///
    /// `swarm_queue_client::status::open_or_create` is called by the hive that
    /// writes and the controller that reads, because either may arrive first
    /// on a fresh swarm — a bucket that must pre-exist turns "deployed in the
    /// wrong order" into a permanent, silent absence of data. Whichever
    /// connects first therefore has to be able to create it.
    ///
    /// Narrower than it looks: this is `CREATE` on one named stream, not
    /// `UPDATE` and not the `$JS.API.>` wildcard. Measured rather than
    /// assumed — a hive holding this grant and running `stream edit` leaves
    /// the stream's config untouched, and no `$JS.API.STREAM.UPDATE` is ever
    /// published. Worth stating because the failure it would hide is quiet: a
    /// hive able to reshape the shared bucket could set `MaxMsgs: 1` and evict
    /// every other hive's status without ever touching `STREAM.DELETE`, and
    /// per-key scoping would still look intact.
    fn create(&self) -> String {
        format!("$JS.API.STREAM.CREATE.{}", self.stream())
    }

    /// What every hive needs to open the shared lifecycle-notices stream:
    /// look it up, and create it if this hive is the first to arrive.
    ///
    /// **Not per-hive templated, unlike `extra_hive_subjects`.** The KV
    /// bucket case above namespaces a shared resource *within* itself
    /// (`$KV.<bucket>.<hive>`, one key per hive); the notices stream has
    /// no such per-hive split at the `STREAM.INFO`/`STREAM.CREATE` layer
    /// — the stream itself, not a slice of it, is what every hive's
    /// `open_or_create` needs to reach before it can publish to its own
    /// `hive-notices.<hive>` subject (which `extra_hive_subjects` already
    /// covers, since that one *is* per-hive). Granting these two subjects
    /// to every hive is therefore correct, not a widening: it is
    /// `CREATE`/`INFO` on one named stream, the same shape already
    /// measured safe for the hive-status bucket in [`Self::create`] —
    /// create-if-absent, never `STREAM.UPDATE`, so no hive can reshape a
    /// stream another hive (or the swarm-controller, once it reads from
    /// this stream) already created.
    fn notices_subjects() -> [String; 2] {
        // Unlike `Self::stream()` above, no `KV_` prefix: the notices
        // stream is a plain `JetStream` stream, not a KV bucket, so its
        // NATS stream name *is* `swarm_queue_client::notices::STREAM`
        // verbatim — the `KV_` prefix is `create_key_value`'s own
        // convention, not something every stream carries.
        let stream = swarm_queue_client::notices::STREAM;
        [
            format!("$JS.API.STREAM.INFO.{stream}"),
            format!("$JS.API.STREAM.CREATE.{stream}"),
        ]
    }

    /// What one hive may publish: the account minimum, the bucket lookup,
    /// creation, and its **own** key.
    ///
    /// Two things a reader would reasonably assume, both false and both
    /// measured (`state/attack-3297-js-api-door.sh` in atlas's notes):
    ///
    /// - `$KV.<bucket>.<key>` alone does **not** let a client write that key.
    ///   The client resolves the bucket first, so the `STREAM.INFO` subject is
    ///   part of the minimum for a plain write.
    /// - `$JS.API.>` is not "the `JetStream` permission". It also covers
    ///   `$JS.API.STREAM.DELETE`, with which a hive correctly refused on a
    ///   neighbour's individual key can destroy the whole bucket — every
    ///   hive's data. Granting it would make per-key scoping decorative, which
    ///   is why these are named one at a time and a test asserts the wildcard
    ///   never returns as a convenience.
    fn hive_subjects(&self, hive: &str) -> Vec<String> {
        let mut subjects = Self::jetstream_minimum().to_vec();
        subjects.extend([
            format!("$JS.API.STREAM.INFO.{}", self.stream()),
            self.create(),
            format!("$KV.{}.{hive}", self.bucket),
        ]);
        subjects.extend(Self::notices_subjects());
        subjects.extend(
            self.extra_hive_subjects
                .iter()
                .map(|s| s.replace(HIVE_PLACEHOLDER, hive)),
        );
        subjects
    }

    fn reader_subjects(&self) -> Vec<String> {
        let stream = self.stream();
        let mut subjects = Self::jetstream_minimum().to_vec();
        subjects.extend([
            format!("$JS.API.STREAM.INFO.{stream}"),
            self.create(),
            // The `.>` form specifically: the bare `$JS.API.DIRECT.GET.<stream>`
            // is not the subject the client uses, and granting it was measured
            // to make no difference.
            format!("$JS.API.DIRECT.GET.{stream}.>"),
            // `store.keys()` — the controller lists before it fetches, so a
            // reader without this can get a key it already knows and discover
            // nothing.
            //
            // BOTH forms, and the bare one is the one that matters. `keys()`
            // creates an **ephemeral** ordered consumer, whose create subject
            // carries no consumer name — and `>` matches one or more tokens,
            // never zero, so the `.>` form alone does not cover it. Granting
            // only that produced `Permissions Violation for Publish to
            // "$JS.API.CONSUMER.CREATE.KV_hive-status"`, which reaches the
            // client as a **timeout** and the operator as a 503. The `.>` form
            // stays for a named/durable consumer.
            format!("$JS.API.CONSUMER.CREATE.{stream}"),
            format!("$JS.API.CONSUMER.CREATE.{stream}.>"),
            // The knowledge event. One writer, many readers: the controller is
            // the only publisher and every hive subscribes, so this is one
            // literal subject rather than a per-hive family — named from the
            // crate the publisher and the subscriber also name, so a rename
            // cannot leave the grant pointing at a subject nobody uses.
            //
            // This is the reader's only non-JetStream subject, and without it
            // the controller cannot emit the event at all. Worth stating because
            // the symptom is unhelpful: a refused publish reaches the client as
            // a **timeout**, so the visible failure is a hive that never hears
            // about a change, with nothing in the controller's log to say a
            // permission was the reason.
            swarm_queue_client::KNOWLEDGE_SUBJECT.to_owned(),
        ]);
        subjects
    }
}

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

    fn policy() -> Policy {
        Policy::new(
            "hive-".to_owned(),
            "hive-status".to_owned(),
            vec!["swarm-controller".to_owned()],
            Vec::new(),
        )
        .expect("the default policy is valid")
    }

    #[test]
    fn an_extra_subject_without_the_placeholder_is_refused() {
        // 🩸 The option exists to put a second stream inside ONE hive's
        // namespace. A template with no `{hive}` expands to the same subject
        // for all of them, so the flag whose purpose is scoping becomes the
        // way to remove it - silently, and only in the deployment that set it.
        let err = Policy::new(
            "hive-".to_owned(),
            "hive-status".to_owned(),
            Vec::new(),
            vec!["$SWARM.events.all".to_owned()],
        )
        .expect_err("a subject shared by every hive must not be accepted");
        let msg = format!("{err}");
        assert!(
            msg.contains("$SWARM.events.all"),
            "the error must name the offending template, got: {msg}"
        );
    }

    #[test]
    fn an_extra_subject_with_the_placeholder_is_accepted() {
        // The other half: a check that only ever rejects would be indistinguishable
        // from the option being broken.
        Policy::new(
            "hive-".to_owned(),
            "hive-status".to_owned(),
            Vec::new(),
            vec!["$SWARM.events.{hive}.>".to_owned()],
        )
        .expect("a per-hive template is the shape this option is for");
    }

    #[test]
    fn a_hive_may_publish_its_own_key_and_nothing_elses() {
        let p = policy()
            .permissions("hive-alpha")
            .expect("a hive is admitted");
        assert!(p.publish.contains(&"$KV.hive-status.alpha".to_owned()));
        assert!(!p.publish.iter().any(|s| s.contains("beta")));
    }

    #[test]
    fn a_reader_may_publish_the_knowledge_event() {
        let p = policy().permissions("swarm-controller").expect("a reader");
        assert!(
            p.publish
                .contains(&swarm_queue_client::KNOWLEDGE_SUBJECT.to_owned()),
            "the controller is the only publisher of this event; without the \
             grant its publish is refused, and a refusal arrives as a timeout"
        );
    }

    #[test]
    fn a_hive_may_not_publish_the_knowledge_event_to_anyone_including_itself() {
        // The controller *interprets* what a delivery means; a hive receives
        // that verdict. A hive able to publish here could tell every other hive
        // in the swarm — or itself — that the knowledge repo changed when it did
        // not, which is an unauthenticated write into someone else's control
        // path wearing an event's shape.
        //
        // One writer and many readers makes this arm matter MORE, not less: with
        // a single shared subject a forged event reaches the whole swarm, where
        // a per-hive subject would have reached one.
        let p = policy()
            .permissions("hive-alpha")
            .expect("a hive is admitted");
        assert!(
            !p.publish
                .iter()
                .any(|s| s == swarm_queue_client::KNOWLEDGE_SUBJECT),
            "a hive must not publish the knowledge event: {:?}",
            p.publish
        );
    }

    #[test]
    fn a_hive_grant_never_includes_the_jetstream_wildcard() {
        // `$JS.API.>` also covers `$JS.API.STREAM.DELETE.KV_hive-status`, with
        // which a hive refused on a neighbour's key can delete the entire
        // bucket. Measured, not theorised - this test is the guard against it
        // coming back as a convenience.
        let p = policy()
            .permissions("hive-alpha")
            .expect("a hive is admitted");
        assert!(!p.publish.iter().any(|s| s.contains("$JS.API.>")));
        assert!(!p.publish.iter().any(|s| s.contains("STREAM.DELETE")));
        assert!(!p.publish.iter().any(|s| s.contains("STREAM.PURGE")));
    }

    #[test]
    fn the_bucket_lookup_is_part_of_a_publishers_minimum() {
        // Without it the put is refused: the client resolves the bucket before
        // it writes. Dropping this line looks like tightening and is breaking.
        let p = policy()
            .permissions("hive-alpha")
            .expect("a hive is admitted");
        assert!(
            p.publish
                .contains(&"$JS.API.STREAM.INFO.KV_hive-status".to_owned())
        );
    }

    #[test]
    fn every_grant_carries_the_jetstream_minimum() {
        // 🩸 Found by the shipping gate, not by any unit test: a grant with
        // every bucket-specific subject and neither of these cannot create the
        // bucket at all. The client times out on `$JS.API.INFO` before it
        // reaches anything that was granted, and a NATS denial looks like a
        // hang from the client side — the server log is what named them.
        for client in ["hive-alpha", "swarm-controller"] {
            let p = policy().permissions(client).expect("admitted");
            for required in ["$JS.API.INFO", "$JS.API.STREAM.NAMES"] {
                assert!(
                    p.publish.iter().any(|s| s == required),
                    "{client} is missing {required}, so it cannot use JetStream at all"
                );
            }
        }
    }

    #[test]
    fn both_ends_may_create_the_bucket_but_not_reshape_it() {
        // 🩸 The bug the subject measurements could not see: they ran against
        // a bucket the *setup* had already created while unscoped, so
        // "minimal" meant minimal-given-a-bucket-that-exists. `open_or_create`
        // is called by both ends by design — whichever arrives first on a
        // fresh swarm makes the bucket — so without CREATE a new swarm never
        // gets one, and every other test still passes.
        for client in ["hive-alpha", "swarm-controller"] {
            let p = policy().permissions(client).expect("admitted");
            assert!(
                p.publish
                    .contains(&"$JS.API.STREAM.CREATE.KV_hive-status".to_owned()),
                "{client} cannot create the bucket on a fresh swarm"
            );
            // CREATE is not UPDATE: a second arrival must not be able to
            // reshape the bucket the first one made.
            assert!(!p.publish.iter().any(|s| s.contains("STREAM.UPDATE")));
        }
    }

    #[test]
    fn a_reader_may_list_and_fetch_but_not_write() {
        let p = policy()
            .permissions("swarm-controller")
            .expect("the reader is admitted");
        // The BARE subject, asserted separately and first: `keys()` creates an
        // ephemeral consumer, so its create subject ends at the stream name,
        // and `>` matches one or more tokens rather than zero. This assertion
        // used to name only the `.>` form below — which reads as covering the
        // bare one and does not, so the suite was green while every list timed
        // out in production.
        assert!(
            p.publish
                .contains(&"$JS.API.CONSUMER.CREATE.KV_hive-status".to_owned()),
            "an ephemeral consumer create carries no name token"
        );
        assert!(
            p.publish
                .contains(&"$JS.API.CONSUMER.CREATE.KV_hive-status.>".to_owned())
        );
        assert!(
            p.publish
                .contains(&"$JS.API.DIRECT.GET.KV_hive-status.>".to_owned())
        );
        assert!(!p.publish.iter().any(|s| s.starts_with("$KV.")));
    }

    #[test]
    fn an_unmatched_client_is_denied_not_granted_everything() {
        // The whole policy is advisory if this returns a grant.
        assert_eq!(policy().permissions("some-other-service"), None);
        assert_eq!(policy().permissions(""), None);
    }

    #[test]
    fn the_prefix_alone_names_no_hive() {
        // `hive-` would otherwise produce `$KV.hive-status.`, a subject nobody
        // reviewed and that no hive owns.
        assert_eq!(policy().permissions("hive-"), None);
    }

    #[test]
    fn a_hive_may_open_the_shared_notices_stream() {
        // 🩸 The actual defect this closes: `open_or_create`'s `get_stream`
        // call needs these two before a hive ever reaches its own
        // `hive-notices.<hive>` publish subject, and neither has a
        // `{hive}` to go through `extra_hive_subjects`.
        let p = policy()
            .permissions("hive-alpha")
            .expect("a hive is admitted");
        assert!(
            p.publish
                .contains(&"$JS.API.STREAM.INFO.hive-notices".to_owned())
        );
        assert!(
            p.publish
                .contains(&"$JS.API.STREAM.CREATE.hive-notices".to_owned())
        );
    }

    #[test]
    fn the_notices_stream_grant_is_identical_across_hives() {
        // Unlike `$KV.<bucket>.<hive>` or an `extra_hive_subjects`
        // template, these two subjects name the stream itself, not a
        // per-hive slice of it — every hive gets the exact same two
        // strings, and that is the correct shape, not an oversight.
        let alpha = policy()
            .permissions("hive-alpha")
            .expect("a hive is admitted");
        let beta = policy()
            .permissions("hive-beta")
            .expect("a hive is admitted");
        for subject in [
            "$JS.API.STREAM.INFO.hive-notices",
            "$JS.API.STREAM.CREATE.hive-notices",
        ] {
            assert!(alpha.publish.contains(&subject.to_owned()));
            assert!(beta.publish.contains(&subject.to_owned()));
        }
    }

    #[test]
    fn the_notices_grant_never_includes_stream_update_or_delete() {
        // Same invariant `a_hive_grant_never_includes_the_jetstream_wildcard`
        // holds for the status bucket, restated for the stream this fold
        // grants CREATE/INFO on: create-if-absent must not become
        // reshape-or-destroy.
        let p = policy()
            .permissions("hive-alpha")
            .expect("a hive is admitted");
        assert!(
            !p.publish
                .iter()
                .any(|s| s.contains("hive-notices") && s.contains("STREAM.UPDATE"))
        );
        assert!(
            !p.publish
                .iter()
                .any(|s| s.contains("hive-notices") && s.contains("STREAM.DELETE"))
        );
    }

    #[test]
    fn extra_subjects_are_scoped_to_the_hive_that_publishes_them() {
        // The extension point: a second stream (lifecycle notices, say) is
        // published by the same `hive-<name>` identity on a different subject.
        // It has to land inside that hive's namespace, or one hive can write
        // another's events even though its status key is scoped.
        let p = Policy::new(
            "hive-".to_owned(),
            "hive-status".to_owned(),
            Vec::new(),
            vec!["$SWARM.events.{hive}.>".to_owned()],
        )
        .expect("a per-hive template is valid");
        let g = p.permissions("hive-alpha").expect("a hive is admitted");
        assert!(g.publish.contains(&"$SWARM.events.alpha.>".to_owned()));
        assert!(!g.publish.iter().any(|s| s.contains("{hive}")));
    }
}