// Copyright 2019 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.

//! Manages peer<->peer connections and routing packets over links between nodes.

// General structure:
// A router is a collection of: - streams (each with a StreamId<LinkData>)
//                                These are streams of information flow between processes.
//                              - peers (each with a PeerId)
//                                These are other overnet instances in the overlay network.
//                                There is a client oriented and a server oriented peer per
//                                node.
//                              - links (each with a LinkId)
//                                These are connections between this instance and other
//                                instances in the mesh.
// For each node in the mesh, a routing table tracks which link on which to send data to
// that node (said link may be a third node that will be requested to forward datagrams
// on our behalf).

mod diagnostics_service;
mod link_status_updater;
mod routes;
pub(crate) mod security_context;
mod service_map;

use self::{
    diagnostics_service::run_diagostic_service_request_handler,
    link_status_updater::{run_link_status_updater, LinkStatePublisher},
    routes::Routes,
    security_context::{quiche_config_from_security_context, SecurityContext},
    service_map::{ListablePeer, ServiceMap},
};
use crate::{
    future_help::{log_errors, Observable, Observer},
    handle_info::{handle_info, HandleKey, HandleType},
    labels::{ConnectionId, Endpoint, NodeId, NodeLinkId, TransferKey},
    link::{new_link, LinkIntroductionFacts, LinkReceiver, LinkRouting, LinkSender},
    peer::{FramedStreamReader, FramedStreamWriter, MessageStats, Peer, PeerConnRef},
    proxy::{IntoProxied, ProxyTransferInitiationReceiver, RemoveFromProxyTable, StreamRefSender},
};
use anyhow::{bail, format_err, Context as _, Error};
use async_utils::mutex_ticket::MutexTicket;
use fidl::{
    endpoints::ClientEnd, AsHandleRef, Channel, EventPair, Handle, HandleBased, Socket, SocketOpts,
};
use fidl_fuchsia_overnet::{ConnectionInfo, ServiceProviderMarker, ServiceProviderProxyInterface};
use fidl_fuchsia_overnet_protocol::{
    ChannelHandle, EventPairHandle, EventPairRights, Implementation, LinkDiagnosticInfo,
    PeerConnectionDiagnosticInfo, RouteMetrics, SocketHandle, SocketType, StreamId, StreamRef,
    ZirconHandle,
};
use fuchsia_async::Task;
use futures::{future::poll_fn, lock::Mutex, prelude::*, ready};
use rand::Rng;
use std::{
    collections::{btree_map, BTreeMap, HashMap},
    convert::TryInto,
    sync::atomic::{AtomicU32, AtomicU64, Ordering},
    sync::{Arc, Weak},
    task::{Context, Poll, Waker},
    time::Duration,
};

// Re-export for link, peer.
pub(crate) use self::routes::ForwardingTable;

/// Configuration object for creating a router.
pub struct RouterOptions {
    node_id: Option<NodeId>,
    diagnostics: Option<Implementation>,
}

impl RouterOptions {
    /// Create with defaults.
    pub fn new() -> Self {
        RouterOptions { diagnostics: None, node_id: None }
    }

    /// Request a specific node id (if unset, one will be generated).
    pub fn set_node_id(mut self, node_id: NodeId) -> Self {
        self.node_id = Some(node_id);
        self
    }

    /// Enable diagnostics with a selected `implementation`.
    pub fn export_diagnostics(mut self, implementation: Implementation) -> Self {
        self.diagnostics = Some(implementation);
        self
    }
}

#[derive(Debug)]
enum PendingTransfer {
    Complete(FoundTransfer),
    Waiting(Waker),
}

type PendingTransferMap = BTreeMap<TransferKey, PendingTransfer>;

#[derive(Debug)]
pub(crate) enum FoundTransfer {
    Fused(Handle),
    Remote(FramedStreamWriter, FramedStreamReader),
}

#[derive(Debug)]
pub(crate) enum OpenedTransfer {
    Fused,
    Remote(FramedStreamWriter, FramedStreamReader, Handle),
}

#[derive(Debug)]
struct PeerMaps {
    clients: BTreeMap<NodeId, Arc<Peer>>,
    servers: BTreeMap<NodeId, Vec<Arc<Peer>>>,
    connections: HashMap<ConnectionId, Arc<Peer>>,
}

impl PeerMaps {
    async fn get_client(
        &mut self,
        local_node_id: NodeId,
        peer_node_id: NodeId,
        router: &Arc<Router>,
    ) -> Result<Arc<Peer>, Error> {
        if peer_node_id == router.node_id {
            bail!("Trying to create loopback client peer");
        }
        if let Some(p) = self.clients.get(&peer_node_id) {
            if p.node_id() != peer_node_id {
                bail!(
                    "Existing client peer gets a packet from a new peer node id {:?} (vs {:?})",
                    peer_node_id,
                    p.node_id()
                );
            }
            Ok(p.clone())
        } else {
            self.new_client(local_node_id, peer_node_id, router).await
        }
    }

    async fn lookup(
        &mut self,
        conn_id: &[u8],
        ty: quiche::Type,
        local_node_id: NodeId,
        peer_node_id: NodeId,
        router: &Arc<Router>,
    ) -> Result<Arc<Peer>, Error> {
        if peer_node_id == router.node_id {
            bail!("Trying to create loopback peer");
        }
        match (ty, conn_id.try_into().map(|conn_id| self.connections.get(&conn_id))) {
            (_, Ok(Some(p))) => {
                if p.node_id() != peer_node_id {
                    bail!(
                        "Existing looked-up peer gets a packet from a new peer node id {:?} (vs {:?})",
                        peer_node_id,
                        p.node_id()
                    );
                }
                Ok(p.clone())
            }
            (quiche::Type::Initial, _) => {
                self.new_server(local_node_id, peer_node_id, router).await
            }
            (_, _) => bail!("Packet received for unknown connection"),
        }
    }

