tehuti 0.22.0

use crate::{
    buffer::Buffer,
    channel::{ChannelId, ChannelMode},
    event::{Duplex, Receiver, Sender, unbounded},
    meeting::{Meeting, MeetingEngineEvent, MeetingInterface, MeetingInterfaceResult},
    peer::{PeerFactory, PeerId, PeerInfo},
    protocol::{ProtocolControlFrame, ProtocolFrame, ProtocolPacketData, ProtocolPacketFrame},
    replication::Replicable,
    third_party::time::{Duration, Instant},
};
use serde::{Deserialize, Serialize};
use std::{
    collections::{BTreeMap, btree_map::Entry},
    error::Error,
    pin::Pin,
    sync::Arc,
    task::{Context, Poll},
    thread::{Builder, JoinHandle, sleep},
};

#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
pub struct EngineId(u128);

impl EngineId {
    pub fn uuid() -> Self {
        Self::new(typid::ID::<()>::default().uuid().as_u128())
    }

    pub const fn new(id: u128) -> Self {
        Self(id)
    }

    pub const fn id(&self) -> u128 {
        self.0
    }
}

impl Replicable for EngineId {
    fn collect_changes(&self, buffer: &mut Buffer) -> Result<(), Box<dyn Error>> {
        self.0.collect_changes(buffer)?;
        Ok(())
    }

    fn apply_changes(&mut self, buffer: &mut Buffer) -> Result<(), Box<dyn Error>> {
        self.0.apply_changes(buffer)?;
        Ok(())
    }
}

impl std::fmt::Display for EngineId {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "#engine:{}", self.0)
    }
}

#[derive(Debug)]
pub struct EnginePacketSender {
    pub mode: ChannelMode,
    pub sender: Sender<ProtocolPacketData>,
}

#[derive(Debug)]
pub struct EnginePacketReceiver {
    pub mode: ChannelMode,
    pub receiver: Receiver<ProtocolPacketData>,
}

#[derive(Debug)]
pub struct EnginePeerDescriptor {
    pub info: PeerInfo,
    pub packet_senders: BTreeMap<ChannelId, EnginePacketSender>,
    pub packet_receivers: BTreeMap<ChannelId, EnginePacketReceiver>,
    pub origin_engine: Option<EngineId>,
}

pub enum EngineMeetingEvent {
    RegisterEngine {
        engine_id: EngineId,
        frames: Duplex<ProtocolFrame>,
    },
    UnregisterEngine {
        engine_id: EngineId,
    },
}

#[derive(Debug, Default, Clone, PartialEq, Eq)]
pub struct EngineMeetingConfig {
    pub warn_duplicate_peer_creation: bool,
    pub warn_unknown_peer_destruction: bool,
    pub warn_unknown_peer_packet: bool,
    pub warn_unknown_channel_packet: bool,
    pub warn_unhandled_engine_event: bool,
}

impl EngineMeetingConfig {
    pub fn disable_all_warnings(mut self) -> Self {
        self.warn_duplicate_peer_creation = false;
        self.warn_unknown_peer_destruction = false;
        self.warn_unknown_peer_packet = false;
        self.warn_unknown_channel_packet = false;
        self.warn_unhandled_engine_event = false;
        self
    }

    pub fn enable_all_warnings(mut self) -> Self {
        self.warn_duplicate_peer_creation = true;
        self.warn_unknown_peer_destruction = true;
        self.warn_unknown_peer_packet = true;
        self.warn_unknown_channel_packet = true;
        self.warn_unhandled_engine_event = true;
        self
    }

    pub fn warn_duplicate_peer_creation(mut self, value: bool) -> Self {
        self.warn_duplicate_peer_creation = value;
        self
    }

    pub fn warn_unknown_peer_destruction(mut self, value: bool) -> Self {
        self.warn_unknown_peer_destruction = value;
        self
    }

    pub fn warn_unknown_peer_packet(mut self, value: bool) -> Self {
        self.warn_unknown_peer_packet = value;
        self
    }

