1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195

//use bs58::decode;
use super::Rtc;
use crate::{node::user_accounts::UserAccounts, utilities::timestamp};
use libp2p::PeerId;
use prost::Message;

pub struct RtcManaging {}
impl RtcManaging {
    /// process session list from cli
    pub fn session_list(_my_user_id: &PeerId) -> super::proto_rpc::RtcSessionListResponse {
        let mut res = super::proto_rpc::RtcSessionListResponse { sessions: vec![] };

        let sessions = super::RTCSESSIONS.get().read().unwrap();
        for (_id, session) in sessions.sessions.iter() {
            let entry = super::proto_rpc::RtcSession {
                group_id: session.group_id.clone(),
                session_type: session.session_type,
                state: session.state as u32,
                created_at: session.created_at,
            };
            res.sessions.push(entry);
        }
        res
    }
    /// process session request from cli
    pub fn session_request(
        my_user_id: &PeerId,
        req: &super::proto_rpc::RtcSessionRequest,
    ) -> Result<Vec<u8>, String> {
        //check if session already exists
        if let Some(_session) = super::Rtc::get_session_from_id(&req.group_id) {
            return Err("session alrady exists!".to_string());
        }

        //insert new session entry
        let session = super::RtcSession {
            user_id: my_user_id.to_bytes(),
            group_id: req.group_id.clone(),
            session_type: 1,
            state: 1, //sent request
            created_at: timestamp::Timestamp::get_timestamp(),
        };
        super::Rtc::update_session(session);

        //send session request message on the messaging service
        let proto_message = super::proto_net::RtcContainer {
            message: Some(super::proto_net::rtc_container::Message::RtcSessionRequest(
                super::proto_net::RtcSessionRequest { session_type: 1 },
            )),
        };

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

        if let Some(user_account) = UserAccounts::get_by_id(*my_user_id) {
            let receiver = PeerId::from_bytes(&req.group_id).unwrap();
            super::Rtc::send_rtc_message_through_message(&user_account, receiver, &message_buff);
        } else {
            return Err("user account has problem".to_string());
        }

        Ok(req.group_id.clone())
    }

    /// send session management message
    fn send_session_management(
        my_user_id: &PeerId,
        group_id: &Vec<u8>,
        option: u32,
    ) -> Result<bool, String> {
        //send message on the messaging service
        let proto_message = super::proto_net::RtcContainer {
            message: Some(
                super::proto_net::rtc_container::Message::RtcSessionManagement(
                    super::proto_net::RtcSessionManagement { option },
                ),
            ),
        };

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

        if let Some(user_account) = UserAccounts::get_by_id(*my_user_id) {
            let receiver = PeerId::from_bytes(group_id).unwrap();
            super::Rtc::send_rtc_message_through_message(&user_account, receiver, &message_buff);
        } else {
            return Err("user account has problem".to_string());
        }
        Ok(true)
    }

    /// process session mangement request from cli
    pub fn session_management(
        my_user_id: &PeerId,
        req: &super::proto_rpc::RtcSessionManagement,
    ) -> Result<Vec<u8>, String> {
        //check if session already exists
        match super::Rtc::get_session_from_id(&req.group_id) {
            Some(mut session) => {
                match req.option {
                    1 => {
                        //update entry
                        session.created_at = timestamp::Timestamp::get_timestamp();
                        session.state = 3;
                        super::Rtc::update_session(session.clone());
                    }
                    2 => {
                        //decline
                        super::Rtc::remove_session(&req.group_id);
                    }
                    3 => {
                        //end
                        super::Rtc::remove_session(&req.group_id);
                    }
                    _ => {
                        return Err("unknown session management option".to_string());
                    }
                }
                if let Err(error) =
                    Self::send_session_management(my_user_id, &session.group_id.clone(), req.option)
                {
                    return Err(error);
                }
                Ok(session.group_id.clone())
            }
            None => Err("session dese not exists!".to_string()),
        }
    }

    /// process session request from network
    pub fn on_session_request(
        sender_id: &PeerId,
        receiver_id: &PeerId,
        _req: &super::proto_net::RtcSessionRequest,
    ) {
        // check session already exist
        if let Some(_session) = super::Rtc::get_session_from_id(&sender_id.to_bytes()) {
            log::error!("session already exists id: {}", sender_id.to_string());
            return;
        }

        // make new entry
        let session = super::RtcSession {
            user_id: receiver_id.to_bytes(),
            group_id: sender_id.to_bytes(),
            session_type: 1,
            state: 2, //received state
            created_at: timestamp::Timestamp::get_timestamp(),
        };
        super::Rtc::update_session(session);
    }

    /// process session request from network
    pub fn on_session_management(
        sender_id: &PeerId,
        _receiver_id: &PeerId,
        req: &super::proto_net::RtcSessionManagement,
    ) {
        match super::Rtc::get_session_from_id(&sender_id.to_bytes()) {
            Some(mut session) => {
                match req.option {
                    1 => {
                        //established
                        session.state = 3;
                        session.created_at = timestamp::Timestamp::get_timestamp();
                        Rtc::update_session(session);
                    }
                    2 => {
                        //decline
                        super::Rtc::remove_session(&sender_id.to_bytes());
                    }
                    3 => {
                        //end
                        super::Rtc::remove_session(&sender_id.to_bytes());
                    }
                    _ => {
                        log::error!("session management unknown option: {}", req.option);
                    }
                }
            }
            None => {
                log::error!(
                    "session management dose not exist session: {}",
                    sender_id.to_string()
                );
            }
        }
        // save chat messge
        //Chat::save_incoming_group_invite_reply_message(receiver_id.clone(), sender_id.clone(), &resp.group_id, resp.accept, signature);
    }
}