    async fn new_client(
        &mut self,
        local_node_id: NodeId,
        peer_node_id: NodeId,
        router: &Arc<Router>,
    ) -> Result<Arc<Peer>, Error> {
        let mut config =
            router.quiche_config().await.context("creating client configuration for quiche")?;
        let conn_id = ConnectionId::new();
        let peer = Peer::new_client(
            peer_node_id,
            local_node_id,
            conn_id,
            &mut config,
            router.service_map.new_local_service_observer(),
            router,
        )?;
        self.clients.insert(peer_node_id, peer.clone());
        self.connections.insert(conn_id, peer.clone());
        Ok(peer)
    }

    async fn new_server(
        &mut self,
        local_node_id: NodeId,
        peer_node_id: NodeId,
        router: &Arc<Router>,
    ) -> Result<Arc<Peer>, Error> {
        let mut config =
            router.quiche_config().await.context("creating client configuration for quiche")?;
        let conn_id = ConnectionId::new();
        let peer = Peer::new_server(peer_node_id, local_node_id, conn_id, &mut config, router)?;
        self.servers.entry(peer_node_id).or_insert(Vec::new()).push(peer.clone());
        self.connections.insert(conn_id, peer.clone());
        Ok(peer)
    }
}

/// Wrapper to get the right list_peers behavior.
#[derive(Debug)]
pub struct ListPeersContext(Mutex<Option<Observer<Vec<ListablePeer>>>>);

static LIST_PEERS_CALL: AtomicU64 = AtomicU64::new(0);

impl ListPeersContext {
    /// Implementation of ListPeers fidl method.
    pub async fn list_peers(&self) -> Result<Vec<fidl_fuchsia_overnet::Peer>, Error> {
        let call_id = LIST_PEERS_CALL.fetch_add(1, Ordering::SeqCst);
        log::trace!("LIST_PEERS_CALL[{}] get observer", call_id);
        let mut obs = self
            .0
            .lock()
            .await
            .take()
            .ok_or_else(|| anyhow::format_err!("Already listing peers"))?;
        log::trace!("LIST_PEERS_CALL[{}] wait for value", call_id);
        let r = obs.next().await;
        log::trace!("LIST_PEERS_CALL[{}] replace observer", call_id);
        *self.0.lock().await = Some(obs);
        log::trace!("LIST_PEERS_CALL[{}] return", call_id);
        Ok(r.unwrap_or_else(Vec::new).into_iter().map(|p| p.into()).collect())
    }
}

/// Factory for local link id's (the next id to be assigned).
static NEXT_NODE_LINK_ID: AtomicU64 = AtomicU64::new(1);

/// Minted for each call to new_link, a ConnectingLinkToken helps us keep track of how many links
/// are still being established, and also acts as a capability for calling publish_link.
pub(crate) struct ConnectingLinkToken {
    router: Weak<Router>,
}

impl Drop for ConnectingLinkToken {
    fn drop(&mut self) {
        if let Some(router) = Weak::upgrade(&self.router) {
            router.connecting_links.fetch_sub(1, Ordering::Relaxed);
        }
    }
}

/// Router maintains global state for one node_id.
/// `LinkData` is a token identifying a link for layers above Router.
/// `Time` is a representation of time for the Router, to assist injecting different platforms
/// schemes.
pub struct Router {
    /// Our node id
    node_id: NodeId,
    security_context: Box<dyn SecurityContext>,
    /// All peers.
    peers: Mutex<PeerMaps>,
    links: Mutex<HashMap<NodeLinkId, Weak<LinkRouting>>>,
    link_state_publisher: LinkStatePublisher,
    service_map: ServiceMap,
    proxied_streams: Mutex<HashMap<HandleKey, ProxiedHandle>>,
    pending_transfers: Mutex<PendingTransferMap>,
    routes: Arc<Routes>,
    current_forwarding_table: Mutex<ForwardingTable>,
    task: Mutex<Option<Task<()>>>,
    connecting_links: AtomicU64,
    implementation: AtomicU32,
}

struct ProxiedHandle {
    remove_sender: futures::channel::oneshot::Sender<RemoveFromProxyTable>,
    original_paired: HandleKey,
    proxy_task: Task<()>,
}

/// Generate a new random node id
pub fn generate_node_id() -> NodeId {
    rand::thread_rng().gen::<u64>().into()
}

fn sorted<T: std::cmp::Ord>(mut v: Vec<T>) -> Vec<T> {
    v.sort();
    v
}

impl std::fmt::Debug for Router {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "Router({:?})", self.node_id)
    }
}

