alkcall 0.8.0

//! The hub-leg install template (ADR-051 §5; review 008 Unit 3b).
//!
//! A relay hub accepts consumer-leg `alk/channels` connections and
//! serves channel 0 over a per-connection fork of its base registry:
//! the fork carries the generic channel ops, the bootstrap discovery
//! ops (closed over the fork, so `services/list` sees the re-exposed
//! ops — review 004 F-06), the stashed plain bundles as-is (the
//! from_call forwarding stubs), and each stashed marked spec via the
//! relay registration ([`ChannelRelay::register_relay_openable`]).
//! The template exports that composition as an in-tree
//! [`InstallChannelZero`] hook — the composition every hub-side test
//! hand-rolled before this module existed.
//!
//! The spoke side needs nothing new: a spoke serving ops through a hub
//! is the existing connect-side serving shape
//! (`ChannelClient::from_connection_with_serving`) plus the
//! producer-leg registration it already does.
//!
//! Two-phase seam (ADR-051 §4): discovery of the producer leg's ops
//! happens once (the [`HubLegImports`] stash, `Clone`, one discovered
//! set serves any number of consumer legs); registration happens per
//! consumer-leg connection inside the install hook. Per-consumer
//! op-subset filtering composes by filtering the stash before handing
//! it to the template (the fork is per consumer leg); ACL layering per
//! ADR-051 §5: the imported spec's own `AccessControl` gates the
//! re-exposed op on the consumer leg, the spoke's ACL sees only the
//! hub identity, the end consumer's identity rides `forwarded_for` as
//! metadata and is never consulted by any `AccessControl::check`.

use std::collections::HashSet;
use std::sync::Arc;

use crate::core::auth::{Identity, IdentityProvider};
use crate::protocol::connection::{split_single_stream, CallConnection};
use crate::protocol::dispatch::Dispatcher;
use crate::registry::discovery::install_bootstrap_discovery;
use crate::registry::registration::{HandlerRegistration, OperationRegistry};

use super::operations::ChannelCore;
use super::relay::ChannelRelay;
use super::{adapter::InstallChannelZero, policy::ChannelLifecyclePolicy};

/// The discover/stash half of the two-phase registration seam
/// (ADR-051 §4 phase 1): the from_call-imported bundles split by the
/// `channel_open` marker — marked specs become relay openables, plain
/// bundles register as-is (the forwarding stubs). `Clone`, so one
/// discovered set serves any number of consumer legs (and any number
/// of per-consumer filtered variants of it).
#[derive(Clone, Default)]
pub struct HubLegImports {
    /// Marked specs (reconstructed WITH the `channel_open` marker,
    /// ADR-047 amendment 3) — registered per consumer leg via the
    /// relay registration.
    marked: Vec<HandlerRegistration>,
    /// Plain bundles — registered per consumer leg as-is.
    plain: Vec<HandlerRegistration>,
}

impl HubLegImports {
    /// Split from_call-imported bundles by the marker. A discovered
    /// `Pub`-typed marked spec is kept here (the stash is data, not
    /// assembly) — [`HubLegTemplate::install_hook`] surfaces it as the
    /// loud assembly error (ADR-051 §6) when a consumer leg installs.
    pub fn from_bundles(bundles: Vec<HandlerRegistration>) -> Self {
        let mut imports = Self::default();
        for bundle in bundles {
            if bundle.spec.channel_open.is_some() {
                imports.marked.push(bundle);
            } else {
                imports.plain.push(bundle);
            }
        }
        imports
    }

    /// The marked specs (relay openables).
    pub fn marked(&self) -> &[HandlerRegistration] {
        &self.marked
    }

    /// The plain bundles (forwarding stubs).
    pub fn plain(&self) -> &[HandlerRegistration] {
        &self.plain
    }

