// Copyright 2020 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::model::{
        events::{
            event::{Event, SyncMode},
            filter::EventFilter,
            registry::{ExecutionMode, SubscriptionOptions, SubscriptionType},
        },
        hooks::{
            Event as ComponentEvent, EventError, EventErrorPayload, EventPayload, HasEventType,
            TransferEvent,
        },
    },
    anyhow::Error,
    cm_rust::DictionaryValue,
    fuchsia_trace as trace,
    futures::{
        channel::{mpsc, oneshot},
        lock::Mutex,
        sink::SinkExt,
    },
    maplit::hashmap,
    moniker::AbsoluteMoniker,
};

/// EventDispatcher and EventStream are two ends of a channel.
///
/// EventDispatcher represents the sending end of the channel.
///
/// An EventDispatcher receives events of a particular event type,
/// and dispatches though events out to the EventStream if they fall within
/// one of the scopes associated with the dispatcher.
///
/// EventDispatchers are owned by EventStreams. If an EventStream is dropped,
/// all corresponding EventDispatchers are dropped.
///
/// An EventStream is owned by the client - usually a test harness or a
/// EventSource. It receives a Event from an EventDispatcher and propagates it
/// to the client.
pub struct EventDispatcher {
    /// The options used to generate this EventDispatcher.
    options: SubscriptionOptions,

    /// Specifies the realms that this EventDispatcher can dispatch events from and under what
    /// conditions.
    scopes: Vec<EventDispatcherScope>,

    /// An `mpsc::Sender` used to dispatch an event. Note that this
    /// `mpsc::Sender` is wrapped in an Mutex<..> to allow it to be passed along
    /// to other tasks for dispatch.
    tx: Mutex<mpsc::UnboundedSender<Event>>,
}

impl EventDispatcher {
    pub fn new(
        options: SubscriptionOptions,
        scopes: Vec<EventDispatcherScope>,
        tx: mpsc::UnboundedSender<Event>,
    ) -> Self {
        // TODO(fxbug.dev/48360): flatten scope_monikers. There might be monikers that are
        // contained within another moniker in the list.
        Self { options, scopes, tx: Mutex::new(tx) }
    }

    /// Sends the event to an event stream, if fired in the scope of `scope_moniker`. Returns
    /// a responder which can be blocked on.
    pub async fn dispatch(
        &self,
        event: &ComponentEvent,
    ) -> Result<Option<oneshot::Receiver<()>>, Error> {
        let maybe_scope = self.find_scope(&event);
        if maybe_scope.is_none() {
            return Ok(None);
        }

        let scope_moniker = maybe_scope.unwrap().moniker.clone();

        trace::duration!("component_manager", "events:send");
        let event_type = format!("{:?}", event.event_type());
        let target_moniker = event.target_moniker.to_string();
        trace::flow_begin!(
            "component_manager",
            "event",
            event.id,
            "event_type" => event_type.as_str(),
            "target_moniker" => target_moniker.as_str()
        );

        let (maybe_responder_tx, maybe_responder_rx) = match self.options.sync_mode {
            SyncMode::Async => (None, None),
            SyncMode::Sync => {
                let (responder_tx, responder_rx) = oneshot::channel();
                (Some(responder_tx), Some(responder_rx))
            }
        };

        {
            let mut tx = self.tx.lock().await;
            tx.send(Event {
                event: event.transfer().await,
                scope_moniker,
                responder: maybe_responder_tx,
            })
            .await?;
        }
        Ok(maybe_responder_rx)
    }

    fn find_scope(&self, event: &ComponentEvent) -> Option<&EventDispatcherScope> {
        // TODO(fxbug.dev/48360): once flattening of monikers is done, we would expect to have a single
        // moniker here. For now taking the first one and ignoring the rest.
        // Ensure that the event is coming from a realm within the scope of this dispatcher and
        // matching the path filter if one exists.
        self.scopes.iter().filter(|scope| scope.contains(&self.options, &event)).next()
    }
}

/// A scope for dispatching and filters on that scope.
#[derive(Debug, Clone, Eq, PartialEq)]
pub struct EventDispatcherScope {
    /// The moniker of the realm
    pub moniker: AbsoluteMoniker,

    /// Filters for an event in that realm.
    pub filter: EventFilter,
}

impl EventDispatcherScope {
    pub fn new(moniker: AbsoluteMoniker) -> Self {
        Self { moniker, filter: EventFilter::new(None) }
    }

    pub fn with_filter(mut self, filter: EventFilter) -> Self {
        self.filter = filter;
        self
    }

    /// For the top-level EventStreams and event strems used in unit tests in the c_m codebase we
    /// don't take filters into account.
    pub fn for_debug(mut self) -> Self {
        self.filter = EventFilter::debug();
        self
    }