impl Router {
    /// New with some set of options
    pub fn new(
        options: RouterOptions,
        security_context: Box<dyn SecurityContext>,
    ) -> Result<Arc<Self>, Error> {
        // Verify that we can read all files in the security_context - this makes debugging easier
        // later on (since errors from quiche are quite non-descript).
        let verify_file = |file: &str| {
            std::fs::File::open(file).context(format_err!("Opening {}", file)).map(drop)
        };
        verify_file(security_context.node_cert())?;
        verify_file(security_context.node_private_key())?;
        verify_file(security_context.root_cert())?;

        let node_id = options.node_id.unwrap_or_else(generate_node_id);
        let service_map = ServiceMap::new(node_id);
        let (link_state_publisher, link_state_receiver) = futures::channel::mpsc::channel(1);
        let routes = Arc::new(Routes::new());
        let router = Arc::new(Router {
            node_id,
            security_context,
            link_state_publisher,
            service_map,
            links: Mutex::new(HashMap::new()),
            peers: Mutex::new(PeerMaps {
                clients: BTreeMap::new(),
                servers: BTreeMap::new(),
                connections: HashMap::new(),
            }),
            proxied_streams: Mutex::new(HashMap::new()),
            pending_transfers: Mutex::new(PendingTransferMap::new()),
            routes: routes.clone(),
            task: Mutex::new(None),
            current_forwarding_table: Mutex::new(ForwardingTable::empty()),
            connecting_links: AtomicU64::new(0),
            implementation: AtomicU32::new(
                options.diagnostics.unwrap_or(Implementation::Unknown).into_primitive(),
            ),
        });

        let link_state_observable = Observable::new(BTreeMap::new());
        let weak_router = Arc::downgrade(&router);
        *router.task.lock().now_or_never().unwrap() = Some(Task::spawn(log_errors(
            async move {
                let router = &weak_router;
                futures::future::try_join4(
                    summon_clients(router.clone(), routes.new_forwarding_table_observer()),
                    routes.run_planner(node_id, link_state_observable.new_observer()),
                    run_link_status_updater(link_state_observable, link_state_receiver),
                    async move {
                        run_diagostic_service_request_handler(router).await?;
                        Ok(())
                    },
                )
                .await
                .map(drop)
            },
            format!("router {:?} support loop failed", node_id),
        )));

        Ok(router)
    }

    /// Create a new link to some node, returning a `LinkId` describing it.
    pub fn new_link(
        self: &Arc<Self>,
        facts: LinkIntroductionFacts,
        config: crate::link::ConfigProducer,
    ) -> (LinkSender, LinkReceiver) {
        self.connecting_links.fetch_add(1, Ordering::Relaxed);
        new_link(
            NEXT_NODE_LINK_ID.fetch_add(1, Ordering::Relaxed).into(),
            self,
            config,
            facts,
            ConnectingLinkToken { router: Arc::downgrade(self) },
        )
    }

    pub(crate) fn connecting_link_count(&self) -> u64 {
        self.connecting_links.load(Ordering::Relaxed)
    }

    pub(crate) async fn publish_link(
        self: Arc<Self>,
        routing: Arc<LinkRouting>,
        rtt_observer: Observer<Option<Duration>>,
        connecting_link_token: ConnectingLinkToken,
    ) -> Result<(), Error> {
        log::info!("publish link {:?}", routing.debug_id());
        self.links.lock().await.insert(routing.id(), Arc::downgrade(&routing));
        self.link_state_publisher
            .clone()
            .send((routing.id(), routing.peer_node_id(), rtt_observer))
            .await?;
        drop(connecting_link_token);
        Ok(())
    }

    pub(crate) async fn get_link(&self, node_link_id: NodeLinkId) -> Option<Arc<LinkRouting>> {
        self.links.lock().await.get(&node_link_id).and_then(Weak::upgrade)
    }

    /// Accessor for the node id of this router.
    pub fn node_id(&self) -> NodeId {
        self.node_id
    }

    /// Accessor for the Implementation of this router.
    pub fn implementation(&self) -> Implementation {
        Implementation::from_primitive(
            self.implementation.load(std::sync::atomic::Ordering::SeqCst),
        )
        .unwrap_or(Implementation::Unknown)
    }

    /// Setter for the Implementation of this router.
    pub fn set_implementation(&self, imp: Implementation) {
        self.implementation.store(imp.into_primitive(), std::sync::atomic::Ordering::SeqCst)
    }

    pub(crate) fn service_map(&self) -> &ServiceMap {
        &self.service_map
    }

    pub(crate) fn new_forwarding_table_observer(&self) -> Observer<ForwardingTable> {
        self.routes.new_forwarding_table_observer()
    }

    pub(crate) async fn update_routes(
        &self,
        via: NodeId,
        routes: impl Iterator<Item = (NodeId, RouteMetrics)>,
    ) {
        self.routes.update(via, routes).await
    }

    /// Create a new stream to advertised service `service` on remote node id `node`.
    pub async fn connect_to_service(
        self: &Arc<Self>,
        node_id: NodeId,
        service_name: &str,
        chan: Channel,
    ) -> Result<(), Error> {
        let is_local = node_id == self.node_id;
        log::trace!(
            "Request connect_to_service '{}' on {:?}{}",
            service_name,
            node_id,
            if is_local { " [local]" } else { " [remote]" }
        );
        if is_local {
            self.service_map()
                .connect(
                    service_name,
                    chan,
                    ConnectionInfo { peer: Some(node_id.into()), ..ConnectionInfo::EMPTY },
                )
                .await
        } else {
            self.client_peer(node_id)
                .await
                .with_context(|| {
                    format_err!(
                        "Fetching client peer for new stream to {:?} for service {:?}",
                        node_id,
                        service_name,
                    )
                })?
                .new_stream(service_name, chan, self)
                .await
        }
    }

