// Copyright 2021 The Fuchsia Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

use {
    crate::{
        capability_source::{
            AggregateInstance, AggregateMember, AnonymizedAggregateCapabilityProvider,
            CapabilitySource, ComponentCapability, FilteredAggregateCapabilityProvider,
            FilteredAggregateCapabilityRouteData,
        },
        component_instance::{
            ComponentInstanceInterface, ExtendedInstanceInterface, ResolvedInstanceInterface,
            WeakComponentInstanceInterface,
        },
        error::RoutingError,
        legacy_router::{
            self, CapabilityVisitor, ErrorNotFoundFromParent, ErrorNotFoundInChild, ExposeVisitor,
            OfferVisitor, RouteBundle, Sources,
        },
        mapper::NoopRouteMapper,
    },
    async_trait::async_trait,
    cm_rust::{ExposeDecl, NameMapping, OfferDecl, OfferServiceDecl},
    cm_types::Name,
    derivative::Derivative,
    futures::future::BoxFuture,
    moniker::ChildName,
    std::collections::HashSet,
};

/// Provides capabilities exposed by an anonymized aggregates.
///
/// Given a set of collections and static children and the name of a capability, this provider
/// returns a list of children within them that expose the capability, and routes to a particular
/// child's exposed capability with that name.
///
/// This is used during collection routing from anonymized aggregate service instances.
#[derive(Derivative)]
#[derivative(Clone(bound = "V: Clone"))]
pub(super) struct AnonymizedAggregateServiceProvider<C: ComponentInstanceInterface, V> {
    /// Component that defines the aggregate.
    pub containing_component: WeakComponentInstanceInterface<C>,

    /// The members relative to `containing_component` that make up the aggregate.
    pub members: Vec<AggregateMember>,

    /// Name of the capability as exposed by children in the collection.
    pub capability_name: Name,

    pub capability_type: cm_rust::CapabilityTypeName,

    pub sources: Sources,
    pub visitor: V,
}

#[async_trait]
impl<C, V> AnonymizedAggregateCapabilityProvider<C> for AnonymizedAggregateServiceProvider<C, V>
where
    C: ComponentInstanceInterface + 'static,
    V: OfferVisitor,
    V: ExposeVisitor,
    V: CapabilityVisitor,
    V: Clone + Send + Sync + 'static,
{
    /// Returns a list of instances contributing capabilities to this provider.
    ///
    /// In the case of service capabilities, they are *not* instances inside that service, but
    /// rather service capabilities with the same name that are exposed by different components.
    async fn list_instances(&self) -> Result<Vec<AggregateInstance>, RoutingError> {
        let mut instances = Vec::new();
        let component = self.containing_component.upgrade()?;
        let mut child_components = vec![];
        let mut parent = None;
        let mut include_self = false;
        {
            let resolved_state = component.lock_resolved_state().await?;
            for s in &self.members {
                match s {
                    AggregateMember::Child(child_name) => {
                        if let Some(child) = resolved_state.get_child(&child_name) {
                            child_components.push((child_name.clone(), child.clone()));
                        }
                    }
                    AggregateMember::Collection(collection) => {
                        child_components.extend(resolved_state.children_in_collection(&collection));
                    }
                    AggregateMember::Parent => {
                        if let Ok(p) = component.try_get_parent() {
                            match p {
                                ExtendedInstanceInterface::AboveRoot(_) => {
                                    // Services from above parent are not supported.
                                }
                                ExtendedInstanceInterface::Component(p) => {
                                    parent = Some(p);
                                }
                            }
                        }
                    }
                    AggregateMember::Self_ => {
                        include_self = true;
                    }
                }
            }
        }
        for (child_name, child_component) in child_components {
            let child_exposes = child_component.lock_resolved_state().await.map(|c| c.exposes());
            match child_exposes {
                Ok(child_exposes) => {
                    if let Some(_) = legacy_router::find_matching_exposes(
                        self.capability_type,
                        &self.capability_name,
                        &child_exposes,
                    ) {
                        instances.push(AggregateInstance::Child(child_name.clone()));
                    }
                }
                // Ignore errors. One misbehaving component should not affect the entire collection.
                Err(_) => {}
            }
        }
        if let Some(parent) = parent {
            let parent_offers = parent.lock_resolved_state().await.map(|c| c.offers());
            match parent_offers {
                Ok(parent_offers) => {
                    let child_moniker = component.child_moniker().expect("ChildName should exist");
                    if let Some(_) = legacy_router::find_matching_offers(
                        self.capability_type,
                        &self.capability_name,
                        &child_moniker,
                        &parent_offers,
                    ) {
                        instances.push(AggregateInstance::Parent);
                    }
                }
                // Ignore errors. One misbehaving component should not affect the entire collection.
                Err(_) => {}
            }
        }
        if include_self {
            instances.push(AggregateInstance::Self_);
        }
        Ok(instances)
    }

    /// Returns a `CapabilitySource` to a capability exposed by a child.
    ///
    /// `instance` is the name of the child that exposes the capability, as returned by
    /// `list_instances`.
    async fn route_instance(
        &self,
        instance: &AggregateInstance,
    ) -> Result<CapabilitySource<C>, RoutingError> {
        match instance {
            AggregateInstance::Child(name) => self.route_child_instance(&name).await,
            AggregateInstance::Parent => self.route_parent_instance().await,
            AggregateInstance::Self_ => self.route_self_instance().await,
        }
    }

    fn clone_boxed(&self) -> Box<dyn AnonymizedAggregateCapabilityProvider<C>> {
        Box::new(self.clone())
    }
}