    /// Given the provided options, indicates whether or not the event is contained
    /// in this scope.
    pub fn contains(&self, options: &SubscriptionOptions, event: &ComponentEvent) -> bool {
        let in_scope = match &event.result {
            Ok(EventPayload::CapabilityRequested { source_moniker, .. })
            | Err(EventError {
                event_error_payload: EventErrorPayload::CapabilityRequested { source_moniker, .. },
                ..
            }) => match &options.subscription_type {
                SubscriptionType::AboveRoot => true,
                SubscriptionType::Component(target) => *source_moniker == *target,
            },
            // CapabilityRouted events are only dispatched when component manager runs
            // in debug mode.
            Ok(EventPayload::CapabilityRouted { .. })
            | Err(EventError {
                event_error_payload: EventErrorPayload::CapabilityRouted { .. },
                ..
            }) => match &options.execution_mode {
                ExecutionMode::Debug => self.moniker.contains_in_realm(&event.target_moniker),
                ExecutionMode::Production => false,
            },
            _ => self.moniker.contains_in_realm(&event.target_moniker),
        };

        if !in_scope {
            return false;
        }

        // TODO(fsamuel): Creating hashmaps on every lookup is not ideal, but in practice this
        // likely doesn't happen too often.
        let filterable_fields = match &event.result {
            Ok(EventPayload::CapabilityRequested { name, .. }) => Some(hashmap! {
                "name".to_string() => DictionaryValue::Str(name.into())
            }),
            Ok(EventPayload::CapabilityReady { name, .. }) => Some(hashmap! {
                "name".to_string() => DictionaryValue::Str(name.into())
            }),
            Err(EventError {
                event_error_payload: EventErrorPayload::CapabilityRequested { name, .. },
                ..
            }) => Some(hashmap! {
                "name".to_string() => DictionaryValue::Str(name.into())
            }),
            Err(EventError {
                event_error_payload: EventErrorPayload::CapabilityReady { name, .. },
                ..
            }) => Some(hashmap! {
                "name".to_string() => DictionaryValue::Str(name.into())
            }),
            _ => None,
        };
        self.filter.has_fields(&filterable_fields)
    }
}

#[cfg(test)]
mod tests {
    use {
        super::*,
        crate::{
            capability::{CapabilitySource, InternalCapability},
            model::events::registry::ExecutionMode,
        },
        fuchsia_zircon as zx,
        futures::StreamExt,
        matches::assert_matches,
        std::{convert::TryInto, sync::Arc},
    };

    struct EventDispatcherFactory {
        /// The receiving end of a channel of Events.
        rx: mpsc::UnboundedReceiver<Event>,

        /// The sending end of a channel of Events.
        tx: mpsc::UnboundedSender<Event>,
    }

    impl EventDispatcherFactory {
        fn new() -> Self {
            let (tx, rx) = mpsc::unbounded();
            Self { rx, tx }
        }

        /// Receives the next event from the sender.
        pub async fn next_event(&mut self) -> Option<ComponentEvent> {
            self.rx.next().await.map(|e| e.event)
        }

        fn create_dispatcher(&self, options: SubscriptionOptions) -> Arc<EventDispatcher> {
            let scopes = vec![EventDispatcherScope::new(AbsoluteMoniker::root()).for_debug()];
            Arc::new(EventDispatcher::new(options, scopes, self.tx.clone()))
        }
    }

    async fn dispatch_capability_requested_event(
        dispatcher: &EventDispatcher,
        source_moniker: &AbsoluteMoniker,
    ) -> Option<oneshot::Receiver<()>> {
        let (_, capability_server_end) = zx::Channel::create().unwrap();
        let capability_server_end = Arc::new(Mutex::new(Some(capability_server_end)));
        let event = ComponentEvent::new_for_test(
            AbsoluteMoniker::root(),
            "fuchsia-pkg://root/a/b/c",
            Ok(EventPayload::CapabilityRequested {
                source_moniker: source_moniker.clone(),
                name: "foo".to_string(),
                capability: capability_server_end,
            }),
        );
        dispatcher.dispatch(&event).await.ok().flatten()
    }

    async fn dispatch_capability_routed_event(
        dispatcher: &EventDispatcher,
    ) -> Option<oneshot::Receiver<()>> {
        let empty_capability_provider = Arc::new(Mutex::new(None));
        let event = ComponentEvent::new_for_test(
            AbsoluteMoniker::root(),
            "fuchsia-pkg://root/a/b/c",
            Ok(EventPayload::CapabilityRouted {
                source: CapabilitySource::Builtin {
                    capability: InternalCapability::Protocol(
                        "fuchsia.sys2.MyAwesomeProtocol".try_into().unwrap(),
                    ),
                },
                capability_provider: empty_capability_provider,
            }),
        );
        dispatcher.dispatch(&event).await.ok().flatten()
    }