    /// Implementation of RegisterService fidl method.
    pub async fn register_service(
        &self,
        service_name: String,
        provider: ClientEnd<ServiceProviderMarker>,
    ) -> Result<(), Error> {
        self.register_raw_service(service_name, Box::new(provider.into_proxy()?)).await
    }

    /// Register a service without needing a FIDL channel.
    pub async fn register_raw_service(
        &self,
        service_name: String,
        provider: Box<dyn ServiceProviderProxyInterface>,
    ) -> Result<(), Error> {
        self.service_map().register_service(service_name, provider).await;
        Ok(())
    }

    /// Create a new list_peers context
    pub fn new_list_peers_context(&self) -> ListPeersContext {
        ListPeersContext(Mutex::new(Some(self.service_map.new_list_peers_observer())))
    }

    /// Diagnostic information for links
    pub(crate) async fn link_diagnostics(&self) -> Vec<LinkDiagnosticInfo> {
        futures::stream::iter(self.links.lock().await.iter())
            .filter_map(|(_, link)| async move {
                if let Some(link) = Weak::upgrade(link) {
                    if !link.is_closed().await {
                        return Some(link.diagnostic_info().await);
                    }
                }
                None
            })
            .collect()
            .await
    }

    /// Diagnostic information for peer connections
    pub(crate) async fn peer_diagnostics(&self) -> Vec<PeerConnectionDiagnosticInfo> {
        futures::stream::iter(self.peers.lock().await.connections.iter())
            .then(|(_, peer)| peer.diagnostics(self.node_id))
            .collect()
            .await
    }

    async fn quiche_config(&self) -> Result<quiche::Config, Error> {
        let mut config = self.new_quiche_config().await.with_context(|| {
            format_err!("applying security context: {:?}", self.security_context)
        })?;
        // TODO(ctiller): don't hardcode these
        config.set_application_protos(b"\x0bovernet/0.2")?;
        config.set_initial_max_data(10_000_000);
        config.set_initial_max_stream_data_bidi_local(1_000_000);
        config.set_initial_max_stream_data_bidi_remote(1_000_000);
        config.set_initial_max_stream_data_uni(1_000_000);
        config.set_initial_max_streams_bidi(1u64 << 60);
        config.set_initial_max_streams_uni(1u64 << 60);
        config.verify_peer(false);
        Ok(config)
    }

    pub(crate) async fn remove_peer(
        self: &Arc<Self>,
        conn_id: ConnectionId,
        due_to_routing_error: bool,
    ) {
        log::trace!("[{:?}] Request remove peer {:?}", self.node_id, conn_id);
        let mut removed_client_peer_node_id = None;
        {
            let mut peers = self.peers.lock().await;
            if let Some(peer) = peers.connections.remove(&conn_id) {
                match peer.endpoint() {
                    Endpoint::Client => {
                        if let btree_map::Entry::Occupied(o) = peers.clients.entry(peer.node_id()) {
                            if Arc::ptr_eq(o.get(), &peer) {
                                removed_client_peer_node_id = Some(peer.node_id());
                                o.remove_entry();
                            }
                        }
                    }
                    Endpoint::Server => {
                        peers
                            .servers
                            .get_mut(&peer.node_id())
                            .map(|v| v.retain(|other| !Arc::ptr_eq(&peer, other)));
                    }
                }
                peer.shutdown().await;
            }
        }
        if let Some(removed_client_peer_node_id) = removed_client_peer_node_id {
            log::trace!(
                "[{:?}] Removed client peer {:?} to {:?}",
                self.node_id,
                conn_id,
                removed_client_peer_node_id
            );
            let revive = !due_to_routing_error
                && self
                    .current_forwarding_table
                    .lock()
                    .await
                    .route_for(removed_client_peer_node_id)
                    .is_some();
            if revive {
                log::trace!(
                    "[{:?}] Revive client peer to {:?}",
                    self.node_id,
                    removed_client_peer_node_id
                );
                if let Err(e) = self
                    .peers
                    .lock()
                    .await
                    .get_client(self.node_id, removed_client_peer_node_id, self)
                    .await
                {
                    log::trace!(
                        "[{:?}] Failed revibing client peer to {:?}: {:?}",
                        self.node_id,
                        removed_client_peer_node_id,
                        e
                    );
                } else {
                    log::trace!(
                        "[{:?}] Revived client peer to {:?}",
                        self.node_id,
                        removed_client_peer_node_id
                    );
                }
            } else {
                log::trace!(
                    "[{:?}] Do not revive client peer to {:?}",
                    self.node_id,
                    removed_client_peer_node_id
                );
            }
        }
    }

    async fn client_peer(self: &Arc<Self>, peer_node_id: NodeId) -> Result<Arc<Peer>, Error> {
        self.peers.lock().await.get_client(self.node_id, peer_node_id, self).await
    }

    pub(crate) async fn lookup_peer(
        self: &Arc<Self>,
        conn_id: &[u8],
        ty: quiche::Type,
        peer_node_id: NodeId,
    ) -> Result<Arc<Peer>, Error> {
        self.peers.lock().await.lookup(conn_id, ty, self.node_id, peer_node_id, self).await
    }