    pub fn warn_unknown_channel_packet(mut self, value: bool) -> Self {
        self.warn_unknown_channel_packet = value;
        self
    }

    pub fn warn_unhandled_engine_event(mut self, value: bool) -> Self {
        self.warn_unhandled_engine_event = value;
        self
    }
}

pub struct EngineMeetingResult {
    pub meeting: EngineMeeting,
    pub interface: MeetingInterface,
    pub events_sender: Sender<EngineMeetingEvent>,
}

pub struct EngineMeeting {
    pub config: EngineMeetingConfig,
    meeting: Meeting,
    engine_event: Duplex<MeetingEngineEvent>,
    peers: BTreeMap<PeerId, EnginePeerDescriptor>,
    engines: BTreeMap<EngineId, (Duplex<ProtocolFrame>, Instant)>,
    events_receiver: Receiver<EngineMeetingEvent>,
    heartbeat_timestamp: u64,
    heartbeat_timer: Instant,
    pub heartbeat_interval: Duration,
    pub engine_disconnect_timeout: Duration,
}

impl EngineMeeting {
    pub fn make(
        name: impl ToString,
        config: EngineMeetingConfig,
        factory: Arc<PeerFactory>,
    ) -> EngineMeetingResult {
        let MeetingInterfaceResult {
            meeting,
            interface,
            engine_event,
        } = MeetingInterface::make(factory, name);
        let (events_sender, events_receiver) = unbounded();
        EngineMeetingResult {
            meeting: Self {
                config,
                meeting,
                engine_event,
                peers: Default::default(),
                engines: Default::default(),
                events_receiver,
                heartbeat_timestamp: 0,
                heartbeat_timer: Instant::now(),
                heartbeat_interval: Duration::from_secs(1),
                engine_disconnect_timeout: Duration::from_secs(5),
            },
            interface,
            events_sender,
        }
    }

    pub fn heartbeat_interval(mut self, interval: Duration) -> Self {
        self.heartbeat_interval = interval;
        self
    }

    pub fn engine_disconnect_timeout(mut self, timeout: Duration) -> Self {
        self.engine_disconnect_timeout = timeout;
        self
    }

    pub fn meeting_factory(&self) -> &Arc<PeerFactory> {
        self.meeting.factory()
    }

    pub fn meeting_peers(&self) -> impl Iterator<Item = PeerId> {
        self.meeting.peers()
    }

    pub fn peers(&self) -> impl Iterator<Item = PeerId> {
        self.peers.keys().copied()
    }

    pub fn engines(&self) -> impl Iterator<Item = EngineId> {
        self.engines.keys().copied()
    }