    /// Keep only the ops whose names satisfy `keep` — the per-consumer
    /// op-subset filter (ADR-051 §4's composition note; a hub
    /// re-exposing different op subsets to different consumers filters
    /// the stash per fork).
    #[must_use]
    pub fn filtered(self, keep: impl Fn(&str) -> bool) -> Self {
        Self {
            marked: self
                .marked
                .into_iter()
                .filter(|b| keep(&b.spec.name))
                .collect(),
            plain: self
                .plain
                .into_iter()
                .filter(|b| keep(&b.spec.name))
                .collect(),
        }
    }

    /// Keep only the named ops (the common filter shape).
    #[must_use]
    pub fn only(self, names: &[&str]) -> Self {
        let allowed: HashSet<&str> = names.iter().copied().collect();
        self.filtered(|name| allowed.contains(name))
    }
}

/// The hub-leg install template (ADR-051 §5): composes the fork +
/// generic channel ops + bootstrap discovery + plain bundles + relay
/// openables + serving identity, and returns the
/// [`InstallChannelZero`] hook to hand [`super::adapter::ChannelsAdapter`].
///
/// The producer-leg surface is shared (ADR-051 §2): the
/// [`ChannelRelay`] (holding `Arc<CallConnection>` + producer-leg
/// `ChannelManager`) plus the consumer-leg `ChannelCore` are closed
/// over per install; the template holds no extra claim on the
/// producer leg.
pub struct HubLegTemplate {
    relay: Arc<ChannelRelay>,
    imports: HubLegImports,
    /// The consumer-leg channel lifecycle policy — the per-identity
    /// cap the consumer leg's open-op wrapper enforces. Shared across
    /// consumer legs (per-identity, not per-connection, ADR-041).
    policy: Arc<dyn ChannelLifecyclePolicy>,
    /// Serving-identity resolution for the dispatch (CF-005
    /// precedence: payload token → explicit override → transport
    /// identity). `identity` mirrors
    /// [`crate::channels::client::ServingConfig::identity`] — an
    /// explicit override for the peer the transport authenticated;
    /// `None` falls through to the channel-0 connection's identity.
    identity_provider: Arc<dyn IdentityProvider>,
    identity: Option<Identity>,
}

impl HubLegTemplate {
    /// Construct the template over a relay (the producer leg) and the
    /// stashed imports. The policy defaults to the crate default
    /// (256/identity, ADR-041); identity resolution defaults to the
    /// noop provider with no override.
    pub fn new(relay: ChannelRelay, imports: HubLegImports) -> Self {
        Self {
            relay: Arc::new(relay),
            imports,
            policy: super::policy::default_policy(),
            identity_provider: Arc::new(crate::core::auth::NoopIdentityProvider),
            identity: None,
        }
    }

    /// Set the consumer-leg channel lifecycle policy.
    pub fn with_policy(mut self, policy: Arc<dyn ChannelLifecyclePolicy>) -> Self {
        self.policy = policy;
        self
    }

    /// Set the identity provider for the serving dispatch (payload
    /// `auth_token` resolution, ADR-017 §7).
    pub fn with_identity_provider(mut self, provider: Arc<dyn IdentityProvider>) -> Self {
        self.identity_provider = provider;
        self
    }

    /// Set an explicit serving identity (the CF-005 seam — wins over
    /// the transport connection's identity).
    pub fn with_identity(mut self, identity: Identity) -> Self {
        self.identity = Some(identity);
        self
    }