    fn add_proxied(
        self: &Arc<Self>,
        proxied_streams: &mut HashMap<HandleKey, ProxiedHandle>,
        this_handle_key: HandleKey,
        pair_handle_key: HandleKey,
        remove_sender: futures::channel::oneshot::Sender<RemoveFromProxyTable>,
        f: impl 'static + Send + Future<Output = Result<(), Error>>,
    ) {
        let router = Arc::downgrade(&self);
        let proxy_task = Task::spawn(async move {
            if let Err(e) = f.await {
                log::trace!("[{:?}:{:?}] Proxy failed: {:?}", this_handle_key, pair_handle_key, e);
            } else {
                log::trace!(
                    "[{:?}:{:?}] Proxy completed successfully",
                    this_handle_key,
                    pair_handle_key
                );
            }
            if let Some(router) = Weak::upgrade(&router) {
                router.remove_proxied(this_handle_key, pair_handle_key).await;
            }
        });
        assert!(proxied_streams
            .insert(
                this_handle_key,
                ProxiedHandle { remove_sender, original_paired: pair_handle_key, proxy_task },
            )
            .is_none());
    }

    // Remove a proxied handle from our table.
    // Called by proxy::Proxy::drop.
    async fn remove_proxied(
        self: &Arc<Self>,
        this_handle_key: HandleKey,
        pair_handle_key: HandleKey,
    ) {
        let mut proxied_streams = self.proxied_streams.lock().await;
        log::trace!(
            "{:?} REMOVE_PROXIED: {:?}:{:?} all={:?}",
            self.node_id,
            this_handle_key,
            pair_handle_key,
            sorted(proxied_streams.keys().map(|x| *x).collect::<Vec<_>>())
        );
        if let Some(removed) = proxied_streams.remove(&this_handle_key) {
            assert_eq!(removed.original_paired, pair_handle_key);
            let _ = removed.remove_sender.send(RemoveFromProxyTable::Dropped);
            removed.proxy_task.detach();
        }
    }

    // Prepare a handle to be sent to another machine.
    // Returns a ZirconHandle describing the established proxy.
    pub(crate) async fn send_proxied(
        self: &Arc<Self>,
        handle: Handle,
        conn: PeerConnRef<'_>,
        stats: Arc<MessageStats>,
    ) -> Result<ZirconHandle, Error> {
        let raw_handle = handle.raw_handle(); // for debugging
        let info = handle_info(handle.as_handle_ref())
            .with_context(|| format!("Getting handle information for {}", raw_handle))?;
        let mut proxied_streams = self.proxied_streams.lock().await;
        log::trace!(
            "{:?} SEND_PROXIED: {:?} {:?} all={:?}",
            self.node_id,
            handle,
            info,
            sorted(proxied_streams.keys().map(|x| *x).collect::<Vec<_>>())
        );
        if let Some(pair) = proxied_streams.remove(&info.pair_handle_key) {
            // This handle is the other end of an already proxied object...
            // Here we need to inform the existing proxy loop that a transfer is going to be
            // initiated, and to where.
            drop(proxied_streams);
            assert_eq!(info.this_handle_key, pair.original_paired);
            log::trace!("Send paired proxied {:?} orig_pair={:?}", handle, pair.original_paired);
            // We allocate a drain stream to flush any messages we've buffered locally to the new
            // endpoint.
            let drain_stream = conn.alloc_uni().into();
            let (stream_ref_sender, stream_ref_receiver) = StreamRefSender::new();
            pair.remove_sender
                .send(RemoveFromProxyTable::InitiateTransfer {
                    paired_handle: handle,
                    drain_stream,
                    stream_ref_sender,
                })
                .map_err(|_| format_err!("Failed to initiate transfer"))?;
            let stream_ref = stream_ref_receiver
                .await
                .with_context(|| format!("waiting for stream_ref for {:?}", raw_handle))?;
            pair.proxy_task.detach();
            match info.handle_type {
                HandleType::Channel(rights) => {
                    Ok(ZirconHandle::Channel(ChannelHandle { stream_ref, rights }))
                }
                HandleType::Socket(socket_type, rights) => {
                    Ok(ZirconHandle::Socket(SocketHandle { stream_ref, socket_type, rights }))
                }
                HandleType::EventPair => Ok(ZirconHandle::EventPair(EventPairHandle {
                    stream_ref,
                    rights: EventPairRights::empty(),
                })),
            }
        } else {
            // This handle (and its pair) is previously unseen... establish a proxy stream for it
            log::trace!("Send proxied {:?}", handle);
            let (tx, rx) = futures::channel::oneshot::channel();
            let rx = ProxyTransferInitiationReceiver::new(rx.map_err(move |_| {
                format_err!(
                    "cancelled transfer via send_proxied {:?}\n{}",
                    info,
                    111 //std::backtrace::Backtrace::force_capture()
                )
            }));
            let (stream_writer, stream_reader) = conn.alloc_bidi();
            let stream_ref = StreamRef::Creating(StreamId { id: stream_writer.id() });
            Ok(match info.handle_type {
                HandleType::Channel(rights) => {
                    self.add_proxied(
                        &mut *proxied_streams,
                        info.this_handle_key,
                        info.pair_handle_key,
                        tx,
                        crate::proxy::spawn_send(
                            Channel::from_handle(handle).into_proxied()?,
                            rx,
                            stream_writer.into(),
                            stream_reader.into(),
                            stats,
                            Arc::downgrade(&self),
                        ),
                    );
                    ZirconHandle::Channel(ChannelHandle { stream_ref, rights })
                }
                HandleType::Socket(socket_type, rights) => {
                    self.add_proxied(
                        &mut *proxied_streams,
                        info.this_handle_key,
                        info.pair_handle_key,
                        tx,
                        crate::proxy::spawn_send(
                            Socket::from_handle(handle).into_proxied()?,
                            rx,
                            stream_writer.into(),
                            stream_reader.into(),
                            stats,
                            Arc::downgrade(&self),
                        ),
                    );
                    ZirconHandle::Socket(SocketHandle { stream_ref, socket_type, rights })
                }
                HandleType::EventPair => {
                    self.add_proxied(
                        &mut *proxied_streams,
                        info.this_handle_key,
                        info.pair_handle_key,
                        tx,
                        crate::proxy::spawn_send(
                            EventPair::from_handle(handle).into_proxied()?,
                            rx,
                            stream_writer.into(),
                            stream_reader.into(),
                            stats,
                            Arc::downgrade(&self),
                        ),
                    );
                    ZirconHandle::EventPair(EventPairHandle {
                        stream_ref,
                        rights: EventPairRights::empty(),
                    })
                }
            })
        }
    }

