// Copyright (c) 2022 Open Community Project Association https://ocpa.ch
// This software is published under the AGPLv3 license.

//! # Qaul Real Time Communication (RTC) Module
//!
//! The RTC module manages and forwards the session requests

use libp2p::PeerId;

use crate::node::user_accounts::UserAccount;
use crate::rpc::Rpc;
use prost::Message;
use state::Storage;
use std::collections::BTreeMap;
use std::sync::RwLock;

mod rtc_managing;
mod rtc_messaging;

use super::chat::Chat;
use super::group;
use super::group::group_id::GroupId;
use super::messaging::{proto, Messaging, MessagingServiceType};
use crate::utilities::timestamp;
use rtc_managing::RtcManaging;
use rtc_messaging::RtcMessaging;

/// Import protobuf message definition generated by
/// the rust module prost-build.
pub mod proto_rpc {
    include!("qaul.rpc.rtc.rs");
}
pub mod proto_net {
    include!("qaul.net.rtc.rs");
}

// Structure of rtc session
#[derive(Clone)]
pub struct RtcSession {
    // user id
    pub user_id: Vec<u8>,
    // group id
    pub group_id: Vec<u8>,
    // session type
    pub session_type: u32,
    // created at
    pub created_at: u64,
    // state 1: sent req, 2:received req, 3: established
    pub state: u8,
}

/// Structure to management for sessions.
#[derive(Clone)]
pub struct RtcSessions {
    // id mapping group id => session
    pub sessions: BTreeMap<Vec<u8>, RtcSession>,
}

/// mutable state for sessions
pub static RTCSESSIONS: Storage<RwLock<RtcSessions>> = Storage::new();

/// Real Time Communication Module
pub struct Rtc {}

impl Rtc {
    /// initialize group chat module
    pub fn init() {
        let rtc_sessions = RtcSessions {
            sessions: BTreeMap::new(),
        };
        RTCSESSIONS.set(RwLock::new(rtc_sessions));
    }

    /// get session from session_id
    pub fn get_session_from_id(group_id: &Vec<u8>) -> Option<RtcSession> {
        let sessions = RTCSESSIONS.get().read().unwrap();
        if sessions.sessions.contains_key(group_id) {
            return Some(sessions.sessions.get(group_id).unwrap().clone());
        }
        None
    }

    /// get session from session_id
    pub fn update_session(session: RtcSession) {
        let mut sessions = RTCSESSIONS.get().write().unwrap();
        sessions.sessions.insert(session.group_id.clone(), session);
    }

    /// remove session on the storage
    pub fn remove_session(session_id: &Vec<u8>) {
        let mut sessions = RTCSESSIONS.get().write().unwrap();
        sessions.sessions.remove(session_id);
    }

    /// Send capsuled group message through messaging service
    pub fn send_rtc_message_through_message(
        user_account: &UserAccount,
        receiver: PeerId,
        data: &Vec<u8>,
    ) {
        // create direct chat room
        let group_id = GroupId::from_peers(&user_account.id, &receiver);
        if !group::GroupStorage::group_exists(user_account.id, group_id.to_bytes()) {
            group::GroupManage::create_new_direct_chat_group(&user_account.id, &receiver);
        }

        //get last index
        let group;
        match group::GroupStorage::get_group(user_account.id, group_id.to_bytes()) {
            Some(v) => group = v,
            None => return,
        }

        let my_member;
        match group.get_member(&user_account.id.to_bytes()) {
            Some(v) => {
                my_member = v.clone();
            }
            _ => {
                return;
            }
        }
        let last_index = my_member.last_message_index + 1;

        let message_id = Chat::generate_message_id(&group.id, &user_account.id, last_index);
        let common_message = proto::CommonMessage {
            message_id: message_id.clone(),
            group_id: group_id.to_bytes(),
            sent_at: timestamp::Timestamp::get_timestamp(),
            payload: Some(proto::common_message::Payload::RtcMessage(
                proto::RtcMessage {
                    content: data.clone(),
                },
            )),
        };
        let send_message = proto::Messaging {
            message: Some(proto::messaging::Message::CommonMessage(
                common_message.clone(),
            )),
        };

        // send message via messaging
        if let Err(e) = Messaging::pack_and_send_message(
            user_account,
            &receiver,
            send_message.encode_to_vec(),
            MessagingServiceType::Rtc,
            &message_id,
            true,
        ) {
            log::error!("rtc message sending failed {}", e.to_string());
        }
    }