impl<C, V> AnonymizedAggregateServiceProvider<C, V>
where
    C: ComponentInstanceInterface + 'static,
    V: OfferVisitor,
    V: ExposeVisitor,
    V: CapabilityVisitor,
    V: Clone + Send + Sync + 'static,
{
    async fn route_child_instance(
        &self,
        instance: &ChildName,
    ) -> Result<CapabilitySource<C>, RoutingError> {
        let containing_component = self.containing_component.upgrade()?;
        let child_component =
            containing_component.lock_resolved_state().await?.get_child(instance).ok_or_else(
                || RoutingError::OfferFromChildInstanceNotFound {
                    child_moniker: instance.clone(),
                    moniker: containing_component.moniker().clone(),
                    capability_id: self.capability_name.clone().into(),
                },
            )?;

        let child_exposes = child_component.lock_resolved_state().await?.exposes();
        let child_exposes = legacy_router::find_matching_exposes(
            self.capability_type,
            &self.capability_name,
            &child_exposes,
        )
        .ok_or_else(|| {
            ExposeDecl::error_not_found_in_child(
                containing_component.moniker().clone(),
                instance.clone(),
                self.capability_name.clone(),
            )
        })?;
        legacy_router::route_from_expose(
            child_exposes,
            child_component,
            self.sources.clone(),
            &mut self.visitor.clone(),
            &mut NoopRouteMapper,
        )
        .await
    }

    async fn route_parent_instance(&self) -> Result<CapabilitySource<C>, RoutingError> {
        let containing_component = self.containing_component.upgrade()?;
        let parent = match containing_component.try_get_parent().map_err(|_| {
            RoutingError::OfferFromParentNotFound {
                moniker: containing_component.moniker().clone(),
                capability_id: self.capability_name.clone().into(),
            }
        })? {
            ExtendedInstanceInterface::AboveRoot(_) => {
                return Err(RoutingError::unsupported_route_source("service above parent"));
            }
            ExtendedInstanceInterface::Component(p) => p,
        };

        let child_moniker = containing_component.child_moniker().expect("ChildName should exist");
        let parent_offers = parent.lock_resolved_state().await?.offers();
        let parent_offers = legacy_router::find_matching_offers(
            self.capability_type,
            &self.capability_name,
            &child_moniker,
            &parent_offers,
        )
        .ok_or_else(|| {
            OfferDecl::error_not_found_from_parent(
                containing_component.moniker().clone(),
                self.capability_name.clone(),
            )
        })?;
        legacy_router::route_from_offer(
            parent_offers,
            parent,
            self.sources.clone(),
            &mut self.visitor.clone(),
            &mut NoopRouteMapper,
        )
        .await
    }

    async fn route_self_instance(&self) -> Result<CapabilitySource<C>, RoutingError> {
        let containing_component = self.containing_component.upgrade()?;
        legacy_router::route_from_self_by_name(
            &self.capability_name,
            containing_component,
            self.sources.clone(),
            &mut self.visitor.clone(),
            &mut NoopRouteMapper,
        )
        .await
    }
}