    // Take a received handle description and construct a fidl::Handle that represents it
    // whilst establishing proxies as required
    pub(crate) async fn recv_proxied(
        self: &Arc<Self>,
        handle: ZirconHandle,
        conn: PeerConnRef<'_>,
        stats: Arc<MessageStats>,
    ) -> Result<Handle, Error> {
        match handle {
            ZirconHandle::Channel(ChannelHandle { stream_ref, rights }) => {
                self.recv_proxied_handle(
                    conn,
                    stats,
                    stream_ref,
                    move || Channel::create().map_err(Into::into),
                    rights,
                )
                .await
            }
            ZirconHandle::Socket(SocketHandle { stream_ref, socket_type, rights }) => {
                let opts = match socket_type {
                    SocketType::Stream => SocketOpts::STREAM,
                    SocketType::Datagram => SocketOpts::DATAGRAM,
                };
                self.recv_proxied_handle(
                    conn,
                    stats,
                    stream_ref,
                    move || Socket::create(opts).map_err(Into::into),
                    rights,
                )
                .await
            }
            ZirconHandle::EventPair(EventPairHandle { stream_ref, rights }) => {
                self.recv_proxied_handle(
                    conn,
                    stats,
                    stream_ref,
                    move || EventPair::create().map_err(Into::into),
                    rights,
                )
                .await
            }
        }
    }

    async fn recv_proxied_handle<Hdl, CreateType>(
        self: &Arc<Self>,
        conn: PeerConnRef<'_>,
        stats: Arc<MessageStats>,
        stream_ref: StreamRef,
        create_handles: impl FnOnce() -> Result<(CreateType, CreateType), Error> + 'static,
        rights: CreateType::Rights,
    ) -> Result<Handle, Error>
    where
        Hdl: 'static + for<'a> crate::proxy::ProxyableRW<'a>,
        CreateType: 'static
            + fidl::HandleBased
            + IntoProxied<Proxied = Hdl>
            + std::fmt::Debug
            + crate::handle_info::WithRights,
    {
        let (tx, rx) = futures::channel::oneshot::channel();
        let rx = ProxyTransferInitiationReceiver::new(
            rx.map_err(move |_| format_err!("cancelled transfer via recv_proxied")),
        );
        let (h, p) = crate::proxy::spawn_recv(
            create_handles,
            rights,
            rx,
            stream_ref,
            conn,
            stats,
            Arc::downgrade(&self),
        )
        .await?;
        if let Some(p) = p {
            let info = handle_info(h.as_handle_ref())?;
            self.add_proxied(
                &mut *self.proxied_streams.lock().await,
                info.pair_handle_key,
                info.this_handle_key,
                tx,
                p,
            );
        }
        Ok(h)
    }

    // Note the endpoint of a transfer that we know about (may complete a transfer operation)
    pub(crate) async fn post_transfer(
        &self,
        transfer_key: TransferKey,
        other_end: FoundTransfer,
    ) -> Result<(), Error> {
        let mut pending_transfers = self.pending_transfers.lock().await;
        match pending_transfers.insert(transfer_key, PendingTransfer::Complete(other_end)) {
            Some(PendingTransfer::Complete(_)) => bail!("Duplicate transfer received"),
            Some(PendingTransfer::Waiting(w)) => w.wake(),
            None => (),
        }
        Ok(())
    }

    fn poll_find_transfer(
        &self,
        ctx: &mut Context<'_>,
        transfer_key: TransferKey,
        lock: &mut MutexTicket<'_, PendingTransferMap>,
    ) -> Poll<Result<FoundTransfer, Error>> {
        let mut pending_transfers = ready!(lock.poll(ctx));
        if let Some(PendingTransfer::Complete(other_end)) = pending_transfers.remove(&transfer_key)
        {
            Poll::Ready(Ok(other_end))
        } else {
            pending_transfers.insert(transfer_key, PendingTransfer::Waiting(ctx.waker().clone()));
            Poll::Pending
        }
    }

    // Lookup a transfer that we're expected to eventually know about
    pub(crate) async fn find_transfer(
        &self,
        transfer_key: TransferKey,
    ) -> Result<FoundTransfer, Error> {
        let mut lock = MutexTicket::new(&self.pending_transfers);
        poll_fn(|ctx| self.poll_find_transfer(ctx, transfer_key, &mut lock)).await
    }

