diff --git a/Cargo.lock b/Cargo.lock
index d10c57f0d75f0f700fd6c81fa06721b21ddb55a4..dca41144e8ef5f990c93b8c0096d513a77fdfd91 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -835,6 +835,19 @@ version = "0.4.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "644f9158b2f133fd50f5fb3242878846d9eb792e445c893805ff0e3824006e35"
+[[package]]
+name = "hive_pubsub"
+version = "0.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e2b08d5028b638db3f5eb475dc7c5e5aad1b0a5b4f737b2c394883034330853e"
+dependencies = [
+ "many-to-many",
+ "mongodb",
+ "serde",
+ "serde_json",
+ "ulid",
+]
+
[[package]]
name = "hmac"
version = "0.7.1"
@@ -1218,6 +1231,12 @@ dependencies = [
"linked-hash-map",
]
+[[package]]
+name = "many-to-many"
+version = "0.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "28806b11671f3ae32bfcacf1948d77af5bddf1aa4ac518207615f11abe31df19"
+
[[package]]
name = "match_cfg"
version = "0.1.0"
@@ -1326,9 +1345,9 @@ dependencies = [
[[package]]
name = "mongodb"
-version = "1.1.0"
+version = "1.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f8cb54f85e6380c8cfe130a55a85cfc2584e8317f4bb8bda464d367fd4f24957"
+checksum = "a726495d7418c4579ecc9465b53ddade8187c0299e5db0e7250672b67d196913"
dependencies = [
"async-std",
"async-trait",
@@ -2031,7 +2050,7 @@ dependencies = [
[[package]]
name = "revolt"
-version = "0.2.10"
+version = "0.2.11"
dependencies = [
"bcrypt",
"bitfield",
@@ -2039,10 +2058,12 @@ dependencies = [
"dotenv",
"env_logger",
"hashbrown",
+ "hive_pubsub",
"lazy_static",
"lettre",
"log 0.4.11",
"lru",
+ "many-to-many",
"mongodb",
"num_enum",
"once_cell",
diff --git a/Cargo.toml b/Cargo.toml
index c08634cbcc2df5685e01c3e115edcfb5afdb01f8..5016d6a6de26b22b2216cfaa775e55d11c1e78fb 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -1,13 +1,13 @@
[package]
name = "revolt"
-version = "0.2.10"
+version = "0.2.11"
authors = ["Paul Makles <paulmakles@gmail.com>"]
edition = "2018"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies]
-mongodb = { version = "1.1.0", default-features = false, features = ["sync"] } # FIXME: rewrite database with async API
+mongodb = { version = "1.1.1", default-features = false, features = ["sync"] } # FIXME: rewrite database with async API
rocket = { version = "0.4.5", default-features = false }
once_cell = "1.4.1"
dotenv = "0.15.0"
@@ -31,3 +31,5 @@ lazy_static = "1.4.0"
log = "0.4.11"
env_logger = "0.7.1"
lettre = "0.10.0-alpha.1"
+hive_pubsub = { version = "0.3.1", features = ["mongo"] }
+many-to-many = "0.1.2"
diff --git a/src/database/guild.rs b/src/database/guild.rs
index a7a90dbed228e5d6c1700c7cf5c353e9f2b97f9e..3edde9c42adf9e459756f79a9b2d8b4e5bd943ea 100644
--- a/src/database/guild.rs
+++ b/src/database/guild.rs
@@ -107,6 +107,7 @@ pub fn fetch_guild(id: &str) -> Result<Option<Guild>, String> {
let col = get_collection("guilds");
if let Ok(result) = col.find_one(doc! { "_id": id }, None) {
if let Some(doc) = result {
+ dbg!(doc.to_string());
if let Ok(guild) = from_bson(Bson::Document(doc)) as Result<Guild, _> {
let mut cache = CACHE.lock().unwrap();
cache.put(id.to_string(), guild.clone());
@@ -152,6 +153,7 @@ pub fn fetch_guilds(ids: &Vec<String>) -> Result<Vec<Guild>, String> {
for item in result {
let mut cache = CACHE.lock().unwrap();
if let Ok(doc) = item {
+ dbg!(doc.to_string());
if let Ok(guild) = from_bson(Bson::Document(doc)) as Result<Guild, _> {
cache.put(guild.id.clone(), guild.clone());
guilds.push(guild);
diff --git a/src/database/message.rs b/src/database/message.rs
index f20d061a9ee6ed8469b9158d853f049cd1c1c918..688365ff41267e787b72a580f1673a8f10137797 100644
--- a/src/database/message.rs
+++ b/src/database/message.rs
@@ -3,6 +3,7 @@ use crate::database::channel::Channel;
use crate::notifications;
use crate::notifications::events::message::Create;
use crate::notifications::events::Notification;
+use crate::pubsub::hive;
use crate::routes::channel::ChannelType;
use mongodb::bson::from_bson;
@@ -11,6 +12,8 @@ use rocket::http::RawStr;
use rocket::request::FromParam;
use serde::{Deserialize, Serialize};
+use log::warn;
+
#[derive(Serialize, Deserialize, Debug)]
pub struct PreviousEntry {
pub content: String,
@@ -51,6 +54,23 @@ impl Message {
&target,
);
+ if hive::publish(
+ &target.id,
+ crate::pubsub::events::Notification::message_create(
+ crate::pubsub::events::message::Create {
+ id: self.id.clone(),
+ nonce: self.nonce.clone(),
+ channel: self.channel.clone(),
+ author: self.author.clone(),
+ content: self.content.clone(),
+ },
+ ),
+ )
+ .is_err()
+ {
+ warn!("Saved message but couldn't send notification.");
+ }
+
let short_content: String = self.content.chars().take(24).collect();
// !! this stuff can be async
diff --git a/src/main.rs b/src/main.rs
index eb2307ae8987d5106fc35e0379d60437e6fd1978..d47bb69aa8b453dee8eab0b185d1aada400e222b 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -11,33 +11,38 @@ extern crate lazy_static;
pub mod database;
pub mod notifications;
+pub mod pubsub;
pub mod routes;
pub mod util;
+use log::info;
use rocket_cors::AllowedOrigins;
use std::thread;
-use log::info;
fn main() {
dotenv::dotenv().ok();
- env_logger::init_from_env(
- env_logger::Env::default()
- .filter_or("RUST_LOG", "info")
- );
+ env_logger::init_from_env(env_logger::Env::default().filter_or("RUST_LOG", "info"));
info!("Starting REVOLT server.");
util::variables::preflight_checks();
database::connect();
+
+ // ! START OLD NOTIF CODE
notifications::start_worker();
thread::spawn(|| {
notifications::pubsub::launch_subscriber();
});
- thread::spawn(|| {
+ notifications::state::init();
+ /*thread::spawn(|| {
notifications::ws::launch_server();
- });
+ });*/
+ // ! END OLD NOTIF CODE
+
+ pubsub::hive::init_hive();
+ pubsub::websocket::launch_server();
let cors = rocket_cors::CorsOptions {
allowed_origins: AllowedOrigins::All,
diff --git a/src/notifications/state.rs b/src/notifications/state.rs
index 9b063ec4178408470b760caa9f8d304bfec891ca..cfa1a3f2552c5af91e20c3d34e98b77a052b2b4e 100644
--- a/src/notifications/state.rs
+++ b/src/notifications/state.rs
@@ -14,17 +14,15 @@ pub enum StateResult {
Success(String),
}
-static mut CONNECTIONS: OnceCell<RwLock<HashMap<String, Sender>>> = OnceCell::new();
+static CONNECTIONS: OnceCell<RwLock<HashMap<String, Sender>>> = OnceCell::new();
pub fn add_connection(id: String, sender: Sender) {
- unsafe {
- CONNECTIONS
- .get()
- .unwrap()
- .write()
- .unwrap()
- .insert(id, sender);
- }
+ CONNECTIONS
+ .get()
+ .unwrap()
+ .write()
+ .unwrap()
+ .insert(id, sender);
}
pub struct User {
@@ -148,33 +146,29 @@ impl GlobalState {
}
}
- unsafe {
- CONNECTIONS
- .get()
- .unwrap()
- .write()
- .unwrap()
- .remove(&connection);
- }
+ CONNECTIONS
+ .get()
+ .unwrap()
+ .write()
+ .unwrap()
+ .remove(&connection);
}
}
-pub static mut DATA: OnceCell<RwLock<GlobalState>> = OnceCell::new();
+pub static DATA: OnceCell<RwLock<GlobalState>> = OnceCell::new();
pub fn init() {
- unsafe {
- if CONNECTIONS.set(RwLock::new(HashMap::new())).is_err() {
- panic!("Failed to set global connections map.");
- }
+ if CONNECTIONS.set(RwLock::new(HashMap::new())).is_err() {
+ panic!("Failed to set global connections map.");
+ }
- if DATA.set(RwLock::new(GlobalState::new())).is_err() {
- panic!("Failed to set global state.");
- }
+ if DATA.set(RwLock::new(GlobalState::new())).is_err() {
+ panic!("Failed to set global state.");
}
}
pub fn send_message(users: Option<Vec<String>>, guild: Option<String>, data: String) {
- let state = unsafe { DATA.get().unwrap().read().unwrap() };
+ let state = DATA.get().unwrap().read().unwrap();
let mut connections = HashSet::new();
let mut users = vec_to_set(&users.unwrap_or(vec![]));
@@ -194,7 +188,7 @@ pub fn send_message(users: Option<Vec<String>>, guild: Option<String>, data: Str
}
}
- let targets = unsafe { CONNECTIONS.get().unwrap().read().unwrap() };
+ let targets = CONNECTIONS.get().unwrap().read().unwrap();
for conn in connections {
if let Some(sender) = targets.get(&conn) {
if sender.send(data.clone()).is_err() {
diff --git a/src/pubsub/events/groups.rs b/src/pubsub/events/groups.rs
new file mode 100644
index 0000000000000000000000000000000000000000..876f13d163faa0a0b8545a6196301038926fb064
--- /dev/null
+++ b/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,
+}
diff --git a/src/pubsub/events/guilds.rs b/src/pubsub/events/guilds.rs
new file mode 100644
index 0000000000000000000000000000000000000000..3b1b95d2052ca750b17aa1f89d6b7151e8c31acb
--- /dev/null
+++ b/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,
+}
diff --git a/src/pubsub/events/message.rs b/src/pubsub/events/message.rs
new file mode 100644
index 0000000000000000000000000000000000000000..ffec878984282a769efff93bee04a98c3ddea214
--- /dev/null
+++ b/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,
+}
diff --git a/src/pubsub/events/mod.rs b/src/pubsub/events/mod.rs
new file mode 100644
index 0000000000000000000000000000000000000000..597502b3f8102f8af090f335f5bfe0336f494c64
--- /dev/null
+++ b/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);
+ }
+}
diff --git a/src/pubsub/events/users.rs b/src/pubsub/events/users.rs
new file mode 100644
index 0000000000000000000000000000000000000000..7a23446e295c3e76534c9a69a1a1d31b9436ae67
--- /dev/null
+++ b/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,
+}
diff --git a/src/pubsub/hive.rs b/src/pubsub/hive.rs
new file mode 100644
index 0000000000000000000000000000000000000000..c157baa4569c9d7fdd1f831ba0e4e3841cee38d5
--- /dev/null
+++ b/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(¬ification) {
+ 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(())
+}
diff --git a/src/pubsub/mod.rs b/src/pubsub/mod.rs
new file mode 100644
index 0000000000000000000000000000000000000000..23abbfd52dd5c810e66c3b4c87e2c6ec3b29d821
--- /dev/null
+++ b/src/pubsub/mod.rs
@@ -0,0 +1,3 @@
+pub mod events;
+pub mod hive;
+pub mod websocket;
diff --git a/src/pubsub/websocket/client.rs b/src/pubsub/websocket/client.rs
new file mode 100644
index 0000000000000000000000000000000000000000..2a491704bf13c9ca3c7879ba24fe9e24711ff1a3
--- /dev/null
+++ b/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
+ );
+ }
+}
diff --git a/src/pubsub/websocket/mod.rs b/src/pubsub/websocket/mod.rs
new file mode 100644
index 0000000000000000000000000000000000000000..e161253ea783e1404995fb139ddaae82e94e58ec
--- /dev/null
+++ b/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);
+ }
+ });
+}
diff --git a/src/pubsub/websocket/state.rs b/src/pubsub/websocket/state.rs
new file mode 100644
index 0000000000000000000000000000000000000000..ec02d9fedf29b5e4541faad90516377f8711f74c
--- /dev/null
+++ b/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(())
+}
diff --git a/src/routes/account.rs b/src/routes/account.rs
index 01c565e18210ee9036eedfe3a7b404b880a045d9..ed19b3702ae9905483e722ccb9e0a69cb0d06b15 100644
--- a/src/routes/account.rs
+++ b/src/routes/account.rs
@@ -1,7 +1,7 @@
use super::Response;
use crate::database;
-use crate::util::{captcha, email, gen_token};
use crate::util::variables::{DISABLE_REGISTRATION, USE_EMAIL};
+use crate::util::{captcha, email, gen_token};
use bcrypt::{hash, verify};
use chrono::prelude::*;
@@ -84,7 +84,7 @@ pub fn create(info: Json<Create>) -> Response {
},
false => doc! {
"verified": true
- }
+ },
};
let id = Ulid::new().to_string();
diff --git a/src/routes/root.rs b/src/routes/root.rs
index 3026701be41e4fa0fae29f286441a947e78d0d84..af33c5ee516da6c342e866c299288a0279f96085 100644
--- a/src/routes/root.rs
+++ b/src/routes/root.rs
@@ -1,5 +1,5 @@
use super::Response;
-use crate::util::variables::{USE_EMAIL, DISABLE_REGISTRATION, USE_HCAPTCHA};
+use crate::util::variables::{DISABLE_REGISTRATION, HCAPTCHA_SITEKEY, USE_EMAIL, USE_HCAPTCHA};
use mongodb::bson::doc;
@@ -7,15 +7,13 @@ use mongodb::bson::doc;
#[get("/")]
pub fn root() -> Response {
Response::Success(json!({
- "revolt": "0.2.10",
- "version": {
- "major": 0,
- "minor": 2,
- "patch": 10
- },
+ "revolt": "0.2.11",
"features": {
"registration": !*DISABLE_REGISTRATION,
- "captcha": *USE_HCAPTCHA,
+ "captcha": {
+ "enabled": *USE_HCAPTCHA,
+ "key": HCAPTCHA_SITEKEY.to_string()
+ },
"email": *USE_EMAIL,
}
}))
diff --git a/src/util/variables.rs b/src/util/variables.rs
index 794658af0b8f5e5e7b6aa2966bf693f4ece25431..682e0326c4a27834ef991875ac9da4086bbdee04 100644
--- a/src/util/variables.rs
+++ b/src/util/variables.rs
@@ -10,10 +10,12 @@ lazy_static! {
pub static ref PUBLIC_URL: String =
env::var("REVOLT_PUBLIC_URL").expect("Missing REVOLT_PUBLIC_URL environment variable.");
pub static ref HCAPTCHA_KEY: String =
- env::var("REVOLT_HCAPTCHA_KEY").unwrap_or_else(|_| "".to_string());
+ env::var("REVOLT_HCAPTCHA_KEY").unwrap_or_else(|_| "0x0000000000000000000000000000000000000000".to_string());
+ pub static ref HCAPTCHA_SITEKEY: String =
+ env::var("REVOLT_HCAPTCHA_SITEKEY").unwrap_or_else(|_| "10000000-ffff-ffff-ffff-000000000001".to_string());
pub static ref WS_HOST: String =
- env::var("REVOLT_WS_HOST").unwrap_or_else(|_| "0.0.0.0:9999".to_string());
-
+ env::var("REVOLT_WS_HOST").unwrap_or_else(|_| "0.0.0.0:9000".to_string());
+
// Application Flags
pub static ref DISABLE_REGISTRATION: bool = env::var("REVOLT_DISABLE_REGISTRATION").map_or(false, |v| v == "*1");
pub static ref USE_EMAIL: bool = env::var("REVOLT_USE_EMAIL_VERIFICATION").map_or(
@@ -39,8 +41,7 @@ pub fn preflight_checks() {
if *USE_EMAIL == false {
#[cfg(not(debug_assertions))]
{
- if !env::var("REVOLT_UNSAFE_NO_EMAIL")
- .map_or(false, |v| v == *"1") {
+ if !env::var("REVOLT_UNSAFE_NO_EMAIL").map_or(false, |v| v == *"1") {
panic!(
"Not letting you run this in production, set REVOLT_UNSAFE_NO_EMAIL=1 to run."
);
@@ -54,8 +55,7 @@ pub fn preflight_checks() {
if *USE_HCAPTCHA == false {
#[cfg(not(debug_assertions))]
{
- if !env::var("REVOLT_UNSAFE_NO_CAPTCHA")
- .map_or(false, |v| v == *"1") {
+ if !env::var("REVOLT_UNSAFE_NO_CAPTCHA").map_or(false, |v| v == *"1") {
panic!("Not letting you run this in production, set REVOLT_UNSAFE_NO_CAPTCHA=1 to run.");
}
}