    // This test verifies that the CapabilityRequested event can only be sent to a source
    // that matches its source moniker.
    #[fuchsia_async::run_singlethreaded(test)]
    async fn can_send_capability_requested_to_source() {
        // Verify we can dispatch to a debug source.
        // Sync events get a responder if the message was dispatched.
        let options = SubscriptionOptions::new(
            SubscriptionType::AboveRoot,
            SyncMode::Async,
            ExecutionMode::Production,
        );
        let mut factory = EventDispatcherFactory::new();
        let dispatcher = factory.create_dispatcher(options);
        let source_moniker = vec!["root:0", "a:0", "b:0", "c:0"].into();
        assert!(dispatch_capability_requested_event(&dispatcher, &source_moniker).await.is_none());
        assert_matches!(
            factory.next_event().await,
            Some(ComponentEvent { result: Ok(EventPayload::CapabilityRequested { .. }), .. })
        );

        // Verify that we cannot dispatch the CapabilityRequested event to the root component.
        let options = SubscriptionOptions::new(
            SubscriptionType::Component(vec!["root:0"].into()),
            SyncMode::Sync,
            ExecutionMode::Production,
        );
        let dispatcher = factory.create_dispatcher(options);
        assert!(dispatch_capability_requested_event(&dispatcher, &source_moniker).await.is_none());

        // Verify that we cannot dispatch the CapabilityRequested event to the root:0/a:0 component.
        let options = SubscriptionOptions::new(
            SubscriptionType::Component(vec!["root:0", "a:0"].into()),
            SyncMode::Sync,
            ExecutionMode::Production,
        );
        let dispatcher = factory.create_dispatcher(options);
        assert!(dispatch_capability_requested_event(&dispatcher, &source_moniker).await.is_none());

        // Verify that we cannot dispatch the CapabilityRequested event to the root:0/a:0/b:0 component.
        let options = SubscriptionOptions::new(
            SubscriptionType::Component(vec!["root:0", "a:0", "b:0"].into()),
            SyncMode::Sync,
            ExecutionMode::Production,
        );
        let dispatcher = factory.create_dispatcher(options);
        assert!(dispatch_capability_requested_event(&dispatcher, &source_moniker).await.is_none());

        // Verify that we CAN dispatch the CapabilityRequested event to the root:0/a:0/b:0/c:0 component.
        let options = SubscriptionOptions::new(
            SubscriptionType::Component(vec!["root:0", "a:0", "b:0", "c:0"].into()),
            SyncMode::Sync,
            ExecutionMode::Production,
        );
        let dispatcher = factory.create_dispatcher(options);
        assert!(dispatch_capability_requested_event(&dispatcher, &source_moniker).await.is_some());
        assert_matches!(
            factory.next_event().await,
            Some(ComponentEvent { result: Ok(EventPayload::CapabilityRequested { .. }), .. })
        );

        // Verify that we cannot dispatch the CapabilityRequested event to the root:0/a:0/b:0/c:0/d:0 component.
        let options = SubscriptionOptions::new(
            SubscriptionType::Component(vec!["root:0", "a:0", "b:0", "c:0", "d:0"].into()),
            SyncMode::Sync,
            ExecutionMode::Production,
        );
        let dispatcher = factory.create_dispatcher(options);
        assert!(dispatch_capability_requested_event(&dispatcher, &source_moniker).await.is_none());
    }

    // This test verifies that the CapabilityRouted event can only be sent in Debug mode and
    // not in production.
    #[fuchsia_async::run_singlethreaded(test)]
    async fn cannot_send_capability_routed_in_production() {
        let mut factory = EventDispatcherFactory::new();

        // Verify we can dispatch a CapabilityRouted event in debug mode.
        let options = SubscriptionOptions::new(
            SubscriptionType::AboveRoot,
            SyncMode::Sync,
            ExecutionMode::Debug,
        );
        let dispatcher = factory.create_dispatcher(options);
        assert!(dispatch_capability_routed_event(&dispatcher).await.is_some());
        assert_matches!(
            factory.next_event().await,
            Some(ComponentEvent { result: Ok(EventPayload::CapabilityRouted { .. }), .. })
        );

        // Verify that we cannot dispatch the CapabilityRouted event in production.
        let options = SubscriptionOptions::new(
            SubscriptionType::Component(vec!["root:0"].into()),
            SyncMode::Sync,
            ExecutionMode::Production,
        );
        let dispatcher = factory.create_dispatcher(options);
        // This indicates that the event was not dispatched.
        assert!(dispatch_capability_routed_event(&dispatcher).await.is_none());
    }
}