    pub fn maintain(&mut self) -> Result<(), Box<dyn Error>> {
        // Send hearbeats to engines and check for engine disconnect.
        if self.heartbeat_timer.elapsed() >= self.heartbeat_interval {
            self.heartbeat_timestamp = self.heartbeat_timestamp.wrapping_add(1);
            let frame = ProtocolFrame::Control(ProtocolControlFrame::Heartbeat {
                timestamp: self.heartbeat_timestamp,
            });
            for (engine_id, (frames, _)) in &self.engines {
                tracing::event!(
                    target: "tehuti::engine::meeting",
                    tracing::Level::TRACE,
                    "Meeting engine {:?} sending heartbeat {:?}",
                    engine_id,
                    self.heartbeat_timestamp,
                );
                if !frames.sender.try_send(frame.clone()) {
                    tracing::event!(
                        target: "tehuti::engine::meeting",
                        tracing::Level::ERROR,
                        "Meeting engine {:?} frame sender disconnected (heartbeat)",
                        engine_id,
                    );
                }
            }
            self.heartbeat_timer = Instant::now();
        }
        self.engines.retain(|engine_id, (frames, last_seen)| {
            if frames.sender.is_disconnected()
                || frames.receiver.is_disconnected()
                || last_seen.elapsed() > self.engine_disconnect_timeout
            {
                tracing::event!(
                    target: "tehuti::engine::meeting",
                    tracing::Level::WARN,
                    "Meeting engine {:?} is disconnected. Unregistering",
                    engine_id,
                );
                self.peers.retain(|peer_id, descriptor| {
                    if descriptor.origin_engine == Some(*engine_id) {
                        tracing::event!(
                            target: "tehuti::engine::meeting",
                            tracing::Level::TRACE,
                            "Meeting destroying peer {:?} due to engine {:?} disconnection",
                            peer_id,
                            engine_id,
                        );
                        if !self
                            .engine_event
                            .sender
                            .try_send(MeetingEngineEvent::PeerLeft(*peer_id))
                        {
                            tracing::event!(
                                target: "tehuti::engine::meeting",
                                tracing::Level::ERROR,
                                "Meeting engine {:?} failed to send peer left event for peer {:?}",
                                engine_id,
                                peer_id,
                            );
                        }
                        false
                    } else {
                        true
                    }
                });
                false
            } else {
                true
            }
        });

        // Handle engine meeting control events.
        'meeting_control_events: for event in self.events_receiver.iter() {
            match event {
                EngineMeetingEvent::RegisterEngine { engine_id, frames } => {
                    tracing::event!(
                        target: "tehuti::engine::meeting",
                        tracing::Level::TRACE,
                        "Meeting registering engine: {:?}",
                        engine_id,
                    );
                    for descriptor in self.peers.values() {
                        tracing::event!(
                            target: "tehuti::engine::meeting",
                            tracing::Level::TRACE,
                            "Meeting engine: {:?} register sending create peer {:?} with role {:?}",
                            engine_id,
                            descriptor.info.peer_id,
                            descriptor.info.role_id,
                        );
                        let frame = ProtocolFrame::Control(ProtocolControlFrame::CreatePeer(
                            descriptor.info.peer_id,
                            descriptor.info.role_id,
                        ));
                        if !frames.sender.try_send(frame) {
                            tracing::event!(
                                target: "tehuti::engine::meeting",
                                tracing::Level::ERROR,
                                "Meeting engine {:?} frame sender disconnected (register create peer)",
                                engine_id,
                            );
                            continue 'meeting_control_events;
                        }
                    }
                    self.engines.insert(engine_id, (frames, Instant::now()));
                }
                EngineMeetingEvent::UnregisterEngine { engine_id } => {
                    tracing::event!(
                        target: "tehuti::engine::meeting",
                        tracing::Level::TRACE,
                        "Meeting unregistering engine: {:?}",
                        engine_id,
                    );
                    self.engines.remove(&engine_id);
                    self.peers.retain(|peer_id, descriptor| {
                        if descriptor.origin_engine == Some(engine_id) {
                            tracing::event!(
                                target: "tehuti::engine::meeting",
                                tracing::Level::TRACE,
                                "Meeting destroying peer {:?} due to engine {:?} unregistration",
                                peer_id,
                                engine_id,
                            );
                            if !self
                                .engine_event
                                .sender
                                .try_send(MeetingEngineEvent::PeerLeft(*peer_id))
                            {
                                tracing::event!(
                                    target: "tehuti::engine::meeting",
                                    tracing::Level::ERROR,
                                    "Meeting engine {:?} failed to send peer left event for peer {:?}",
                                    engine_id,
                                    peer_id,
                                );
                            }
                            false
                        } else {
                            true
                        }
                    });
                }
            }
        }