#[derive(Derivative)]
#[derivative(Clone(bound = ""))]
pub(super) struct OfferFilteredServiceProvider<C: ComponentInstanceInterface> {
    /// Component that offered the filtered service
    component: WeakComponentInstanceInterface<C>,
    /// The service capability
    capability: ComponentCapability,
    /// The service offer that has a filter.
    offer_decl: OfferServiceDecl,
}

impl<C> OfferFilteredServiceProvider<C>
where
    C: ComponentInstanceInterface + 'static,
{
    pub(super) fn new(
        offer_decl: OfferServiceDecl,
        component: WeakComponentInstanceInterface<C>,
        capability: ComponentCapability,
    ) -> Self {
        Self { offer_decl, component, capability }
    }
}

impl<C> FilteredAggregateCapabilityProvider<C> for OfferFilteredServiceProvider<C>
where
    C: ComponentInstanceInterface + 'static,
{
    fn route_instances(
        &self,
    ) -> Vec<BoxFuture<'_, Result<FilteredAggregateCapabilityRouteData<C>, RoutingError>>> {
        let capability_source = CapabilitySource::Component {
            capability: self.capability.clone(),
            component: self.component.clone(),
        };
        let instance_filter = get_instance_filter(&self.offer_decl);
        let fut = async move {
            Ok(FilteredAggregateCapabilityRouteData { capability_source, instance_filter })
        };
        // Without the explicit type, this does not compile
        let mut out: Vec<
            BoxFuture<'_, Result<FilteredAggregateCapabilityRouteData<C>, RoutingError>>,
        > = vec![];
        out.push(Box::pin(fut));
        out
    }

    fn clone_boxed(&self) -> Box<dyn FilteredAggregateCapabilityProvider<C>> {
        Box::new(self.clone())
    }
}

fn get_instance_filter(offer_decl: &OfferServiceDecl) -> Vec<NameMapping> {
    let renamed_instances = offer_decl.renamed_instances.as_ref().unwrap_or_else(|| {
        static EMPTY_VEC: Vec<NameMapping> = vec![];
        &EMPTY_VEC
    });
    if !renamed_instances.is_empty() {
        let source_instance_filter: HashSet<_> = offer_decl
            .source_instance_filter
            .as_ref()
            .unwrap_or_else(|| {
                static EMPTY_VEC: Vec<Name> = vec![];
                &EMPTY_VEC
            })
            .iter()
            .map(|s| s.as_str())
            .collect();
        renamed_instances
            .clone()
            .into_iter()
            .filter_map(|m| {
                if source_instance_filter.is_empty()
                    || source_instance_filter.contains(&m.target_name.as_str())
                {
                    Some(NameMapping { source_name: m.source_name, target_name: m.target_name })
                } else {
                    None
                }
            })
            .collect()
    } else {
        offer_decl
            .source_instance_filter
            .clone()
            .unwrap_or_default()
            .into_iter()
            .map(|n| NameMapping { source_name: n.clone(), target_name: n })
            .collect()
    }
}

#[derive(Derivative)]
#[derivative(Clone(bound = "V: Clone"))]
pub(super) struct OfferAggregateServiceProvider<C: ComponentInstanceInterface, V> {
    /// Component that offered the aggregate service
    component: WeakComponentInstanceInterface<C>,

    /// List of offer decl to follow for routing each service provider used in the overall aggregation
    offer_decls: Vec<OfferServiceDecl>,

    sources: Sources,
    visitor: V,
}

impl<C, V> OfferAggregateServiceProvider<C, V>
where
    C: ComponentInstanceInterface + 'static,
    V: OfferVisitor + ExposeVisitor + CapabilityVisitor,
    V: Send + Sync + Clone + 'static,
{
    pub(super) fn new(
        offer_decls: Vec<OfferServiceDecl>,
        component: WeakComponentInstanceInterface<C>,
        sources: Sources,
        visitor: V,
    ) -> Self {
        Self { offer_decls, sources, visitor, component }
    }
}

