Bundling

not optimal but its tomorrow

auth/src/server.rs

@@ -460,45 +460,13 @@ impl RealmAuth for RealmAuthServer {
 		}
 	}
 
-	async fn delete_account_flow(self, _: Context, username: String, token: String) -> Result<(), ErrorCode> {
+	async fn delete_account(self, _: Context, username: String, token: String) -> Result<(), ErrorCode> {
 		info!("API Request: delete_account_flow( username -> {}, token -> {} )", username, token);
 
 		if !self.is_authorized(&username, &token).await? {
 			return Err(Unauthorized);
 		}
 
-		let email = match query!("SELECT email FROM user WHERE username = ?;", username).fetch_one(&self.db_pool).await {
-			Ok(row) => Ok(row.email),
-			Err(_) => Err(InvalidUsername),
-		}?;
-
-		let code = self.gen_login_code();
-
-		let result = query!("UPDATE user SET login_code = ? WHERE username = ?;", code, username)
-			.execute(&self.db_pool).await;
-
-		match result {
-			Ok(_) => {
-				self.send_login_message(&username, &email, code);
-				Ok(())
-			}
-			Err(_) => Err(InvalidUsername)
-		}
-	}
-
-	async fn finish_delete_account_flow(self, _: Context, username: String, token: String, login_code: u32) -> Result<(), ErrorCode> {
-		info!("API Request: finish_delete_account_flow( username -> {}, token -> {}, login_code -> {} )", username, token, login_code);
-
-		if !self.is_authorized(&username, &token).await? {
-			return Err(Unauthorized);
-		}
-
-		if !self.is_login_code_valid(&username, login_code).await? {
-			return Err(InvalidLoginCode);
-		}
-
-		self.reset_login_code(&username).await?;
-
 		let result = query!("DELETE FROM user WHERE username = ?", username).execute(&self.db_pool).await;
 		match result {
 			Ok(_) => Ok(()),

auth/src/types.rs

@@ -16,8 +16,7 @@ pub trait RealmAuth {
     async fn change_avatar(username: String, token: String, new_avatar: String) -> Result<(), ErrorCode>;
     async fn get_all_data(username: String, token: String) -> Result<AuthUser, ErrorCode>;
     async fn sign_out(username: String, token: String) -> Result<(), ErrorCode>;
-    async fn delete_account_flow(username: String, token: String) -> Result<(), ErrorCode>;
+    async fn delete_account(username: String, token: String) -> Result<(), ErrorCode>;
-    async fn finish_delete_account_flow(username: String, token: String, login_code: u32) -> Result<(), ErrorCode>;
     async fn add_server(username: String, token: String, domain: String, port: u16) -> Result<(), ErrorCode>;
     async fn remove_server(username: String, token: String, domain: String, port: u16) -> Result<(), ErrorCode>;
     async fn get_joined_servers(username: String, token: String) -> Result<Vec<String>, ErrorCode>;

client/Cargo.toml

@@ -3,6 +3,14 @@ name = "realm_client"
 version = "0.1.0"
 edition = "2021"
 
+[target.x86_64-pc-windows-gnu]
+rustflags = ["-C", "link-args=-mwindows "]
+
+[package.metadata.bundle]
+name = "Realm Chat"
+identifier = "com.abunchofknowitalls.realm"
+icon = ["32x32.png", "128x128.png"]
+
 [dependencies]
 realm_auth = { path = "../auth" }
 realm_server = { path = "../server" }
@@ -20,3 +28,4 @@ tracing = "0.1.40"
 tracing-subscriber = "0.3.18"
 regex = "1.10.6"
 native-dialog = "0.7.0"
+chrono = "0.4.38"

client/src/app.rs

@@ -1,10 +1,14 @@
+use std::time::Duration;
 use tarpc::context;
 use tarpc::tokio_serde::formats::Json;
 use tokio::sync::broadcast;
 use tokio::sync::broadcast::{Receiver, Sender};
+use tokio::task::JoinHandle;
+use tokio::time::sleep;
 use tracing::log::*;
 use realm_auth::types::RealmAuthClient;
-use realm_server::types::{RealmChatClient, Room, User};
+use realm_server::events::Event;
+use realm_server::types::{RealmChatClient, Room};
 use realm_shared::stoken;
 use realm_shared::types::ErrorCode::*;
 use realm_shared::types::ErrorCode;
@@ -31,7 +35,7 @@ pub struct RealmApp {
 	pub active_servers: Option<Vec<CServer>>,
 
 	#[serde(skip)]
-	pub value: f32,
+	pub text_message_input: String,
 	#[serde(skip)]
 	pub login_window_open: bool,
 	#[serde(skip)]
@@ -95,7 +99,12 @@ pub struct RealmApp {
 	#[serde(skip)]
 	pub delete_room_channel: (Sender<Result<CServer, ErrorCode>>, Receiver<Result<CServer, ErrorCode>>),
 	#[serde(skip)]
-	pub room_changes_channel: (Sender<Result<(CServer, Vec<Room>), ErrorCode>>, Receiver<Result<(CServer, Vec<Room>), ErrorCode>>)
+	pub room_changes_channel: (Sender<Result<(CServer, Vec<Room>), ErrorCode>>, Receiver<Result<(CServer, Vec<Room>), ErrorCode>>),
+
+	#[serde(skip)]
+	pub event_channel: (Sender<(String, (i64, Event))>, Receiver<(String, (i64, Event))>),
+	#[serde(skip)]
+	pub polling_threads: Vec<(String, JoinHandle<()>)>,
 }
 
 impl Default for RealmApp {
@@ -108,22 +117,22 @@ impl Default for RealmApp {
 			saved_token: None,
 			saved_auth_address: None,
 			active_servers: None,
-			value: 2.7,
+			text_message_input: String::new(),
 
 			login_window_open: false,
 			login_window_username: String::new(),
 			login_window_code: String::new(),
-			login_window_server_domain: String::new(),
+			login_window_server_domain: "auth.realm.abunchofknowitalls.com".to_string(),
 			login_window_server_port: "5052".to_string(),
-			login_start_channel: broadcast::channel(10),
+			login_start_channel: broadcast::channel(256),
-			login_ending_channel: broadcast::channel(10),
+			login_ending_channel: broadcast::channel(256),
 			login_ready_for_code_input: false,
 			login_window_email: String::new(),
 
 			signup_window_open: false,
 
 			server_window_open: false,
-			server_window_domain: String::new(),
+			server_window_domain: "realm.abunchofknowitalls.com".to_string(),
 			server_window_port: "5051".to_string(),
 			
 			room_window_open: false,
@@ -133,15 +142,17 @@ impl Default for RealmApp {
 
 			info_window_open: false,
 
-			fetching_user_data_channel: broadcast::channel(10),
+			fetching_user_data_channel: broadcast::channel(256),
-			add_server_channel: broadcast::channel(10),
+			add_server_channel: broadcast::channel(256),
-			remove_server_channel: broadcast::channel(10),
+			remove_server_channel: broadcast::channel(256),
-			join_server_channel: broadcast::channel(10),
+			join_server_channel: broadcast::channel(256),
-			leave_server_channel: broadcast::channel(10),
+			leave_server_channel: broadcast::channel(256),
-			fetching_servers_channel: broadcast::channel(10),
+			fetching_servers_channel: broadcast::channel(256),
-			add_room_channel: broadcast::channel(10),
+			add_room_channel: broadcast::channel(256),
-			delete_room_channel: broadcast::channel(10),
+			delete_room_channel: broadcast::channel(256),
-			room_changes_channel: broadcast::channel(10),
+			room_changes_channel: broadcast::channel(256),
+			event_channel: broadcast::channel(256),
+			polling_threads: Vec::new(),
 		}
 	}
 }
@@ -154,9 +165,12 @@ impl RealmApp {
 
 		// Load previous app state (if any).
 		// Note that you must enable the `persistence` feature for this to work.
+		if !cfg!(debug_assertions) {
 			if let Some(storage) = cc.storage {
 				return eframe::get_value(storage, eframe::APP_KEY).unwrap_or_default();
 			}
+		}
+
 
 		Default::default()
 	}
@@ -237,6 +251,8 @@ pub fn fetch_server_data(channel: Sender<Result<CServer, ErrorCode>>, addresses:
 				port,
 				is_admin,
 				is_owner,
+				last_event_index: 0,
+				messages: Vec::new(),
 				rooms,
 			})).unwrap();
 		});
@@ -535,6 +551,87 @@ impl eframe::App for RealmApp {
 			}
 		}
 