        // Process receiving frames from all engines.
        let mut refresh_engines = Vec::new();
        for (engine_id, (frames, _)) in &self.engines {
            for frame in frames.receiver.iter() {
                match frame {
                    ProtocolFrame::Control(frame) => {
                        if !matches!(frame, ProtocolControlFrame::Heartbeat { .. }) {
                            for (engine_id2, (frames2, _)) in &self.engines {
                                if engine_id != engine_id2 {
                                    tracing::event!(
                                        target: "tehuti::engine::meeting",
                                        tracing::Level::TRACE,
                                        "Meeting engine {:?} forwarding control frame to {:?}: {:?}",
                                        engine_id,
                                        engine_id2,
                                        frame,
                                    );
                                    frames2
                                        .sender
                                        .send(ProtocolFrame::Control(frame.clone()))
                                        .map_err(|err| {
                                            format!(
                                                "Meeting engine {:?} frame sender error (forwarding control frame): {:?}",
                                                engine_id, err
                                            )
                                        })?;
                                }
                            }
                        }

                        match frame {
                            ProtocolControlFrame::Heartbeat { timestamp } => {
                                tracing::event!(
                                    target: "tehuti::engine::meeting",
                                    tracing::Level::TRACE,
                                    "Meeting engine {:?} got heartbeat {:?}",
                                    engine_id,
                                    timestamp,
                                );
                                refresh_engines.push(*engine_id);
                            }
                            ProtocolControlFrame::CreatePeer(peer_id, peer_role_id) => {
                                tracing::event!(
                                    target: "tehuti::engine::meeting",
                                    tracing::Level::TRACE,
                                    "Meeting engine {:?} got create peer {:?} with role {:?}",
                                    engine_id,
                                    peer_id,
                                    peer_role_id,
                                );
                                if let Err(error) = self
                                    .engine_event
                                    .sender
                                    .send(MeetingEngineEvent::PeerJoined(
                                        *engine_id,
                                        peer_id,
                                        peer_role_id,
                                    ))
                                    .map_err(|err| {
                                        format!(
                                            "Meeting engine {:?} outside engine sender error (peer joined): {:?}",
                                            engine_id, err
                                        )
                                    })
                                {
                                    tracing::event!(
                                        target: "tehuti::engine::meeting",
                                        tracing::Level::ERROR,
                                        "Meeting engine {:?} failed to send peer joined event for peer {:?}: {:?}",
                                        engine_id,
                                        peer_id,
                                        error,
                                    );
                                }
                            }
                            ProtocolControlFrame::DestroyPeer(peer_id) => {
                                tracing::event!(
                                    target: "tehuti::engine::meeting",
                                    tracing::Level::TRACE,
                                    "Meeting engine {:?} got destroy peer {:?}",
                                    engine_id,
                                    peer_id,
                                );
                                if let Err(error) = self
                                    .engine_event
                                    .sender
                                    .send(MeetingEngineEvent::PeerLeft(peer_id))
                                    .map_err(|err| {
                                        format!(
                                            "Meeting engine {:?} outside engine sender error (peer left): {:?}",
                                            engine_id, err
                                        )
                                    })
                                {
                                    tracing::event!(
                                        target: "tehuti::engine::meeting",
                                        tracing::Level::ERROR,
                                        "Meeting engine {:?} failed to send peer left event for peer {:?}: {:?}",
                                        engine_id,
                                        peer_id,
                                        error,
                                    );
                                }
                            }
                        }
                    }
                    ProtocolFrame::Packet(frame) => {
                        for (engine_id2, (frames2, _)) in &self.engines {
                            if !frame.data.recepients.is_empty()
                                && !frame.data.recepients.contains(engine_id2)
                            {
                                continue;
                            }
                            if engine_id != engine_id2 {
                                tracing::event!(
                                    target: "tehuti::engine::meeting",
                                    tracing::Level::TRACE,
                                    "Meeting engine {:?} forwarding packet frame for peer {:?} channel {:?} to {:?}. {} bytes",
                                    engine_id,
                                    frame.peer_id,
                                    frame.channel_id,
                                    engine_id2,
                                    frame.data.data.len(),
                                );
                                frames2
                                    .sender
                                    .send(ProtocolFrame::Packet(frame.clone()))
                                    .map_err(|err| {
                                        format!(
                                            "Meeting engine {:?} frame sender error (forwarding packet frame): {:?}",
                                            engine_id2, err
                                        )
                                    })?;
                            }
                        }

                        if let Some(peer) = self.peers.get(&frame.peer_id) {
                            if let Some(sender) = peer.packet_senders.get(&frame.channel_id) {
                                tracing::event!(
                                    target: "tehuti::engine::meeting",
                                    tracing::Level::TRACE,
                                    "Meeting engine {:?} got packet frame for peer {:?} channel {:?}: {} bytes",
                                    engine_id,
                                    frame.peer_id,
                                    frame.channel_id,
                                    frame.data.data.len(),
                                );
                                if let Err(error) = sender.sender.send(frame.data).map_err(|err| {
                                    format!(
                                        "Meeting engine {:?} packet sender error (sending packet data): {:?}",
                                        engine_id, err
                                    )
                                }) {
                                    tracing::event!(
                                        target: "tehuti::engine::meeting",
                                        tracing::Level::ERROR,
                                        "Meeting engine {:?} failed to send packet data to peer {:?} channel {:?}: {:?}",
                                        engine_id,
                                        frame.peer_id,
                                        frame.channel_id,
                                        error,
                                    );
                                }
                            } else if self.config.warn_unknown_channel_packet {
                                tracing::event!(
                                    target: "tehuti::engine::meeting",
                                    tracing::Level::WARN,
                                    "Meeting engine {:?} got packet frame for unknown channel {:?} of peer {:?}",
                                    engine_id,
                                    frame.channel_id,
                                    frame.peer_id,
                                );
                            }
                        } else if self.config.warn_unknown_peer_packet {
                            tracing::event!(
                                target: "tehuti::engine::meeting",
                                tracing::Level::WARN,
                                "Meeting engine {:?} got packet frame for unknown peer {:?}",
                                engine_id,
                                frame.peer_id,
                            );
                        }
                    }
                }
            }
        }
        for engine_id in refresh_engines {
            if let Some((_, last_seen)) = self.engines.get_mut(&engine_id) {
                *last_seen = Instant::now();
            }
        }