    /// Build the `install_channel_zero` hook (ADR-051 §5 — the
    /// composition the hub/spoke family all share). Per consumer-leg
    /// connection:
    ///
    /// 1. Fork the (empty) per-leg base — the fork is the dispatch
    ///    registry, registered fresh per connection because the
    ///    openables close over that connection's `ChannelCore`
    ///    (ADR-047 §4 amendment).
    /// 2. Register the generic channel ops
    ///    (`ChannelOperations::register_on`, closed over the consumer
    ///    leg's manager + policy).
    /// 3. Register the stashed plain bundles as-is.
    /// 4. Register each stashed marked spec via
    ///    [`ChannelRelay::register_relay_openable`] on the consumer
    ///    leg's `ChannelCore` (the full open-op wrapper machinery —
    ///    ACL, cap/ledger, establishment bound, teardown). A
    ///    Pub-typed marked spec surfaces here as the loud assembly
    ///    error (ADR-051 §6) — the install task ends and the leg's
    ///    channel 0 never dispatches (never a silent stub).
    /// 5. Install the bootstrap discovery ops closed over the fork
    ///    (review 004 F-06 — `services/list` sees the re-exposed ops).
    /// 6. Resolve the serving identity (payload token → explicit
    ///    override → transport identity) and run the single-stream
    ///    dispatch loop until the transport EOF.
    pub fn install_hook(&self) -> InstallChannelZero {
        let relay = Arc::clone(&self.relay);
        let imports = self.imports.clone();
        let policy = Arc::clone(&self.policy);
        let identity_provider = Arc::clone(&self.identity_provider);
        let identity = self.identity.clone();
        Arc::new(move |consumer_manager, channel0_conn, auth| {
            let relay = Arc::clone(&relay);
            let imports = imports.clone();
            let policy = Arc::clone(&policy);
            let identity_provider = Arc::clone(&identity_provider);
            let identity = identity.clone();
            tokio::spawn(async move {
                // The CF-005 seam (review 004 F-05 / 0.7.0): an
                // explicit override wins; else the transport
                // connection's identity propagates to channel 0 and
                // the dispatch's fallback resolves through it.
                // `set_identity` is once-only; silently skip when the
                // adapter pre-set one.
                if let Some(identity) = identity {
                    let _ = channel0_conn.set_identity(identity);
                }
                let Ok(channel0_bidi) = channel0_conn.accept_bi().await else {
                    return;
                };
                let (writer, reader) = split_single_stream(channel0_bidi);
                let registry = Arc::new(OperationRegistry::new());

                let operations = super::operations::ChannelOperations::new(
                    consumer_manager.clone(),
                    Arc::clone(&policy),
                );
                if let Err(e) = operations.register_on(&registry) {
                    tracing::warn!(error = %e, "hub-leg template: generic channel ops failed to register");
                    return;
                }

                let consumer_core = ChannelCore::new(consumer_manager, Arc::clone(&policy));
                for bundle in &imports.plain {
                    // The bootstrap discovery ops the spoke serves are
                    // re-discovered by from_call like any plain op;
                    // the template's own install (closed over this
                    // fork, review 004 F-06) supersedes the imported
                    // copies — re-registering both would double-book
                    // the names (the fork's discovery must see the
                    // fork's ops, not the spoke's).
                    if crate::registry::discovery::BOOTSTRAP_DISCOVERY_OPS
                        .contains(&bundle.spec.name.as_str())
                    {
                        continue;
                    }
                    if let Err(e) = registry.register(bundle.clone()) {
                        tracing::warn!(error = %e, "hub-leg template: plain bundle registration failed");
                        return;
                    }
                }
                let install_auth = auth.clone();
                for bundle in &imports.marked {
                    if let Err(e) = relay.register_relay_openable(
                        &consumer_core,
                        &registry,
                        bundle.spec.clone(),
                        install_auth.clone(),
                    ) {
                        // The loud assembly posture (ADR-051 §6): a
                        // Pub-typed (or otherwise unregistrable)
                        // marked spec never becomes a silent stub —
                        // the leg's channel 0 never dispatches.
                        tracing::warn!(error = %e, "hub-leg template: relay openable registration failed");
                        return;
                    }
                }

                if let Err(e) = install_bootstrap_discovery(&registry) {
                    tracing::warn!(error = %e, "hub-leg template: bootstrap discovery install failed");
                    return;
                }

                let call_connection = Arc::new(CallConnection::new_single_stream(
                    channel0_conn,
                    Arc::clone(&writer),
                ));
                Dispatcher::new(registry, identity_provider)
                    .run_loop_single_stream(call_connection, reader, writer)
                    .await;
            })
        })
    }
}

#[cfg(test)]
#[path = "hub_leg_tests.rs"]
mod tests;

#[cfg(test)]
#[path = "gate2_tests.rs"]
mod gate2_tests;