    // Begin a transfer operation (opposite of find_transfer), publishing an endpoint on the remote
    // nodes transfer table.
    pub(crate) async fn open_transfer(
        self: &Arc<Router>,
        target: NodeId,
        transfer_key: TransferKey,
        handle: Handle,
    ) -> Result<OpenedTransfer, Error> {
        if target == self.node_id {
            // The target is local: we just file away the handle.
            // Later, find_transfer will find this and we'll collapse away Overnet's involvement and
            // reunite the channel ends.
            let info = handle_info(handle.as_handle_ref())?;
            let mut proxied_streams = self.proxied_streams.lock().await;
            log::trace!(
                "{:?} OPEN_TRANSFER_REMOVE_PROXIED: key={:?} {:?} all={:?}",
                self.node_id,
                transfer_key,
                info,
                sorted(proxied_streams.keys().map(|x| *x).collect::<Vec<_>>())
            );
            if let Some(removed) = proxied_streams.remove(&info.this_handle_key) {
                assert_eq!(removed.original_paired, info.pair_handle_key);
                assert!(removed.remove_sender.send(RemoveFromProxyTable::Dropped).is_ok());
                removed.proxy_task.detach();
            }
            if let Some(removed) = proxied_streams.remove(&info.pair_handle_key) {
                assert_eq!(removed.original_paired, info.this_handle_key);
                assert!(removed.remove_sender.send(RemoveFromProxyTable::Dropped).is_ok());
                removed.proxy_task.detach();
            }
            self.post_transfer(transfer_key, FoundTransfer::Fused(handle)).await?;
            Ok(OpenedTransfer::Fused)
        } else {
            if let Some((writer, reader)) =
                self.client_peer(target).await?.send_open_transfer(transfer_key).await
            {
                Ok(OpenedTransfer::Remote(writer, reader, handle))
            } else {
                bail!("{:?} failed sending open transfer to {:?}", self.node_id, target)
            }
        }
    }

    /// Generate a quiche configuration using this routers certificates
    pub async fn new_quiche_config(&self) -> Result<quiche::Config, Error> {
        quiche_config_from_security_context(&*self.security_context).await
    }
}

async fn summon_clients(
    router: Weak<Router>,
    mut forwarding_table: Observer<ForwardingTable>,
) -> Result<(), Error> {
    let get_router = move || Weak::upgrade(&router).ok_or_else(|| format_err!("router gone"));
    while let Some(forwarding_table) = forwarding_table.next().await {
        let router = get_router()?;
        *router.current_forwarding_table.lock().await = forwarding_table.clone();
        let mut peers = router.peers.lock().await;
        for (destination, _) in forwarding_table.iter() {
            let _ = peers.get_client(router.node_id, destination, &router).await?;
        }
    }
    Ok(())
}

#[cfg(test)]
mod tests {

    use super::*;
    use crate::test_util::*;

    #[fuchsia::test]
    async fn no_op(run: usize) {
        let mut node_id_gen = NodeIdGenerator::new("router::no_op", run);
        node_id_gen.new_router().unwrap();
        let id = node_id_gen.next().unwrap();
        assert_eq!(
            Router::new(RouterOptions::new().set_node_id(id), Box::new(test_security_context()))
                .unwrap()
                .node_id,
            id
        );
    }

    async fn forward(mut sender: LinkSender, mut receiver: LinkReceiver) -> Result<(), Error> {
        while let Some(mut packet) = sender.next_send().await {
            packet.drop_inner_locks();
            receiver.received_frame(packet.bytes_mut()).await;
        }
        Ok(())
    }

    async fn register_test_service(
        serving_router: Arc<Router>,
        client_router: Arc<Router>,
        service: &'static str,
    ) -> futures::channel::oneshot::Receiver<(NodeId, Channel)> {
        use parking_lot::Mutex;
        let (send, recv) = futures::channel::oneshot::channel();
        struct TestService(
            Mutex<Option<futures::channel::oneshot::Sender<(NodeId, Channel)>>>,
            &'static str,
        );
        impl fidl_fuchsia_overnet::ServiceProviderProxyInterface for TestService {
            fn connect_to_service(
                &self,
                chan: fidl::Channel,
                connection_info: fidl_fuchsia_overnet::ConnectionInfo,
            ) -> std::result::Result<(), fidl::Error> {
                println!("{} got request", self.1);
                self.0
                    .lock()
                    .take()
                    .unwrap()
                    .send((connection_info.peer.unwrap().id.into(), chan))
                    .unwrap();
                println!("{} forwarded channel", self.1);
                Ok(())
            }
        }
        serving_router
            .service_map()
            .register_service(
                service.to_string(),
                Box::new(TestService(Mutex::new(Some(send)), service)),
            )
            .await;
        let serving_node_id = serving_router.node_id();
        println!("{} wait for service to appear @ client", service);
        let lpc = client_router.new_list_peers_context();
        loop {
            let peers = lpc.list_peers().await.unwrap();
            println!("{} got peers {:?}", service, peers);
            if peers
                .iter()
                .find(move |peer| {
                    serving_node_id == peer.id.into()
                        && peer
                            .description
                            .services
                            .as_ref()
                            .unwrap()
                            .iter()
                            .find(move |&advertised_service| advertised_service == service)
                            .is_some()
                })
                .is_some()
            {
                break;
            }
        }
        recv
    }