        // Handle sending packets from peers to all sessions.
        for peer in self.peers.values() {
            for (channel_id, receiver) in &peer.packet_receivers {
                for data in receiver.receiver.iter() {
                    let size = data.data.len();
                    let recepients = data.recepients.to_owned();
                    let frame = ProtocolFrame::Packet(ProtocolPacketFrame {
                        peer_id: peer.info.peer_id,
                        channel_id: *channel_id,
                        data,
                    });
                    for (engine_id, (frames, _)) in &self.engines {
                        if !recepients.is_empty() && !recepients.contains(engine_id) {
                            continue;
                        }
                        tracing::event!(
                            target: "tehuti::engine::meeting",
                            tracing::Level::TRACE,
                            "Meeting engine {:?} sending packet frame for peer {:?} channel {:?}. {} bytes",
                            engine_id,
                            peer.info.peer_id,
                            channel_id,
                            size,
                        );
                        frames.sender.send(frame.clone()).map_err(|err| {
                            format!(
                                "Meeting engine {:?} frame sender error (forwarding packet frame): {:?}",
                                engine_id, err
                            )
                        })?;
                    }
                }
            }
        }

        // Handle meeting pump.
        if let Err(err) = self.meeting.pump_all() {
            tracing::event!(
                target: "tehuti::engine::meeting",
                tracing::Level::ERROR,
                "Meeting encountered error: {}. Terminating",
                err,
            );
            return Err(err);
        }