impl<C, V> FilteredAggregateCapabilityProvider<C> for OfferAggregateServiceProvider<C, V>
where
    C: ComponentInstanceInterface + 'static,
    V: OfferVisitor + ExposeVisitor + CapabilityVisitor,
    V: Send + Sync + Clone + 'static,
{
    fn route_instances(
        &self,
    ) -> Vec<BoxFuture<'_, Result<FilteredAggregateCapabilityRouteData<C>, RoutingError>>> {
        // Without the explicit type, this does not compile
        let mut out: Vec<
            BoxFuture<'_, Result<FilteredAggregateCapabilityRouteData<C>, RoutingError>>,
        > = vec![];
        for offer_decl in &self.offer_decls {
            let instance_filter = get_instance_filter(offer_decl);
            if instance_filter.is_empty() {
                continue;
            }
            let mut offer_decl = offer_decl.clone();
            offer_decl.source_instance_filter = None;
            offer_decl.renamed_instances = None;
            let offer_decl = OfferDecl::Service(offer_decl);
            let fut = async {
                let component = self.component.upgrade().map_err(|e| {
                    RoutingError::unsupported_route_source(format!(
                        "error upgrading aggregation point component {}",
                        e
                    ))
                })?;
                let capability_source = legacy_router::route_from_offer(
                    RouteBundle::from_offer(offer_decl),
                    component,
                    self.sources.clone(),
                    &mut self.visitor.clone(),
                    &mut NoopRouteMapper,
                )
                .await?;
                Ok(FilteredAggregateCapabilityRouteData { capability_source, instance_filter })
            };
            out.push(Box::pin(fut));
        }
        out
    }

    fn clone_boxed(&self) -> Box<dyn FilteredAggregateCapabilityProvider<C>> {
        Box::new(self.clone())
    }
}

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

    #[test]
    fn test_get_instance_filter() {
        fn get_instance_filter(
            source_instance_filter: Option<Vec<Name>>,
            renamed_instances: Option<Vec<NameMapping>>,
        ) -> Vec<NameMapping> {
            let offer = OfferBuilder::service()
                .name("foo")
                .target_name("bar")
                .source(cm_rust::OfferSource::Parent)
                .target_static_child("a")
                .build();
            let OfferDecl::Service(mut offer) = offer else {
                unreachable!();
            };
            offer.source_instance_filter = source_instance_filter;
            offer.renamed_instances = renamed_instances;
            super::get_instance_filter(&offer)
        }

        assert_eq!(get_instance_filter(None, None), vec![]);
        assert_eq!(get_instance_filter(Some(vec![]), Some(vec![])), vec![]);
        let same_name_map = vec![
            NameMapping { source_name: "a".parse().unwrap(), target_name: "a".parse().unwrap() },
            NameMapping { source_name: "b".parse().unwrap(), target_name: "b".parse().unwrap() },
        ];
        assert_eq!(
            get_instance_filter(Some(vec!["a".parse().unwrap(), "b".parse().unwrap()]), None),
            same_name_map
        );
        assert_eq!(
            get_instance_filter(
                Some(vec!["a".parse().unwrap(), "b".parse().unwrap()]),
                Some(vec![])
            ),
            same_name_map
        );
        let renamed_map = vec![
            NameMapping { source_name: "one".parse().unwrap(), target_name: "a".parse().unwrap() },
            NameMapping { source_name: "two".parse().unwrap(), target_name: "b".parse().unwrap() },
        ];
        assert_eq!(get_instance_filter(None, Some(renamed_map.clone())), renamed_map);
        assert_eq!(get_instance_filter(Some(vec![]), Some(renamed_map.clone())), renamed_map);
        assert_eq!(
            get_instance_filter(Some(vec!["b".parse().unwrap()]), Some(renamed_map.clone())),
            vec![NameMapping {
                source_name: "two".parse().unwrap(),
                target_name: "b".parse().unwrap()
            }]
        );
    }
}