+		// Polling events
+		while let Ok((serverid, (index, event))) = self.event_channel.1.try_recv() {
+			if let Some(active_servers) = &mut self.active_servers {
+				for server in active_servers {
+					if server.server_id.eq(&serverid) {
+						match event.clone() {
+							Event::NewMessage(message) => {
+								server.messages.push(message);
+							}
+							Event::NewRoom(room) => {
+								server.rooms.push(room);
+							}
+							Event::DeleteRoom(roomid) => {
+								server.rooms.retain(|r| !r.roomid.eq(&roomid));
+								if self.selected_roomid.eq(&roomid) {
+									self.selected_roomid.clear();
+								}
+							},
+							_ => {  }
+						}
+						if index > server.last_event_index {
+							server.last_event_index = index;
+						}
+					}
+				}
+			}
+		}
+
+		// Manage polling threads
+		if let Some(active_servers) = &mut self.active_servers {
+			if self.polling_threads.len() != active_servers.len() {
+				let running_thread_serverids = self.polling_threads.iter().map(|t| t.0.clone()).collect::<Vec<String>>();
+				let missing_servers = active_servers.clone().into_iter().filter(|s| !running_thread_serverids.contains(&s.server_id)).collect::<Vec<CServer>>();
+				for server in missing_servers {
+					let send_channel = self.event_channel.0.clone();
+					let serverid = server.server_id.clone();
+					let token = self.current_user.as_ref().unwrap().token.clone();
+					let userid = self.current_user.as_ref().unwrap().username.clone();
+					let handle = tokio::spawn(async move {
+						let mut transport = tarpc::serde_transport::tcp::connect(format!("{}:{}", server.domain, server.port), Json::default);
+						transport.config_mut().max_frame_length(usize::MAX);
+						let result = transport.await;
+						let connection = match result {
+							Ok(connection) => connection,
+							Err(_) => {
+								return;
+							}
+						};
+						let client = RealmChatClient::new(tarpc::client::Config::default(), connection).spawn();
+						let mut last_event_index = 0;
+						
+						loop {
+							let result = client.get_messages_since(
+								context::current(),
+								stoken(&token, &serverid, &server.domain, server.port),
+								userid.clone(),
+								last_event_index
+							).await;
+
+							match result {
+								Ok(messages) => {
+									if let Ok(messages) = messages {
+										if let Some(last) = messages.last() {
+											last_event_index = last.id;
+										}
+										for message in messages {
+											send_channel.send((serverid.clone(), (message.id, Event::NewMessage(message)))).unwrap();
+										}
+									}
+								}
+								Err(_) => break,
+							}
+
+							sleep(Duration::from_millis(1000)).await;
+						}
+					});
+					self.polling_threads.push((server.server_id.clone(), handle));
+				}
+			}
+		}
+
 		// File -> Quit
 		gui::top_panel(self, ctx);
 
