greentic-start-dev 1.1.27024967183

//! M1.4c-ii admit-gate table: the projection of
//! [`Environment::messaging_endpoints`](greentic_deploy_spec::Environment) that
//! the revision ingress consults when a request carries a
//! `x-greentic-messaging-endpoint-id` header.
//!
//! M1.4c-i started receiving that header in [`crate::revision_serve`] and
//! threading it through to the runtime context, but accepted any value. This
//! module turns the header into a real authorization gate by answering one
//! question per request: *given the endpoint id the caller claims, is the
//! resolved deployment's bundle in that endpoint's `linked_bundles` ACL?*
//!
//! See `greentic-deploy-spec::messaging_endpoint` on why `linked_bundles` is
//! an ACL rather than a deployment selector — the runtime resolves the
//! concrete [`BundleDeployment`](greentic_deploy_spec::BundleDeployment) via
//! existing route binding + traffic-split routing first, and only then asks
//! this table whether that bundle is reachable through the asserted endpoint.
//!
//! The empty-table case (env declares no messaging endpoints) intentionally
//! fail-closes every header-asserted request: declaring an endpoint is the
//! one and only way to opt in.
//!
//! Composition with the rest of the serve pipeline is two-step but a *single*
//! conceptual gate:
//!
//! 1. Before dispatch we look the endpoint up — unknown endpoint ⇒ refuse the
//!    request cheaply with `UNAUTHORIZED` and don't waste a dispatch on it.
//! 2. After the dispatcher picks a revision we check membership of
//!    `outcome.bundle_id` in the ACL — outside-of-ACL ⇒ `FORBIDDEN`.
//!
//! Requests without the header take neither branch and stay on the legacy
//! single-instance path (back-compat for environments that never adopt M1).

use std::collections::{HashMap, HashSet};

use greentic_deploy_spec::{BundleId, Environment, WelcomeFlowRef};

/// The per-endpoint state the revision ingress needs at request time:
/// the `linked_bundles` ACL plus the M1.5 welcome-flow ref (if declared).
/// Co-locating these prevents drift between two parallel maps keyed on the
/// same endpoint id.
#[derive(Clone, Debug)]
struct EndpointEntry {
    linked_bundles: HashSet<String>,
    /// [`MessagingEndpoint::welcome_flow`](greentic_deploy_spec::MessagingEndpoint::welcome_flow)
    /// cloned in so the lookup is one map. Read by the producer at
    /// `revision_serve::serve` to build the per-request `WelcomeFlowHint`.
    welcome_flow: Option<WelcomeFlowRef>,
}

/// Per-endpoint ACL projection of `Environment.messaging_endpoints` consulted
/// by the revision ingress; see the module docs.
#[derive(Clone, Debug, Default)]
pub struct EndpointAdmit {
    /// Key: the on-wire `endpoint_id` form (the same string the caller asserts
    /// in `x-greentic-messaging-endpoint-id`, which matches
    /// `MessagingEndpointId::to_string`).
    by_id: HashMap<String, EndpointEntry>,
    /// Two-level index: `provider_type → provider_id → endpoint_id`. Drives
    /// the M1 IID.4 resolver: when a request arrives without
    /// `x-greentic-messaging-endpoint-id`, the host invokes each enabled
    /// provider component's `identify-instance` export; the returned
    /// `provider_id` is paired with the resolver's known `provider_type` and
    /// looked up here to recover the matching `endpoint_id`.
    ///
    /// Nested over a flat `(String, String)` key map because:
    /// * Two-step lookup borrows on `&str` directly, no per-call allocation.
    /// * `provider_types()` is the outer map's keys.
    /// * `endpoint_count_for_provider_type(t)` is the inner map's `.len()` —
    ///   the separate count field this replaced is redundant.
    ///
    /// (`provider_type`, `provider_id`) uniqueness is an
    /// [`Environment::validate`](greentic_deploy_spec::Environment::validate)
    /// invariant, so duplicates can't reach this table.
    by_provider_type: HashMap<String, HashMap<String, String>>,
}

