New pubsub backend.

src/pubsub/events/groups.rs

@@ -0,0 +1,13 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct UserJoin {
+    pub id: String,
+    pub user: String,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct UserLeave {
+    pub id: String,
+    pub user: String,
+}

src/pubsub/events/guilds.rs

@@ -0,0 +1,33 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct UserJoin {
+    pub id: String,
+    pub user: String,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct UserLeave {
+    pub id: String,
+    pub user: String,
+    pub banned: bool,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct ChannelCreate {
+    pub id: String,
+    pub channel: String,
+    pub name: String,
+    pub description: String,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct ChannelDelete {
+    pub id: String,
+    pub channel: String,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct Delete {
+    pub id: String,
+}

src/pubsub/events/message.rs

@@ -0,0 +1,23 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct Create {
+    pub id: String,
+    pub nonce: Option<String>,
+    pub channel: String,
+    pub author: String,
+    pub content: String,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct Edit {
+    pub id: String,
+    pub channel: String,
+    pub author: String,
+    pub content: String,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct Delete {
+    pub id: String,
+}

src/pubsub/events/mod.rs

@@ -0,0 +1,31 @@
+use serde::{Deserialize, Serialize};
+
+pub mod groups;
+pub mod guilds;
+pub mod message;
+pub mod users;
+
+#[allow(non_camel_case_types)]
+#[derive(Serialize, Deserialize, Debug, Clone)]
+#[serde(tag = "type", content = "data")]
+pub enum Notification {
+    message_create(message::Create),
+    message_edit(message::Edit),
+    message_delete(message::Delete),
+    group_user_join(groups::UserJoin),
+    group_user_leave(groups::UserLeave),
+    guild_user_join(guilds::UserJoin),
+    guild_user_leave(guilds::UserLeave),
+    guild_channel_create(guilds::ChannelCreate),
+    guild_channel_delete(guilds::ChannelDelete),
+    guild_delete(guilds::Delete),
+    user_friend_status(users::FriendStatus),
+}
+
+impl Notification {    
+    pub fn push_to_cache(&self) {
+        //crate::database::channel::process_event(&self);
+        //crate::database::guild::process_event(&self);
+        //crate::database::user::process_event(&self);
+    }
+}

src/pubsub/events/users.rs

@@ -0,0 +1,8 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct FriendStatus {
+    pub id: String,
+    pub user: String,
+    pub status: i32,
+}

src/pubsub/hive.rs

@@ -0,0 +1,61 @@
+use super::events::Notification;
+use super::websocket;
+use crate::database::get_collection;
+
+use hive_pubsub::backend::mongo::{listen_thread, MongodbPubSub};
+use hive_pubsub::PubSub;
+use once_cell::sync::OnceCell;
+use serde_json::to_string;
+use log::{error, debug};
+
+static HIVE: OnceCell<MongodbPubSub<String, String, Notification>> = OnceCell::new();
+
+pub fn init_hive() {
+    let hive = MongodbPubSub::new(
+        |ids, notification| {
+            if let Ok(data) = to_string(&notification) {
+                debug!("Pushing out notification. {}", data);
+                if let Err(err) = websocket::publish(ids, data) {
+                    error!("Failed to publish notification through WebSocket! {}", err);
+                }
+            } else {
+                error!("Failed to serialise notification.");
+            }
+        },
+        get_collection("hive"),
+    );
+
+    listen_thread(hive.clone());
+
+    if HIVE.set(hive).is_err() {
+        panic!("Failed to set global pubsub instance.");
+    }
+}
+
+pub fn publish(topic: &String, data: Notification) -> Result<(), String> {
+    let hive = HIVE.get().expect("Global pubsub instance not available.");
+    hive.publish(topic, data)
+}
+
+pub fn subscribe(user: String, topics: Vec<String>) -> Result<(), String> {
+    let hive = HIVE.get().expect("Global pubsub instance not available.");
+    for topic in topics {
+        hive.subscribe(user.clone(), topic)?;
+    }
+
+    Ok(())
+}
+
+pub fn drop_user(user: &String) -> Result<(), String> {
+    let hive = HIVE.get().expect("Global pubsub instance not available.");
+    hive.drop_client(user)?;
+
+    Ok(())
+}
+
+pub fn drop_topic(topic: &String) -> Result<(), String> {
+    let hive = HIVE.get().expect("Global pubsub instance not available.");
+    hive.drop_topic(topic)?;
+
+    Ok(())
+}

src/pubsub/mod.rs

@@ -0,0 +1,3 @@
+pub mod events;
+pub mod hive;
+pub mod websocket;

src/pubsub/websocket/client.rs

@@ -0,0 +1,199 @@
+use super::state;
+use crate::database::get_collection;
+use crate::pubsub::hive;
+
+use log::{error, info};
+use mongodb::bson::doc;
+use mongodb::options::FindOneOptions;
+use serde_json::{from_str, json, Value};
+use ulid::Ulid;
+use ws::{CloseCode, Error, Handler, Handshake, Message, Result, Sender};
+
+pub struct Client {
+    id: String,
+    sender: Sender,
+    user_id: Option<String>,
+}
+
+impl Client {
+    pub fn new(sender: Sender) -> Client {
+        Client {
+            id: Ulid::new().to_string(),
+            user_id: None,
+            sender,
+        }
+    }
+}
+
+impl Handler for Client {
+    fn on_open(&mut self, handshake: Handshake) -> Result<()> {
+        info!("Client connected. [{}] {:?}", self.id, handshake.peer_addr);
+
+        Ok(())
+    }
+
+    // Client sends { "type": "authenticate", "token": token }.
+    // Receives { "type": "authorised" } and waits.
+    // Client then receives { "type": "ready", "data": payload }.
+    // If at any point we hit an error, send { "type": "error", "error": error }.
+    fn on_message(&mut self, msg: Message) -> Result<()> {
+        if let Message::Text(text) = msg {
+            if let Ok(data) = from_str(&text) as std::result::Result<Value, _> {
+                if let Value::String(packet_type) = &data["type"] {
+                    if packet_type == "authenticate" {
+                        if self.user_id.is_some() {
+                            return self.sender.send(
+                                json!({
+                                    "type": "error",
+                                    "error": "Already authenticated!"
+                                })
+                                .to_string(),
+                            );
+                        } else if let Value::String(token) = &data["token"] {
+                            let user = get_collection("users").find_one(
+                                doc! {
+                                    "access_token": token
+                                },
+                                FindOneOptions::builder()
+                                    .projection(doc! { "_id": 1 })
+                                    .build(),
+                            );
+
+                            if let Ok(result) = user {
+                                if let Some(doc) = result {
+                                    self.sender.send(
+                                        json!({
+                                            "type": "authorised"
+                                        })
+                                        .to_string(),
+                                    )?;
+
+                                    // FIXME: fetch above when we switch to new token system
+                                    // or auth cache system, something like that
+                                    let user = crate::database::user::fetch_user(
+                                        doc.get_str("_id").unwrap(),
+                                    )
+                                    .unwrap()
+                                    .unwrap(); // this should be guranteed, I think, maybe? I'm getting rid of it later. FIXME
+
+                                    self.user_id = Some(user.id.clone());
+
+                                    match user.create_payload() {
+                                        Ok(payload) => {
+                                            // ! Grab the ids from the payload,
+                                            // ! there's probably a better way to
+                                            // ! do this. I'll rewrite it at some point.
+                                            let mut ids = vec![
+                                                self.user_id.as_ref().unwrap().clone()
+                                            ];
+
+                                            {
+                                                // This is bad code. But to be fair
+                                                // it should work just fine.
+                                                for user in payload.get("users").unwrap().as_array().unwrap() {
+                                                    ids.push(user.as_object().unwrap().get("id").unwrap().as_str().unwrap().to_string());
+                                                }
+
+                                                for channel in payload.get("channels").unwrap().as_array().unwrap() {
+                                                    ids.push(channel.as_object().unwrap().get("id").unwrap().as_str().unwrap().to_string());
+                                                }
+
+                                                for guild in payload.get("guilds").unwrap().as_array().unwrap() {
+                                                    ids.push(guild.as_object().unwrap().get("id").unwrap().as_str().unwrap().to_string());
+                                                }
+                                            }
+
+                                            if let Err(err) = hive::subscribe(self.user_id.as_ref().unwrap().clone(), ids) {
+                                                self.sender.send(
+                                                    json!({
+                                                        "type": "warn",
+                                                        "error": "Failed to subscribe you to the Hive. You may not receive all notifications."
+                                                    })
+                                                    .to_string(),
+                                                )?;
+                                            }
+
+                                            self.sender.send(
+                                                json!({
+                                                    "type": "ready",
+                                                    "data": payload
+                                                })
+                                                .to_string(),
+                                            )?;
+
+                                            if state::accept(
+                                                self.id.clone(),
+                                                self.user_id.as_ref().unwrap().clone(),
+                                                self.sender.clone(),
+                                            )
+                                            .is_err()
+                                            {
+                                                self.sender.send(
+                                                    json!({
+                                                        "type": "warn",
+                                                        "error": "Failed to accept your connection. You will not receive any notifications."
+                                                    })
+                                                    .to_string(),
+                                                )?;
+                                            }
+                                        }
+                                        Err(error) => {
+                                            error!("Failed to create payload! {}", error);
+                                            self.sender.send(
+                                                json!({
+                                                    "type": "error",
+                                                    "error": "Failed to create payload."
+                                                })
+                                                .to_string(),
+                                            )?;
+                                        }
+                                    }
+                                } else {
+                                    self.sender.send(
+                                        json!({
+                                            "type": "error",
+                                            "error": "Invalid token."
+                                        })
+                                        .to_string(),
+                                    )?;
+                                }
+                            } else {
+                                self.sender.send(
+                                    json!({
+                                        "type": "error",
+                                        "error": "Failed to fetch from database."
+                                    })
+                                    .to_string(),
+                                )?;
+                            }
+                        } else {
+                            self.sender.send(
+                                json!({
+                                    "type": "error",
+                                    "error": "Missing token."
+                                })
+                                .to_string(),
+                            )?;
+                        }
+                    }
+                }
+            }
+        }
+
+        Ok(())
+    }
+
+    fn on_close(&mut self, _code: CloseCode, reason: &str) {
+        info!("Client disconnected. [{}] {}", self.id, reason);
+        if let Err(error) = state::drop(&self.id) {
+            error!("Also failed to drop client from state! {}", error);
+        }
+    }
+
+    fn on_error(&mut self, err: Error) {
+        error!(
+            "A client disconnected due to an error. [{}] {}",
+            self.id, err
+        );
+    }
+}

src/pubsub/websocket/mod.rs

@@ -0,0 +1,23 @@
+use crate::util::variables::WS_HOST;
+
+use log::{error, info};
+use std::thread;
+use ws::listen;
+
+mod client;
+mod state;
+
+pub use state::publish;
+
+pub fn launch_server() {
+    thread::spawn(|| {
+        if listen(WS_HOST.to_string(), |sender| client::Client::new(sender)).is_err() {
+            error!(
+                "Failed to listen for WebSocket connections on {:?}!",
+                *WS_HOST
+            );
+        } else {
+            info!("Listening for WebSocket connections on {:?}", *WS_HOST);
+        }
+    });
+}

src/pubsub/websocket/state.rs

@@ -0,0 +1,89 @@
+use crate::pubsub::hive;
+
+use many_to_many::ManyToMany;
+use std::collections::HashMap;
+use std::sync::{Arc, RwLock};
+use log::{error, info};
+use ws::Sender;
+
+lazy_static! {
+    static ref CONNECTIONS: Arc<RwLock<HashMap<String, Sender>>> =
+        Arc::new(RwLock::new(HashMap::new()));
+    static ref CLIENTS: Arc<RwLock<ManyToMany<String, String>>> =
+        Arc::new(RwLock::new(ManyToMany::new()));
+}
+
+pub fn accept(id: String, user_id: String, sender: Sender) -> Result<(), String> {
+    let mut conns = CONNECTIONS
+        .write()
+        .map_err(|_| "Failed to lock connections for writing.")?;
+    
+    conns.insert(id.clone(), sender);
+
+    let mut clients = CLIENTS
+        .write()
+        .map_err(|_| "Failed to lock clients for writing.")?;
+    
+    clients.insert(user_id.clone(), id.clone());
+
+    info!("Accepted user [{}] for connection {}.", user_id, id);
+    Ok(())
+}
+
+pub fn drop(id: &String) -> Result<(), String> {
+    let mut conns = CONNECTIONS
+        .write()
+        .map_err(|_| "Failed to lock connections for writing.")?;
+    
+    conns.remove(id);
+
+    let mut clients = CLIENTS
+        .write()
+        .map_err(|_| "Failed to lock clients for writing.")?;
+    
+    let uid = if let Some(ids) = clients.get_right(id) {
+        let user_id: String = ids.into_iter().next().unwrap();
+        info!("Dropped user [{}] for connection {}.", user_id, id);
+        Some(user_id)
+    } else {
+        None
+    };
+
+    clients.remove_right(id);
+
+    if let Some(user_id) = &uid {
+        if let None = clients.get_left(user_id) {
+            if let Err(error) = hive::drop_user(user_id) {
+                error!("Failed to drop user from hive! {}", error);
+            } else {
+                info!("User [{}] has completed disconnected from node.", user_id);
+            }
+        }
+    }
+
+    Ok(())
+}
+
+pub fn publish(clients: Vec<String>, data: String) -> Result<(), String> {
+    let conns = CONNECTIONS
+        .read()
+        .map_err(|_| "Failed to lock connections for reading.")?;
+
+    let client_map = CLIENTS
+        .read()
+        .map_err(|_| "Failed to lock clients for reading.")?;
+    
+    for client in clients {
+        if let Some(targets) = client_map.get_left(&client) {
+            for target in &targets {
+                if let Some(connection) = conns.get(target) {
+                    if let Err(err) = connection.send(data.clone()) {
+                        error!("Failed to publish notification to client [{}]! {}", target, err);
+                    }
+                }
+            }
+        }
+    }
+
+    Ok(())
+}