@@ -549,6 +646,8 @@ impl eframe::App for RealmApp {
 
 	/// Called by the frame work to save state before shutdown.
 	fn save(&mut self, storage: &mut dyn eframe::Storage) {
+		if !cfg!(debug_assertions) {
 			eframe::set_value(storage, eframe::APP_KEY, self);
 		}
 	}
+}

client/src/lib.rs

@@ -1,4 +1,3 @@
-pub mod client;
 pub mod types;
 pub mod app;
 pub mod ui;

client/src/types.rs

@@ -1,4 +1,4 @@
-use realm_server::types::{RealmChatClient, Room};
+use realm_server::types::{Message, RealmChatClient, Room};
 
 #[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
 pub struct CUser {
@@ -19,5 +19,7 @@ pub struct CServer {
 	pub port: u16,
 	pub is_admin: bool,
 	pub is_owner: bool,
-	pub rooms: Vec<Room>
+	pub rooms: Vec<Room>,
+	pub last_event_index: i64,
+	pub messages: Vec<Message>,
 }

client/src/ui/gui.rs

@@ -1,14 +1,13 @@
+use chrono::Utc;
 use egui::{Context, SelectableLabel};
 use tarpc::context;
 use tarpc::tokio_serde::formats::Json;
 use realm_auth::types::RealmAuthClient;
 use realm_shared::types::ErrorCode::RPCError;
 use regex::Regex;
-use tarpc::client::RpcError;
 use tracing::log::*;
-use realm_server::types::{RealmChatClient, Room};
+use realm_server::types::{Message, MessageData, Room, User};
 use realm_shared::stoken;
-use realm_shared::types::ErrorCode;
 use crate::app::RealmApp;
 use crate::types::CServer;
 
@@ -60,17 +59,55 @@ pub fn top_panel(app: &mut RealmApp, ctx: &Context) {
 				app.selected_serverid.clear();
 			}
 
-			if ui.button("Info").clicked() {
-				app.info_window_open = true;
-			}
-
 			if ui.button("Quit").clicked() {
 				ctx.send_viewport_cmd(egui::ViewportCommand::Close);
 			}
 
-			ui.add_space(16.0);
 			
+			ui.with_layout(egui::Layout::right_to_left(egui::Align::TOP), |ui| {
 				egui::widgets::global_theme_preference_buttons(ui);
+
+				if ui.button("ℹ").clicked() {
+					app.info_window_open = true;
+				}
+				
+				if app.current_user.is_some() && ui.button("Delete Account").clicked() {
+					let address = app.current_user.clone().unwrap().auth_address;
+					let username = app.current_user.clone().unwrap().username;
+					let token = app.current_user.clone().unwrap().token;
+					let _handle = tokio::spawn(async move {
+						let mut transport = tarpc::serde_transport::tcp::connect(address, Json::default);
+						transport.config_mut().max_frame_length(usize::MAX);
+
+						let result = transport.await;
+						let connection = match result {
+							Ok(connection) => connection,
+							Err(e) => {
+								tracing::error!("Failed to connect to server: {}", e);
+								return;
+							}
+						};
+
+						let client = RealmAuthClient::new(tarpc::client::Config::default(), connection).spawn();
+						let result = client.delete_account(context::current(), username, token).await;
+
+						match result {
+							Ok(_) => info!("Account deleted successfully!"),
+							Err(e) => error!("Error deleting account: {:?}", e),
+						}
+					});
+					
+					app.current_user = None;
+					app.saved_username = None;
+					app.saved_token = None;
+					app.saved_auth_address = None;
+
+					app.active_servers = None;
+					app.selected_roomid.clear();
+					app.selected_serverid.clear();
+				}
+			});
+			
 		});
 	});
 }