    async fn run_two_node<
        F: 'static + Clone + Sync + Send + Fn(Arc<Router>, Arc<Router>) -> Fut,
        Fut: 'static + Send + Future<Output = Result<(), Error>>,
    >(
        name: &'static str,
        run: usize,
        f: F,
    ) -> Result<(), Error> {
        let mut node_id_gen = NodeIdGenerator::new(name, run);
        let router1 = node_id_gen.new_router()?;
        let router2 = node_id_gen.new_router()?;
        let (link1_sender, link1_receiver) =
            router1.new_link(Default::default(), Box::new(|| None));
        let (link2_sender, link2_receiver) =
            router2.new_link(Default::default(), Box::new(|| None));
        let _fwd = Task::spawn(async move {
            if let Err(e) = futures::future::try_join(
                forward(link1_sender, link2_receiver),
                forward(link2_sender, link1_receiver),
            )
            .await
            {
                log::trace!("forwarding failed: {:?}", e)
            }
        });
        f(router1, router2).await
    }

    #[fuchsia::test]
    async fn no_op_env(run: usize) -> Result<(), Error> {
        run_two_node("router::no_op_env", run, |_router1, _router2| async { Ok(()) }).await
    }

    #[fuchsia::test]
    async fn create_stream(run: usize) -> Result<(), Error> {
        run_two_node("create_stream", run, |router1, router2| async move {
            let (_, p) = fidl::Channel::create()?;
            println!("create_stream: register service");
            let s = register_test_service(router2.clone(), router1.clone(), "create_stream").await;
            println!("create_stream: connect to service");
            router1.connect_to_service(router2.node_id, "create_stream", p).await?;
            println!("create_stream: wait for connection");
            let (node_id, _) = s.await?;
            assert_eq!(node_id, router1.node_id);
            Ok(())
        })
        .await
    }

    #[fuchsia::test]
    async fn send_datagram_immediately(run: usize) -> Result<(), Error> {
        run_two_node("send_datagram_immediately", run, |router1, router2| async move {
            let (c, p) = fidl::Channel::create()?;
            println!("send_datagram_immediately: register service");
            let s = register_test_service(
                router2.clone(),
                router1.clone(),
                "send_datagram_immediately",
            )
            .await;
            println!("send_datagram_immediately: connect to service");
            router1.connect_to_service(router2.node_id, "send_datagram_immediately", p).await?;
            println!("send_datagram_immediately: wait for connection");
            let (node_id, s) = s.await?;
            assert_eq!(node_id, router1.node_id);
            let c = fidl::AsyncChannel::from_channel(c)?;
            let s = fidl::AsyncChannel::from_channel(s)?;
            c.write(&[1, 2, 3, 4, 5], &mut Vec::new())?;
            let mut buf = fidl::MessageBufEtc::new();
            println!("send_datagram_immediately: wait for datagram");
            s.recv_etc_msg(&mut buf).await?;
            assert_eq!(buf.n_handle_infos(), 0);
            assert_eq!(buf.bytes(), &[1, 2, 3, 4, 5]);
            Ok(())
        })
        .await
    }

    #[fuchsia::test]
    async fn ping_pong(run: usize) -> Result<(), Error> {
        run_two_node("ping_pong", run, |router1, router2| async move {
            let (c, p) = fidl::Channel::create()?;
            println!("ping_pong: register service");
            let s = register_test_service(router2.clone(), router1.clone(), "ping_pong").await;
            println!("ping_pong: connect to service");
            router1.connect_to_service(router2.node_id, "ping_pong", p).await?;
            println!("ping_pong: wait for connection");
            let (node_id, s) = s.await?;
            assert_eq!(node_id, router1.node_id);
            let c = fidl::AsyncChannel::from_channel(c)?;
            let s = fidl::AsyncChannel::from_channel(s)?;
            println!("ping_pong: send ping");
            c.write(&[1, 2, 3, 4, 5], &mut Vec::new())?;
            println!("ping_pong: receive ping");
            let mut buf = fidl::MessageBufEtc::new();
            s.recv_etc_msg(&mut buf).await?;
            assert_eq!(buf.n_handle_infos(), 0);
            assert_eq!(buf.bytes(), &[1, 2, 3, 4, 5]);
            println!("ping_pong: send pong");
            s.write(&[9, 8, 7, 6, 5, 4, 3, 2, 1], &mut Vec::new())?;
            println!("ping_pong: receive pong");
            let mut buf = fidl::MessageBufEtc::new();
            c.recv_etc_msg(&mut buf).await?;
            assert_eq!(buf.n_handle_infos(), 0);
            assert_eq!(buf.bytes(), &[9, 8, 7, 6, 5, 4, 3, 2, 1]);
            Ok(())
        })
        .await
    }

    fn ensure_pending(f: &mut (impl Send + Unpin + Future<Output = ()>)) {
        let mut ctx = Context::from_waker(futures::task::noop_waker_ref());
        // Poll a bunch of times to convince ourselves the future is pending forever...
        for _ in 0..1000 {
            assert!(f.poll_unpin(&mut ctx).is_pending());
        }
    }

    #[fuchsia::test]
    async fn concurrent_list_peer_calls_will_error(run: usize) -> Result<(), Error> {
        let mut node_id_gen = NodeIdGenerator::new("concurrent_list_peer_calls_will_error", run);
        let n = node_id_gen.new_router().unwrap();
        let lp = n.new_list_peers_context();
        lp.list_peers().await.unwrap();
        let mut never_completes = async {
            lp.list_peers().await.unwrap();
        }
        .boxed();
        ensure_pending(&mut never_completes);
        lp.list_peers().await.expect_err("Concurrent list peers should fail");
        ensure_pending(&mut never_completes);
        Ok(())
    }
}