    /// Process incoming NET messages for rtc module
    pub fn net(sender_id: &PeerId, receiver_id: &PeerId, data: &Vec<u8>) {
        match proto_net::RtcContainer::decode(&data[..]) {
            Ok(messaging) => match messaging.message {
                Some(proto_net::rtc_container::Message::RtcSessionRequest(session_req)) => {
                    log::error!("on_session_request");
                    rtc_managing::RtcManaging::on_session_request(
                        &sender_id,
                        &receiver_id,
                        &session_req,
                    );
                }
                Some(proto_net::rtc_container::Message::RtcSessionManagement(session_mgr)) => {
                    log::error!("on_session_management");
                    rtc_managing::RtcManaging::on_session_management(
                        &sender_id,
                        &receiver_id,
                        &session_mgr,
                    );
                }
                // Some(proto_net::rtc_container::Message::RtcMessage(message)) => {
                //     log::error!("on_session_message");
                //     rtc_messaging::RtcMessaging::on_message(
                //         &sender_id,
                //         &receiver_id,
                //         &message,
                //         signature,
                //     );
                // }
                _ => {
                    log::error!("rtc message from {} was empty", sender_id.to_base58())
                }
            },
            Err(e) => {
                log::error!(
                    "Error decoding Rtc Message from {} to {}: {}",
                    sender_id.to_base58(),
                    receiver_id.to_base58(),
                    e
                );
            }
        }
    }

    /// Process incoming RPC request messages
    pub fn rpc(data: Vec<u8>, user_id: Vec<u8>) {
        let my_user_id = PeerId::from_bytes(&user_id).unwrap();

        match proto_rpc::RtcRpc::decode(&data[..]) {
            Ok(rtc_rpc) => {
                match rtc_rpc.message {
                    Some(proto_rpc::rtc_rpc::Message::RtcSessionRequest(session_req)) => {
                        match RtcManaging::session_request(&my_user_id, &session_req) {
                            Err(error) => {
                                log::error!("rtc request error {}", error);
                            }
                            Ok(_id) => {
                                // make response
                                let proto_message = proto_rpc::RtcRpc {
                                    message: Some(proto_rpc::rtc_rpc::Message::RtcSessionResponse(
                                        proto_rpc::RtcSessionResponse {
                                            group_id: session_req.group_id.clone(),
                                        },
                                    )),
                                };

                                // encode message
                                let mut buf = Vec::with_capacity(proto_message.encoded_len());
                                proto_message
                                    .encode(&mut buf)
                                    .expect("Vec<u8> provides capacity as needed");

                                // send message
                                Rpc::send_message(
                                    buf,
                                    crate::rpc::proto::Modules::Rtc.into(),
                                    "".to_string(),
                                    Vec::new(),
                                );
                            }
                        }
                    }
                    Some(proto_rpc::rtc_rpc::Message::RtcSessionManagement(session_req)) => {
                        if let Err(error) =
                            RtcManaging::session_management(&my_user_id, &session_req)
                        {
                            log::error!("rtc management error {}", error);
                        }
                    }
                    Some(proto_rpc::rtc_rpc::Message::RtcOutgoing(session_req)) => {
                        if let Err(error) = RtcMessaging::send_message(&my_user_id, &session_req) {
                            log::error!("rtc message error {}", error);
                        }
                    }
                    Some(proto_rpc::rtc_rpc::Message::RtcSessionListRequest(_session_req)) => {
                        let res = RtcManaging::session_list(&my_user_id);
                        //make response
                        let proto_message = proto_rpc::RtcRpc {
                            message: Some(proto_rpc::rtc_rpc::Message::RtcSessionListResponse(res)),
                        };

                        // encode message
                        let mut buf = Vec::with_capacity(proto_message.encoded_len());
                        proto_message
                            .encode(&mut buf)
                            .expect("Vec<u8> provides capacity as needed");

                        // send message
                        Rpc::send_message(
                            buf,
                            crate::rpc::proto::Modules::Rtc.into(),
                            "".to_string(),
                            Vec::new(),
                        );
                    }

                    _ => {
                        log::error!("Unhandled Protobuf RTC message");
                    }
                }
            }
            Err(error) => {
                log::error!("{:?}", error);
            }
        }
    }
}