@@ -185,7 +222,72 @@ pub fn rooms(app: &mut RealmApp, ctx: &Context) {
 
 pub fn messages(app: &mut RealmApp, ctx: &Context) {
 	egui::CentralPanel::default().show(ctx, |ui| {
+		ui.with_layout(egui::Layout::bottom_up(egui::Align::TOP), |ui| {
+			ui.with_layout(egui::Layout::right_to_left(egui::Align::Max), |ui| {
+				if ui.button("✉").on_hover_text("Send a message").clicked() {
+					if let Some(active_servers) = &app.active_servers {
+						for server in active_servers.clone() {
+							if server.server_id.eq(&app.selected_serverid) {
+								let username = app.current_user.as_ref().unwrap().username.clone();
+								let token = app.current_user.as_ref().unwrap().token.clone();
+								let room = server.rooms.iter().find(|r| r.roomid.eq(&app.selected_roomid)).unwrap().clone();
+								let text_message = app.text_message_input.clone();
+								let _handle = tokio::spawn(async move {
+									let result = server.tarpc_conn.send_message(
+										context::current(),
+										stoken(&token, &server.server_id, &server.domain, server.port),
+										Message {
+											id: 0,
+											timestamp: Utc::now(),
+											user: server.tarpc_conn.get_user(context::current(), username.clone()).await.unwrap().unwrap(),
+											room,
+											data: MessageData::Text(text_message),
+										}
+									).await;
+								});
 								
+								app.text_message_input.clear();
+							}
+						}
+					}
+				}
+				
+				ui.add(
+					egui::TextEdit::multiline(&mut app.text_message_input)
+						.desired_rows(1)
+						.desired_width(ui.available_width())
+						.hint_text("Send a message...")
+				);
+			});
+			ui.with_layout(egui::Layout::top_down_justified(egui::Align::Min), |ui| {
+				egui::ScrollArea::vertical().show(ui, |ui| {
+					if let Some(active_servers) = &app.active_servers {
+						for server in active_servers {
+							let messages_to_display = server.messages
+								.iter()
+								.filter(|m| m.room.roomid.eq(&app.selected_roomid))
+								.collect::<Vec<&Message>>();
+							
+							for message in messages_to_display {
+								match message.clone().data {
+									MessageData::Text(text) => {
+										ui.label(format!("{} - {}: {}",
+														 message.timestamp.format("%Y-%m-%d %H:%M:%S"),
+														 message.user.userid.split(':').collect::<Vec<&str>>()[0], 
+														 text));
+									}
+									MessageData::Attachment(_) => {}
+									MessageData::Reply(_) => {}
+									MessageData::Edit(_) => {}
+									MessageData::Reaction(_) => {}
+									MessageData::Redaction(_) => {}
+								}
+							}
+						}
+					}
+				});
+			});
+		});
 	});
 }
 

server/src/events.rs

@@ -1,64 +1,15 @@
-// use durian::bincode_packet;
+use crate::types::{Message, Room, User};
-// use crate::types::{Message, Room, User};
+
-// 
+#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
-// #[bincode_packet]
+pub enum Event {
-// pub struct Greet {
+	// UserJoined(User),
-// 	pub id: u32
+	// UserLeft(User),
-// }
+	None,
-// 
+	NewMessage(Message),
-// #[bincode_packet]
+	NewRoom(Room),
-// pub struct UserJoinedEvent {
+	DeleteRoom(String),
-// 	pub user: User,
+	// KickedUser(KickedUser),
-// }
+	// BannedUser(BannedUser),
-// 
+	// PromotedUser(PromotedUser),
-// #[bincode_packet]
+	// DemotedUser(DemotedUser),
-// pub struct UserLeftEvent {
+}
-// 	pub user: User,
-// }
-// 
-// #[bincode_packet]
-// pub struct NewMessageEvent {
-// 	pub message: Message,
-// }
-// 
-// // #[bincode_packet]
-// // pub struct StartTypingEvent {
-// // 	pub user: User,
-// // 	pub room: String,
-// // }
-// // 
-// // #[bincode_packet]
-// // pub struct StopTypingEvent {
-// // 	pub user: User,
-// // 	pub room: String,
-// // }
-// 
-// #[bincode_packet]
-// pub struct NewRoomEvent {
-// 	pub room: Room,
-// }
-// 
-// #[bincode_packet]
-// pub struct DeleteRoomEvent {
-// 	pub roomid: String,
-// }
-// 
-// #[bincode_packet]
-// pub struct KickedUserEvent {
-// 	pub userid: String, 
-// }
-// 
-// #[bincode_packet]
-// pub struct BannedUserEvent {
-// 	pub userid: String,
-// }
-// 
-// #[bincode_packet]
-// pub struct PromotedUserEvent {
-// 	pub userid: String,
-// }
-// 
-// #[bincode_packet]
-// pub struct DemotedUserEvent {
-// 	pub userid: String,
-// }

server/src/main.rs

@@ -15,6 +15,7 @@ use tarpc::server::incoming::Incoming;
 use tarpc::server::BaseChannel;
 use tokio::sync::Mutex;
 use tracing::{info, subscriber, warn};
+use tracing::instrument::WithSubscriber;
 use realm_server::events::*;
 use realm_server::server::RealmChatServer;
 use realm_server::types::{RealmChat};
@@ -93,7 +94,7 @@ async fn main() -> anyhow::Result<()> {
 		.filter_map(|r| future::ready(r.ok()))
 		.map(BaseChannel::with_defaults)
 		// Limit channels to 1 per IP.
-		.max_channels_per_key(1, |t| t.transport().peer_addr().unwrap().ip())
+		.max_channels_per_key(1024, |t| t.transport().peer_addr().unwrap().ip())
 		// serve is generated by the service attribute. It takes as input any type implementing
 		// the generated World trait.
 		.map(|channel| {
@@ -101,7 +102,7 @@ async fn main() -> anyhow::Result<()> {
 			channel.execute(server.serve()).for_each(spawn)
 		})
 		// Max 10 channels.
-		.buffer_unordered(10)
+		.buffer_unordered(10240)
 		.for_each(|_| async {})
 		.await;
 

server/src/server.rs

@@ -25,12 +25,13 @@ pub struct RealmChatServer {
 	pub db_pool: Pool<Sqlite>,
 	pub typing_users: Vec<(String, String)>, //NOTE: user.userid, room.roomid
 	pub cache: Cache<String, String>,
+	//pub events: Arc<Mutex<Vec<(u32, Event)>>>,
 }
 
 const FETCH_MESSAGE: &str = "SELECT message.*,
         room.id AS 'room_id', room.roomid AS 'room_roomid', room.admin_only_send AS 'room_admin_only_send', room.admin_only_view AS 'room_admin_only_view',
         user.id AS 'user_id', user.userid AS 'user_userid', user.name AS 'user_name', user.owner AS 'user_owner', user.admin AS 'user_admin'
-	    FROM message INNER JOIN room ON message.room = room.id INNER JOIN user ON message.user = user.id WHERE room.admin_only_view = ? OR false";
+	    FROM message INNER JOIN room ON message.room = room.id INNER JOIN user ON message.user = user.id";
 
 impl RealmChatServer {
 	pub fn new(server_id: String, socket: SocketAddr, db_pool: Pool<Sqlite>) -> RealmChatServer {
@@ -46,6 +47,7 @@ impl RealmChatServer {
 				.time_to_idle(Duration::from_secs(5*60))
 				.time_to_live(Duration::from_secs(60*60))
 				.build(),
+			//events: Arc::new(Mutex::new(Vec::new())),
 		}
 	}
 	
@@ -217,6 +219,21 @@ impl RealmChat for RealmChatServer {
 		self.internal_is_user_owner(&userid).await
 	}
 
+	async fn poll_events_since(self, _: Context, index: u32) -> Vec<(u32, Event)> {
+		//self.events.lock().await.iter().filter(|(i, _)| i > &index).map(|(i, e)| (*i, e.clone())).collect()
+		
+		// let mut events_to_send = Vec::new();
+		// 
+		// for (i, event) in self.events.lock().await.clone() {
+		// 	if i > index {
+		// 		events_to_send.push((i, event));
+		// 	}
+		// }
+		// 
+		// events_to_send
+		Vec::new()
+	}
+
 	async fn join_server(self, _: Context, stoken: String, userid: String) -> Result<User, ErrorCode> {
 		if !self.is_stoken_valid(&userid, &stoken).await {
 			return Err(Unauthorized)
@@ -282,42 +299,42 @@ impl RealmChat for RealmChatServer {
 	}
 
 	async fn send_message(self, _: Context, stoken: String, mut message: Message) -> Result<Message, ErrorCode> {
-		if !self.is_stoken_valid(&message.user.userid, &stoken).await { // Check sender userid
+		// if !self.is_stoken_valid(&message.user.userid, &stoken).await { // Check sender userid
-			return Err(Unauthorized)
+		// 	return Err(Unauthorized)
-		}
+		// }
+		// 
+		// // Assert all the data in message is correct
+		// message.user = self.inner_get_user(&message.user.userid).await?;
 
-		// Assert all the data in message is correct
+		// match &message.data { // Check that the sender is the owner of the referencing msg
-		message.user = self.inner_get_user(&message.user.userid).await?;
+		// 	MessageData::Edit(e) => {
+		// 		let ref_msg = self.inner_get_message(&message.user.userid, e.referencing_id).await?;
+		// 		if !ref_msg.user.userid.eq(&message.user.userid) {
+		// 			return Err(Unauthorized)
+		// 		}
+		// 	}
+		// 	MessageData::Redaction(r)=> {
+		// 		let ref_msg = self.inner_get_message(&message.user.userid, r.referencing_id).await?;
+		// 		if !ref_msg.user.userid.eq(&message.user.userid) || !self.internal_is_user_admin(&message.user.userid).await {
+		// 			return Err(Unauthorized)
+		// 		}
+		// 	}
+		// 	_ => {}
+		// }
 
-		match &message.data { // Check that the sender is the owner of the referencing msg
+		// let is_admin = self.internal_is_user_admin(&message.user.userid).await;
-			MessageData::Edit(e) => {
+		// let admin_only_send = query!(
-				let ref_msg = self.inner_get_message(&message.user.userid, e.referencing_id).await?;
+		// 	"SELECT admin_only_send FROM room WHERE roomid = ?",
-				if !ref_msg.user.userid.eq(&message.user.userid) {
+		// 	message.room.roomid).fetch_one(&self.db_pool).await;
-					return Err(Unauthorized)
+		// if let Ok(record) = admin_only_send {
-				}
+		// 	if record.admin_only_send && !is_admin {
-			}
+		// 		return Err(Unauthorized)
-			MessageData::Redaction(r)=> {
+		// 	}
-				let ref_msg = self.inner_get_message(&message.user.userid, r.referencing_id).await?;
+		// } else {
-				if !ref_msg.user.userid.eq(&message.user.userid) || !self.internal_is_user_admin(&message.user.userid).await {
+		// 	return Err(RoomNotFound)
-					return Err(Unauthorized)
+		// }
-				}
+		// 
-			}
+		// message.room = self.inner_get_room(&message.user.userid, &message.room.roomid).await?;
-			_ => {}
-		}
-
-		let is_admin = self.internal_is_user_admin(&message.user.userid).await;
-		let admin_only_send = query!(
-			"SELECT admin_only_send FROM room WHERE roomid = ?",
-			message.room.roomid).fetch_one(&self.db_pool).await;
-		if let Ok(record) = admin_only_send {
-			if record.admin_only_send && !is_admin {
-				return Err(Unauthorized)
-			}
-		} else {
-			return Err(RoomNotFound)
-		}
-
-		message.room = self.inner_get_room(&message.user.userid, &message.room.roomid).await?;
 
 		let result = match &message.data {
 			MessageData::Text(text) => {
@@ -357,6 +374,9 @@ impl RealmChat for RealmChatServer {
 				// if result.is_err() {
 				// 	error!("Error broadcasting NewMessageEvent!");
 				// }
+				// self.events.lock().await.push(
+				// 	(self.events.lock().await.len() as u32 + 1,
+				// 	 Event::NewMessage(message.clone())));
 
 				Ok(message)
 			},
@@ -397,15 +417,15 @@ impl RealmChat for RealmChatServer {
 		}
 	}
 
-	async fn get_messages_since(self, _: Context, stoken: String, userid: String, time: DateTime<Utc>) -> Result<Vec<Message>, ErrorCode> {
+	async fn get_messages_since(self, _: Context, stoken: String, userid: String, id: i64) -> Result<Vec<Message>, ErrorCode> {
 		if !self.is_stoken_valid(&userid, &stoken).await {
 			return Err(Unauthorized)
 		}
 		
 		let is_admin = self.internal_is_user_admin(&userid).await;
-		let result = sqlx::query(&format!("{}{}", FETCH_MESSAGE, "AND message.timestamp >= ?"))
+		let result = sqlx::query(&format!("{}{}", FETCH_MESSAGE, " AND message.id > ?"))
-			.bind(is_admin)
+			//.bind(is_admin)
-			.bind(time)
+			.bind(id)
 			.fetch_all(&self.db_pool).await;
 
 		match result {
@@ -484,6 +504,10 @@ impl RealmChat for RealmChatServer {
 				// 	error!("Error broadcasting NewRoomEvent!");
 				// }
 
+				// self.events.lock().await.push(
+				// 	(self.events.lock().await.len() as u32 + 1,
+				// 	 Event::NewRoom(room.clone())));
+				
 				Ok(room)
 			}
 			Err(_) => Err(MalformedDBResponse)
@@ -511,6 +535,10 @@ impl RealmChat for RealmChatServer {
 				// 	error!("Error broadcasting DeleteRoomEvent!");
 				// }
 
+				// self.events.lock().await.push(
+				// 	(self.events.lock().await.len() as u32 + 1,
+				// 	 Event::DeleteRoom(roomid.clone())));
+				
 				Ok(())
 			}
 			Err(_) => Err(MalformedDBResponse)

server/src/types.rs

@@ -4,7 +4,7 @@ use sqlx::sqlite::SqliteRow;
 use tarpc::serde::{Deserialize, Serialize};
 
 use realm_shared::types::ErrorCode;
-
+use crate::events::Event;
 use crate::types::MessageData::*;
 
 #[tarpc::service]
@@ -14,7 +14,7 @@ pub trait RealmChat {
 	async fn get_info() -> ServerInfo;
 	async fn is_user_admin(stoken: String) -> bool;
 	async fn is_user_owner(stoken: String) -> bool;
-	
+	async fn poll_events_since(index: u32) -> Vec<(u32, Event)>;
 	async fn join_server(stoken: String, userid: String) -> Result<User, ErrorCode>;
 	async fn leave_server(stoken: String, userid: String) -> Result<(), ErrorCode>;
 
@@ -26,7 +26,7 @@ pub trait RealmChat {
 
 	//NOTE: Any user can call, if they are in the server
 	async fn get_message(stoken: String, userid: String, id: i64) -> Result<Message, ErrorCode>;
-	async fn get_messages_since(stoken: String, userid: String, time: DateTime<Utc>) -> Result<Vec<Message>, ErrorCode>;
+	async fn get_messages_since(stoken: String, userid: String, id: i64) -> Result<Vec<Message>, ErrorCode>;
 	async fn get_all_direct_replies(stoken: String, userid: String, head: i64) -> Result<Vec<Message>, ErrorCode>;
 	async fn get_reply_chain(stoken: String, userid: String, head: Message, depth: u8) -> Result<ReplyChain, ErrorCode>;
 	async fn get_rooms(stoken: String, userid: String) -> Result<Vec<Room>, ErrorCode>;