        // Handle meeting engine events.
        for event in self.engine_event.receiver.iter() {
            match event {
                MeetingEngineEvent::MeetingDestroyed => {
                    tracing::event!(
                        target: "tehuti::engine::meeting",
                        tracing::Level::TRACE,
                        "Meeting got destroyed. Terminating",
                    );
                    return Err("Meeting destroyed".into());
                }
                MeetingEngineEvent::PeerCreated(descriptor) => {
                    if let Entry::Vacant(entry) = self.peers.entry(descriptor.info.peer_id) {
                        tracing::event!(
                            target: "tehuti::engine::meeting",
                            tracing::Level::TRACE,
                            "Meeting created peer {:?}",
                            descriptor.info.peer_id,
                        );
                        if !descriptor.info.remote {
                            let frame = ProtocolFrame::Control(ProtocolControlFrame::CreatePeer(
                                descriptor.info.peer_id,
                                descriptor.info.role_id,
                            ));
                            for (engine_id, (frames, _)) in &self.engines {
                                tracing::event!(
                                    target: "tehuti::engine::meeting",
                                    tracing::Level::TRACE,
                                    "Meeting engine {:?} sending create peer {:?} with role {:?}",
                                    engine_id,
                                    descriptor.info.peer_id,
                                    descriptor.info.role_id,
                                );
                                frames.sender.send(frame.clone()).map_err(|err| {
                                    format!(
                                        "Meeting engine {:?} frame sender error (create peer): {:?}",
                                        engine_id, err
                                    )
                                })?;
                            }
                        }
                        entry.insert(descriptor);
                    } else if self.config.warn_duplicate_peer_creation {
                        tracing::event!(
                            target: "tehuti::engine::meeting",
                            tracing::Level::WARN,
                            "Meeting tries to create duplicate peer {:?} created",
                            descriptor.info.peer_id,
                        );
                    }
                }
                MeetingEngineEvent::PeerDestroyed(peer_id) => {
                    if self.peers.contains_key(&peer_id) {
                        tracing::event!(
                            target: "tehuti::engine::meeting",
                            tracing::Level::TRACE,
                            "Meeting destroyed peer {:?}",
                            peer_id,
                        );
                        let frame =
                            ProtocolFrame::Control(ProtocolControlFrame::DestroyPeer(peer_id));
                        for (engine_id, (frames, _)) in &self.engines {
                            tracing::event!(
                                target: "tehuti::engine::meeting",
                                tracing::Level::TRACE,
                                "Meeting engine {:?} sending destroy peer {:?}",
                                engine_id,
                                peer_id,
                            );
                            frames.sender.send(frame.clone()).map_err(|err| {
                                format!(
                                    "Meeting engine {:?} frame sender error (destroy peer): {:?}",
                                    engine_id, err
                                )
                            })?;
                        }
                        self.peers.remove(&peer_id);
                    } else if self.config.warn_unknown_peer_destruction {
                        tracing::event!(
                            target: "tehuti::engine::meeting",
                            tracing::Level::WARN,
                            "Meeting tries to destroy unknown peer {:?} destroyed",
                            peer_id,
                        );
                    }
                }
                event => {
                    if self.config.warn_unhandled_engine_event {
                        tracing::event!(
                            target: "tehuti::engine::meeting",
                            tracing::Level::WARN,
                            "Meeting got unhandled engine event: {:?}",
                            event,
                        );
                    }
                }
            }
        }

        Ok(())
    }

    pub fn run(
        mut self,
        interval: Duration,
        terminate_receiver: Receiver<()>,
    ) -> Result<JoinHandle<()>, Box<dyn Error>> {
        Ok(Builder::new()
            .name(self.meeting.name().to_string())
            .spawn(move || {
                loop {
                    if terminate_receiver.try_recv().is_some() {
                        tracing::event!(
                            target: "tehuti::engine::meeting",
                            tracing::Level::TRACE,
                            "Meeting terminating on request",
                        );
                        break;
                    }
                    if let Err(err) = self.maintain() {
                        tracing::event!(
                            target: "tehuti::engine::meeting",
                            tracing::Level::ERROR,
                            "Meeting terminated with error: {}",
                            err,
                        );
                        return;
                    }
                    sleep(interval);
                }
            })?)
    }
}

impl Future for EngineMeeting {
    type Output = ();

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        match self.maintain() {
            Ok(_) => {
                cx.waker().wake_by_ref();
                Poll::Pending
            }
            Err(e) => {
                tracing::event!(
                    target: "tehuti::engine::meeting",
                    tracing::Level::ERROR,
                    "Meeting encountered error: {}. Terminating",
                    e,
                );
                Poll::Ready(())
            }
        }
    }
}

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

    #[test]
    fn test_async() {
        fn is_send<T: Send>() {}

        is_send::<EnginePacketSender>();
        is_send::<EnginePacketReceiver>();
        is_send::<EnginePeerDescriptor>();
    }
}