impl EndpointAdmit {
    /// Build an admit table from the env's declared endpoints. Each endpoint's
    /// `linked_bundles` is materialized as a `HashSet<String>` so membership
    /// checks at request time are O(1) regardless of ACL size.
    pub fn from_environment(env: &Environment) -> Self {
        let mut by_id = HashMap::with_capacity(env.messaging_endpoints.len());
        let mut by_provider_type: HashMap<String, HashMap<String, String>> = HashMap::new();
        for ep in &env.messaging_endpoints {
            let endpoint_id = ep.endpoint_id.to_string();
            by_id.insert(
                endpoint_id.clone(),
                EndpointEntry {
                    linked_bundles: ep
                        .linked_bundles
                        .iter()
                        .map(|b| b.as_str().to_string())
                        .collect(),
                    welcome_flow: ep.welcome_flow.clone(),
                },
            );
            by_provider_type
                .entry(ep.provider_type.clone())
                .or_default()
                .insert(ep.provider_id.clone(), endpoint_id);
        }
        Self {
            by_id,
            by_provider_type,
        }
    }

    /// Look up the on-wire `endpoint_id` for a `(provider_type, provider_id)`
    /// pair. Used by the M1 IID.4 resolver to recover an `endpoint_id` from a
    /// `provider_id` returned by a component's `identify-instance` probe.
    pub(crate) fn endpoint_id_for_provider(
        &self,
        provider_type: &str,
        provider_id: &str,
    ) -> Option<&str> {
        self.by_provider_type
            .get(provider_type)
            .and_then(|m| m.get(provider_id))
            .map(String::as_str)
    }

    /// Iterate over the distinct `provider_type` values declared in this env.
    /// The resolver uses this to pick which provider components to probe per
    /// request (one probe per type, not per endpoint).
    pub(crate) fn provider_types(&self) -> impl Iterator<Item = &str> {
        self.by_provider_type.keys().map(String::as_str)
    }

    /// Number of endpoints declared for `provider_type`. ≥2 means a missing
    /// header MUST fail closed when the resolver cannot disambiguate.
    pub(crate) fn endpoint_count_for_provider_type(&self, provider_type: &str) -> usize {
        self.by_provider_type
            .get(provider_type)
            .map(HashMap::len)
            .unwrap_or(0)
    }

    /// Look up the ACL set for `endpoint_id`. `None` means *this env has never
    /// declared that endpoint* — the caller MUST refuse the request, not fall
    /// through to the legacy path. The set itself may be empty (a declared but
    /// unwired endpoint), in which case any subsequent bundle check rejects.
    pub fn linked_bundles(&self, endpoint_id: &str) -> Option<&HashSet<String>> {
        self.by_id.get(endpoint_id).map(|e| &e.linked_bundles)
    }

    /// Look up the raw M1.5 welcome-flow ref for `endpoint_id`, if the
    /// endpoint is declared AND has `welcome_flow` set. Returns `None` for
    /// both unknown endpoints and known-but-unset welcome flows — the
    /// producer cannot distinguish those at this site because
    /// [`linked_bundles`] has already classified unknowns as `UNAUTHORIZED`
    /// upstream.
    ///
    /// Most callers want [`welcome_flow_for_bundle`]; this raw lookup is
    /// kept for projection-level tests.
    ///
    /// [`linked_bundles`]: EndpointAdmit::linked_bundles
    /// [`welcome_flow_for_bundle`]: EndpointAdmit::welcome_flow_for_bundle
    pub(crate) fn welcome_flow(&self, endpoint_id: &str) -> Option<&WelcomeFlowRef> {
        self.by_id
            .get(endpoint_id)
            .and_then(|e| e.welcome_flow.as_ref())
    }

    /// Look up the welcome-flow ref for `endpoint_id` **only when the
    /// dispatched bundle matches the welcome ref's bundle**. Returns `None`
    /// when the endpoint is unknown, has no welcome ref, OR dispatched into
    /// a sibling bundle.
    ///
    /// The deploy-spec only invariant is `welcome_flow.bundle_id ∈
    /// linked_bundles`, NOT that dispatch lands on the welcome bundle.
    /// Endpoints can `linked_bundles` more than one bundle, and dispatch
    /// picks one via traffic splits / pins. Crossing bundles would target
    /// the welcome bundle's pack/flow on a sibling bundle's revision —
    /// either misroute (pack-id collision) or 500 (pack absent).
    pub(crate) fn welcome_flow_for_bundle(
        &self,
        endpoint_id: &str,
        dispatched_bundle: &BundleId,
    ) -> Option<&WelcomeFlowRef> {
        self.welcome_flow(endpoint_id)
            .filter(|ref_| &ref_.bundle_id == dispatched_bundle)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::test_fixtures::{endpoint, endpoint_typed, env_with};
    use greentic_deploy_spec::{BundleId, PackId};

    #[test]
    fn empty_env_yields_empty_table() {
        let admit = EndpointAdmit::from_environment(&env_with(Vec::new()));
        assert!(admit.linked_bundles("anything").is_none());
    }

    #[test]
    fn endpoint_lookup_keys_on_endpoint_id_string() {
        let ep = endpoint("teams-legal", &["legal-bundle", "shared-utils"]);
        let id = ep.endpoint_id.to_string();
        let admit = EndpointAdmit::from_environment(&env_with(vec![ep]));

        let bundles = admit
            .linked_bundles(&id)
            .expect("declared endpoint should resolve");
        assert!(bundles.contains("legal-bundle"));
        assert!(bundles.contains("shared-utils"));
        assert!(!bundles.contains("finance-bundle"));
    }

    #[test]
    fn endpoint_with_empty_acl_is_known_but_never_admits() {
        let ep = endpoint("teams-bare", &[]);
        let id = ep.endpoint_id.to_string();
        let admit = EndpointAdmit::from_environment(&env_with(vec![ep]));

        let bundles = admit
            .linked_bundles(&id)
            .expect("declared endpoint must resolve even with empty ACL");
        assert!(bundles.is_empty());
    }

    #[test]
    fn unknown_endpoint_id_returns_none() {
        let admit =
            EndpointAdmit::from_environment(&env_with(vec![endpoint("teams-legal", &["legal"])]));
        assert!(admit.linked_bundles("bogus-endpoint-id").is_none());
        assert!(admit.linked_bundles("").is_none());
    }

    #[test]
    fn welcome_flow_lookup_returns_ref_when_declared() {
        let mut ep = endpoint("teams-legal", &["legal-bundle"]);
        ep.welcome_flow = Some(WelcomeFlowRef {
            bundle_id: BundleId::new("legal-bundle"),
            pack_id: PackId::new("legal-pack"),
            flow_id: "welcome".to_string(),
        });
        let id = ep.endpoint_id.to_string();
        let admit = EndpointAdmit::from_environment(&env_with(vec![ep]));

        let ref_ = admit.welcome_flow(&id).expect("welcome ref present");
        assert_eq!(ref_.bundle_id.as_str(), "legal-bundle");
        assert_eq!(ref_.pack_id.as_str(), "legal-pack");
        assert_eq!(ref_.flow_id, "welcome");
    }

    #[test]
    fn welcome_flow_lookup_returns_none_when_unset() {
        // Both shapes that lack a welcome flow collapse to None at this site:
        // an unknown endpoint AND a known endpoint whose `welcome_flow` is None.
        // The unknown-vs-unset distinction belongs upstream at `linked_bundles`,
        // which has already refused the unknown case with `UNAUTHORIZED`.
        let ep = endpoint("teams-legal", &["legal-bundle"]); // welcome_flow: None
        let id = ep.endpoint_id.to_string();
        let admit = EndpointAdmit::from_environment(&env_with(vec![ep]));

        assert!(admit.welcome_flow(&id).is_none());
        assert!(admit.welcome_flow("bogus-endpoint-id").is_none());
    }

    #[test]
    fn endpoint_id_for_provider_returns_matching_endpoint() {
        let teams_legal = endpoint_typed("teams", "28:legal-bot", &["legal-bundle"]);
        let teams_accounting = endpoint_typed("teams", "28:acct-bot", &["acct-bundle"]);
        let slack = endpoint_typed("slack", "T0LEGAL", &["legal-bundle"]);
        let (legal_id, acct_id, slack_id) = (
            teams_legal.endpoint_id.to_string(),
            teams_accounting.endpoint_id.to_string(),
            slack.endpoint_id.to_string(),
        );
        let admit =
            EndpointAdmit::from_environment(&env_with(vec![teams_legal, teams_accounting, slack]));

        assert_eq!(
            admit.endpoint_id_for_provider("teams", "28:legal-bot"),
            Some(legal_id.as_str())
        );
        assert_eq!(
            admit.endpoint_id_for_provider("teams", "28:acct-bot"),
            Some(acct_id.as_str())
        );
        assert_eq!(
            admit.endpoint_id_for_provider("slack", "T0LEGAL"),
            Some(slack_id.as_str())
        );
    }

    #[test]
    fn endpoint_id_for_provider_returns_none_on_cross_type_or_unknown_id() {
        let admit = EndpointAdmit::from_environment(&env_with(vec![endpoint_typed(
            "teams",
            "28:legal-bot",
            &["legal-bundle"],
        )]));

        // Wrong type (Slack key against Teams entry).
        assert!(
            admit
                .endpoint_id_for_provider("slack", "28:legal-bot")
                .is_none()
        );
        // Right type, unknown provider_id.
        assert!(
            admit
                .endpoint_id_for_provider("teams", "28:unknown")
                .is_none()
        );
        // Empty inputs.
        assert!(admit.endpoint_id_for_provider("", "").is_none());
    }

    #[test]
    fn endpoint_counts_by_provider_type_reflect_declared_endpoints() {
        let admit = EndpointAdmit::from_environment(&env_with(vec![
            endpoint_typed("teams", "28:a", &["b1"]),
            endpoint_typed("teams", "28:b", &["b2"]),
            endpoint_typed("slack", "T0", &["b1"]),
        ]));

        assert_eq!(admit.endpoint_count_for_provider_type("teams"), 2);
        assert_eq!(admit.endpoint_count_for_provider_type("slack"), 1);
        assert_eq!(admit.endpoint_count_for_provider_type("telegram"), 0);

        let mut types: Vec<&str> = admit.provider_types().collect();
        types.sort_unstable();
        assert_eq!(types, vec!["slack", "teams"]);
    }

    #[test]
    fn welcome_flow_for_bundle_returns_ref_only_when_bundle_matches() {
        let mut ep = endpoint("teams-legal", &["bundle-a", "bundle-b"]);
        ep.welcome_flow = Some(WelcomeFlowRef {
            bundle_id: BundleId::new("bundle-b"),
            pack_id: PackId::new("legal-pack"),
            flow_id: "welcome".to_string(),
        });
        let id = ep.endpoint_id.to_string();
        let admit = EndpointAdmit::from_environment(&env_with(vec![ep]));

        // Dispatch to the welcome bundle ⇒ ref returned.
        let hit = admit
            .welcome_flow_for_bundle(&id, &BundleId::new("bundle-b"))
            .expect("ref present");
        assert_eq!(hit.bundle_id.as_str(), "bundle-b");

        // Dispatch to a sibling bundle ⇒ None (cross-bundle hint would
        // target B's pack/flow on A's revision).
        assert!(
            admit
                .welcome_flow_for_bundle(&id, &BundleId::new("bundle-a"))
                .is_none()
        );

        // Unknown endpoint ⇒ None regardless of bundle.
        assert!(
            admit
                .welcome_flow_for_bundle("bogus-endpoint-id", &BundleId::new("bundle-b"))
                .is_none()
        );
    }
}