diff --git a/packages/core/Cargo.lock b/packages/core/Cargo.lock index 10902ec..3c933b1 100644 --- a/packages/core/Cargo.lock +++ b/packages/core/Cargo.lock @@ -11,7 +11,7 @@ dependencies = [ "macroific", "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -130,8 +130,8 @@ dependencies = [ "gloo-net 0.7.0", "gloo-timers 0.4.0", "irc-proto", - "ordermap", "rand", + "serde", "thiserror 2.0.18", "time", "tokio", @@ -184,7 +184,7 @@ dependencies = [ "macroific", "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -211,16 +211,10 @@ dependencies = [ "proc-macro2", "quote", "rustc_version", - "syn", + "syn 2.0.118", "unicode-xid", ] -[[package]] -name = "equivalent" -version = "1.0.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f" - [[package]] name = "fancy_constructor" version = "2.1.0" @@ -230,7 +224,7 @@ dependencies = [ "macroific", "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -310,7 +304,7 @@ checksum = "e835b70203e41293343137df5c0664546da5745f82ec9b84d40be8336958447b" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -580,12 +574,6 @@ dependencies = [ "web-sys", ] -[[package]] -name = "hashbrown" -version = "0.17.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" - [[package]] name = "http" version = "0.2.12" @@ -615,7 +603,7 @@ checksum = "0ab604ee7085efba6efc65e4ebca0e9533e3aff6cb501d7d77b211e3a781c6d5" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -652,17 +640,7 @@ dependencies = [ "macroific", "proc-macro2", "quote", - "syn", -] - -[[package]] -name = "indexmap" -version = "2.14.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d466e9454f08e4a911e14806c24e16fba1b4c121d1ea474396f396069cf949d9" -dependencies = [ - "equivalent", - "hashbrown", + "syn 2.0.118", ] [[package]] @@ -732,7 +710,7 @@ dependencies = [ "proc-macro2", "quote", "sealed", - "syn", + "syn 2.0.118", ] [[package]] @@ -744,7 +722,7 @@ dependencies = [ "proc-macro2", "quote", "sealed", - "syn", + "syn 2.0.118", ] [[package]] @@ -757,7 +735,7 @@ dependencies = [ "macroific_core", "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -787,15 +765,6 @@ version = "1.21.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" -[[package]] -name = "ordermap" -version = "1.2.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7f7476a5b122ff1fce7208e7ee9dccd0a516e835f5b8b19b8f3c98a34cf757c1" -dependencies = [ - "indexmap", -] - [[package]] name = "percent-encoding" version = "2.3.2" @@ -819,7 +788,7 @@ checksum = "c96395f0a926bc13b1c17622aaddda1ecb55d49c8f1bf9777e4d877800a43f8b" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -896,7 +865,7 @@ checksum = "22f968c5ea23d555e670b449c1c5e7b2fc399fdaec1d304a17cd48e288abc107" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -907,9 +876,9 @@ checksum = "8a7852d02fc848982e0c167ef163aaff9cd91dc640ba85e263cb1ce46fae51cd" [[package]] name = "serde" -version = "1.0.228" +version = "1.0.229" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9a8e94ea7f378bd32cbbd37198a4a91436180c5bb472411e48b5ec2e2124ae9e" +checksum = "4148590afebada386688f18773da617792bf2ef03ffc1e4cbd2b1d45b023e0ba" dependencies = [ "serde_core", "serde_derive", @@ -939,22 +908,22 @@ dependencies = [ [[package]] name = "serde_core" -version = "1.0.228" +version = "1.0.229" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "41d385c7d4ca58e59fc732af25c3983b67ac852c1a25000afe1175de458b67ad" +checksum = "67dca2c9c51e58a4791a4b1ed58308b39c64224d349a935ab5039aa360942a48" dependencies = [ "serde_derive", ] [[package]] name = "serde_derive" -version = "1.0.228" +version = "1.0.229" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d540f220d3187173da220f885ab66608367b6574e925011a9353e4badda91d79" +checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 3.0.3", ] [[package]] @@ -965,7 +934,7 @@ checksum = "18d26a20a969b9e3fdf2fc2d9f21eda6c40e2de84c9408bb5d3b05d499aae711" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -1034,6 +1003,17 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "syn" +version = "3.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53e9bae58849f64dfa4f5d5ae372c8341f7305f82a3868709269343628b659a3" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + [[package]] name = "thiserror" version = "1.0.69" @@ -1060,7 +1040,7 @@ checksum = "4fee6c4efc90059e10f81e6d42c60a18f76588c3d74cb83a0b242a2b6c7504c1" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -1071,7 +1051,7 @@ checksum = "ebc4ee7f67670e9b64d05fa4253e753e016c6c95ff35b89b7941d6b856dec1d5" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -1131,7 +1111,7 @@ checksum = "385a6cb71ab9ab790c5fe8d67f1645e6c450a7ce006a33de03daa956cf70a496" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -1166,7 +1146,7 @@ checksum = "7490cfa5ec963746568740651ac6781f701c9c5ea257c58e057f3ba8cf69e8da" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -1249,7 +1229,7 @@ dependencies = [ "proc-macro2", "quote", "serde_derive_internals", - "syn", + "syn 2.0.118", ] [[package]] @@ -1312,7 +1292,7 @@ dependencies = [ "bumpalo", "proc-macro2", "quote", - "syn", + "syn 2.0.118", "wasm-bindgen-shared", ] diff --git a/packages/core/core-shared/Cargo.toml b/packages/core/core-shared/Cargo.toml index bebd5ed..a61c28e 100644 --- a/packages/core/core-shared/Cargo.toml +++ b/packages/core/core-shared/Cargo.toml @@ -10,7 +10,6 @@ base64 = "0.22.1" blake3 = "1.8.5" futures = "0.3.32" irc-proto = "1.1.0" -ordermap = "1.2.0" thiserror = "2.0.18" time = { version = "0.3.53", features = ["parsing"] } tracing = "0.1.44" @@ -23,6 +22,7 @@ gloo-net = { version = "0.7.0", optional = true } tokio = { version = "1.52.3", features = ["macros", "rt", "time", "sync"], optional = true } gloo-timers = { version = "0.4.0", features = ["futures"], optional = true } rand = { version = "0.10.2", default-features = false } +serde = { version = "1.0.229", features = ["derive"] } [features] web = ["dep:gloo-net", "dep:wasm-bindgen", "dep:gloo-console", "dep:tsify", "dep:web-time", "dep:gloo-timers"] diff --git a/packages/core/core-shared/src/actor.rs b/packages/core/core-shared/src/actor.rs index 0f86893..9a85fba 100644 --- a/packages/core/core-shared/src/actor.rs +++ b/packages/core/core-shared/src/actor.rs @@ -1,7 +1,7 @@ use std::{fmt, time::Duration}; use futures::{FutureExt, future::FusedFuture}; -use rand::{SeedableRng, rngs::SmallRng, seq::IndexedRandom}; +use rand::{SeedableRng, rngs::SmallRng}; #[cfg(not(feature = "web"))] use std::time::Instant; #[cfg(feature = "web")] @@ -11,13 +11,11 @@ use web_time::Instant; use crate::dbg; use crate::{ SendCommand, - state::{ - Channel, ChannelRole, ChannelUser, History, Message, MessageMetadata, MessageReference, - MessageType, OrbitError, Server, ServerEvent, SignedIn, Tags, TextMessage, User, - }, + database::Database, + response_channels::{CommandKey, CommandResponse, ResponseChannels}, + state::{Channel, Message, OrbitError, Server, ServerEvent, User}, }; -use anyhow::{Context, anyhow}; -use base64::prelude::*; +use anyhow::Context; use futures::{ SinkExt, StreamExt, channel::{ @@ -26,105 +24,8 @@ use futures::{ }, stream::FusedStream, }; -use irc_proto::{BatchSubCommand, CapSubCommand, Command::*, Message as IrcMessage, Response}; -use ordermap::OrderMap; -use tracing::{debug, error, warn}; - -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum CommandKey { - RequestCaps, - SignIn, - Join(String), - Privmsg { target: String, text: String }, - History, - Label(String), -} - -#[derive(Debug)] -pub enum CommandResponse { - GetState(Box), - GetChannelState(Box>), - Capabilities, - SignIn(Result), - Join(String), - Privmsg(Box), - History(History), -} - -const LABEL_CHARSET: &str = "abcdefghijklmnopqrstuvwxyz\ - ABCDEFGHIJKLMNOPQRSTUVWXYZ\ - 1234567890"; - -fn generate_label(rng: &mut SmallRng) -> String { - let char_vec = LABEL_CHARSET - .split("") - .filter(|c| !c.is_empty()) - .collect::>(); - - std::iter::repeat_with(|| char_vec.choose(rng).expect("CHARSET is not empty")) - .take(10) - .copied() - .collect::>() - .join("") -} - -#[derive(Debug)] -pub struct ResponseChannels { - channels: Vec<(CommandKey, Instant, oneshot::Sender)>, - rng: SmallRng, -} - -impl Default for ResponseChannels { - #[tracing::instrument] - fn default() -> Self { - Self { - channels: Vec::new(), - rng: SmallRng::from_seed([0; 32]), - } - } -} - -impl ResponseChannels { - #[tracing::instrument] - pub fn register(&mut self, key: CommandKey, os_tx: oneshot::Sender) { - self.channels.push((key, Instant::now(), os_tx)); - } - - #[tracing::instrument] - pub fn register_labeled(&mut self, os_tx: oneshot::Sender) -> String { - let label = generate_label(&mut self.rng); - - self.channels - .push((CommandKey::Label(label.clone()), Instant::now(), os_tx)); - - label - } - - #[tracing::instrument] - pub fn reply( - &mut self, - key: &CommandKey, - response: CommandResponse, - ) -> Result { - if let Some(idx) = self.channels.iter().position(|(rk, _, _)| rk == key) { - let (_, _, ch) = self.channels.remove(idx); - ch.send(response)?; - - Ok(true) - } else { - if let CommandKey::Label(label) = key { - warn!("Failed to find response channel for label {label:?}"); - } - // warn!("Failed to find response channel"); - Ok(false) - } - } - - pub fn check_timeouts(&mut self) { - self.channels - .retain(|(_, creation, _)| creation.elapsed() < Duration::from_secs(1)); - } -} +use irc_proto::Message as IrcMessage; +use tracing::warn; #[derive(Debug)] pub struct ActorMessage { @@ -178,19 +79,19 @@ pub trait IrcConnection: fmt::Debug { fn address(&self) -> &str; } -struct RequestedHistory { - channel: String, - label: Option, +pub(crate) struct RequestedHistory { + pub target: String, + pub label: Option, } #[derive(Debug)] -struct CurrentBatch { - id: String, - data: BatchData, +pub(crate) struct CurrentBatch { + pub id: String, + pub data: BatchData, } #[derive(Debug)] -enum BatchData { +pub(crate) enum BatchData { History { label: Option, channel: String, @@ -204,13 +105,13 @@ enum BatchData { } impl CurrentBatch { - fn is_chathistory(&self) -> bool { + pub fn is_chathistory(&self) -> bool { matches!(self.data, BatchData::History { .. }) } } #[derive(Default, Clone)] -enum SaslState { +pub(crate) enum SaslState { #[default] Unauthed, Requested { @@ -221,28 +122,30 @@ enum SaslState { }, } -pub struct IrcActor { - cmd_rx: mpsc::UnboundedReceiver, - incoming: C::Incoming, - outgoing: C::Outgoing, - state: Server, - response_channels: ResponseChannels, - event_handlers: Vec>, - error_handlers: Vec>, - disconnect_handlers: Vec>, - - current_batches: Vec, - requested_history_batches: Vec<(RequestedHistory, Instant)>, - sasl_state: SaslState, - rng: SmallRng, +pub struct IrcActor { + pub(crate) cmd_rx: mpsc::UnboundedReceiver, + pub(crate) incoming: C::Incoming, + pub(crate) outgoing: C::Outgoing, + pub(crate) state: Server, + pub(crate) database: DB, + pub(crate) response_channels: ResponseChannels, + pub(crate) event_handlers: Vec>, + pub(crate) error_handlers: Vec>, + pub(crate) disconnect_handlers: Vec>, + + pub(crate) current_batches: Vec, + pub(crate) requested_history_batches: Vec<(RequestedHistory, Instant)>, + pub(crate) sasl_state: SaslState, + pub(crate) rng: SmallRng, } -impl IrcActor { +impl IrcActor { #[tracing::instrument] pub async fn start( id: i32, connection: C, - spawn: fn(IrcActor) -> (), + database: DB, + spawn: fn(IrcActor) -> (), ) -> Result, OrbitError> { let address = connection.address().to_string(); let (incoming, outgoing) = connection.in_out(); @@ -253,6 +156,7 @@ impl IrcActor { incoming, outgoing, state: Server::new(id, address), + database, response_channels: ResponseChannels::default(), event_handlers: Default::default(), error_handlers: Default::default(), @@ -278,7 +182,7 @@ impl IrcActor { #[tracing::instrument(skip(self))] pub async fn run(mut self) { - fn create_timeout() -> impl FusedFuture { + fn create_timeout() -> impl FusedFuture { #[cfg(feature = "web")] let timeout = gloo_timers::future::TimeoutFuture::new(1000).fuse(); #[cfg(not(feature = "web"))] @@ -327,542 +231,7 @@ impl IrcActor { } } - #[tracing::instrument(err, skip(self))] - pub async fn handle_incoming(&mut self, mut message: IrcMessage) -> Result<(), OrbitError> { - match message.command { - CAP(_, sub, param, caps) => { - self.handle_caps(sub, param, caps).await?; - } - PING(server1, server2) => { - self.outgoing - .pong(server1, server2) - .await - .context("Failed to send pong")?; - } - Response(rpl, params) => { - self.handle_response(rpl, params).await?; - } - JOIN(ref channel_name, _, _) => { - let source = message.source_nickname().unwrap(); - - let mut tags = Tags::default(); - if let Some(ref t) = message.tags { - tags = Tags::parse(t); - } - - if self.current_batches.iter().any(|b| b.is_chathistory()) { - assert!(tags.server_time.is_some()); - } - - let state_message = Message { - text: None, - metadata: MessageMetadata { - msgid: tags.msgid_with_fallback(&["JOIN", source]), - server_time: tags.server_time_with_fallback() as f64, - message_type: MessageType::Join, - user: source.to_string(), - }, - }; - - if self - .push_batch(channel_name.clone(), state_message.clone()) - .await - { - return Ok(()); - } - - if source == self.state.me.as_ref().unwrap().nickname { - let channel = Channel::new(channel_name.clone()); - self.state - .channels - .insert(channel_name.clone(), channel.clone()); - - if !self.state.capabilities.history.enabled { - self.response_channels - .reply( - &CommandKey::Join(channel_name.clone()), - CommandResponse::Join(channel_name.clone()), - ) - .map_err(|e| anyhow!("Failed to reply to JOIN command {e:?}"))?; - } - - self.on_event(ServerEvent::Joined(channel)).await?; - - if self.state.capabilities.history.enabled { - let label = if self.state.capabilities.labeled_response.enabled { - Some(generate_label(&mut self.rng)) - } else { - None - }; - - self.requested_history_batches.push(( - RequestedHistory { - channel: channel_name.clone(), - label: label.clone(), - }, - Instant::now(), - )); - self.history_latest(channel_name.clone(), None, 5, label) - .await - .context("Failed to request latest history")?; - } - } else { - let channel = self.channel_mut(channel_name.clone()).await; - - channel.users.push(ChannelUser { - nickname: source.to_string(), - role: ChannelRole::Regular, - }); - } - - self.on_event(ServerEvent::Privmsg { - channel: channel_name.to_string(), - message: state_message, - }) - .await?; - } - PART(ref channel_name, ref comment) => { - let mut tags = Tags::default(); - if let Some(ref t) = message.tags { - tags = Tags::parse(t); - } - let source = message.source_nickname().unwrap(); - - let state_message = Message { - text: None, - metadata: MessageMetadata { - msgid: tags.msgid_with_fallback(&["PART", source]), - server_time: tags.server_time_with_fallback() as f64, - message_type: MessageType::Part, - user: source.to_string(), - }, - }; - - if self - .push_batch(channel_name.clone(), state_message.clone()) - .await - { - return Ok(()); - } - - self.on_event(ServerEvent::Privmsg { - channel: channel_name.to_string(), - message: state_message, - }) - .await?; - - let channel = self.channel_mut(channel_name.clone()).await; - - channel.users.retain(|u| u.nickname != source); - } - QUIT(ref comment) => { - let mut tags = Tags::default(); - if let Some(ref t) = message.tags { - tags = Tags::parse(t); - } - let source = message.source_nickname().unwrap(); - - let state_message = Message { - text: None, - metadata: MessageMetadata { - msgid: tags.msgid_with_fallback(&["QUIT", source]), - server_time: tags.server_time_with_fallback() as f64, - message_type: MessageType::Quit, - user: source.to_string(), - }, - }; - - for batch in &mut self.current_batches { - if let BatchData::History { messages, .. } = &mut batch.data { - messages.push(state_message); - return Ok(()); - } - } - - self.on_event(ServerEvent::Privmsg { - channel: String::new(), - message: state_message, - }) - .await?; - - self.state.users.remove(source); - for channel in self.state.channels.values_mut() { - channel.users.retain(|u| u.nickname != source); - } - } - PRIVMSG(ref target, ref text) => { - let mut tags = Tags::default(); - if let Some(ref t) = message.tags { - tags = Tags::parse(t); - } - assert_eq!( - self.current_batches.iter().last().map(|b| b.id.as_str()), - tags.batch.as_deref(), - ); - - if let Some(batch) = self.current_batches.iter_mut().last() - && let BatchData::Multiline { - message, - target: channel, - } = &mut batch.data - && let Some(t) = message.text.as_mut() - { - if t.content.is_empty() { - *channel = target.to_string(); - t.content = text.to_string(); - } else { - t.content = format!("{}\n{}", t.content, text); - } - return Ok(()); - } - - let source = message.source_nickname().unwrap(); - - let msgid = tags.msgid_with_fallback(&["PRIVMSG", source, target, text]); - - let reply = tags - .reply - .as_ref() - .map(|r| { - self.state - .channels - .get(target) - .and_then(|c| c.messages.get(r)) - }) - .map(|m| MessageReference { - text: m.and_then(|m| m.text.clone().map(|t| t.content)), - username: m.map(|m| m.metadata.user.clone()), - }); - - let state_message = Message { - metadata: MessageMetadata { - msgid: msgid.clone(), - server_time: tags.server_time_with_fallback() as f64, - message_type: MessageType::Privmsg, - user: source.to_string(), - }, - text: Some(TextMessage { - content: text.clone(), - reactions: OrderMap::new(), - reply, - redacted: false, - edited: false, - relayed_by: tags.relayed_by, - }), - }; - - if self.push_batch(target.clone(), state_message.clone()).await { - return Ok(()); - } - - if let Some(username) = tags.account.clone() { - let user = self.user_mut(source.to_string()).await; - user.username = Some(username); - } - - let channel = self.channel_mut(target.clone()).await; - channel.messages.insert(msgid, state_message.clone()); - - if source == self.state.me.as_ref().unwrap().nickname - && let Err(e) = self.response_channels.reply( - &CommandKey::Privmsg { - target: target.clone(), - text: text.clone(), - }, - CommandResponse::Privmsg(Box::new(state_message.clone())), - ) - { - error!("Failed to reply to PRIVMSG command {e:?}"); - } - - self.on_event(ServerEvent::Privmsg { - channel: target.clone(), - message: state_message, - }) - .await?; - } - BATCH(ref reference, ref typ, ref param) => { - let mut tags = Tags::default(); - if let Some(ref t) = message.tags { - tags = Tags::parse(t); - } - if let Some(id) = reference.strip_prefix('+') { - match typ { - Some(BatchSubCommand::CUSTOM(c)) if c.as_str() == "CHATHISTORY" => { - let idx = self - .requested_history_batches - .iter() - .position(|b| b.0.label == tags.label) - .expect("Chat history was requested"); - let channel = self.requested_history_batches.remove(idx).0.channel; - - self.current_batches.push(CurrentBatch { - id: id.to_string(), - data: BatchData::History { - label: tags.label, - channel, - messages: Vec::new(), - }, - }); - } - Some(BatchSubCommand::CUSTOM(c)) if c.as_str() == "DRAFT/MULTILINE" => { - let source = message.source_nickname().unwrap(); - let target = message.response_target().unwrap(); - let msgid = tags.msgid_with_fallback(&["MULTILINE", source, target]); - - let reply = tags - .reply - .as_ref() - .map(|r| { - self.state - .channels - .get(target) - .and_then(|c| c.messages.get(r)) - }) - .map(|m| MessageReference { - text: m.and_then(|m| m.text.clone().map(|t| t.content)), - username: m.map(|m| m.metadata.user.clone()), - }); - - self.current_batches.push(CurrentBatch { - id: id.to_string(), - data: BatchData::Multiline { - target: String::new(), - message: Message { - metadata: MessageMetadata { - msgid, - message_type: MessageType::Privmsg, - server_time: tags.server_time_with_fallback() as f64, - user: source.to_string(), - }, - text: Some(TextMessage { - content: Default::default(), - reactions: Default::default(), - reply, - redacted: false, - edited: false, - relayed_by: tags.relayed_by, - }), - }, - }, - }); - } - _ => { - self.current_batches.push(CurrentBatch { - id: id.to_string(), - data: BatchData::Unhandled, - }); - warn!(?typ, ?param, "unhandled BATCH type"); - } - } - } else { - assert_eq!( - self.current_batches.iter().last().map(|b| b.id.as_str()), - Some(&reference[1..]) - ); - - if let Some(batch) = self.current_batches.pop() { - match batch.data { - BatchData::History { - label, - channel: channel_name, - messages, - } => { - let channel = self.channel_mut(channel_name.clone()).await; - for message in &messages { - channel - .messages - .insert(message.metadata.msgid.clone(), message.clone()); - } - - let history = History { - channel: channel_name.clone(), - messages, - }; - - let key = if let Some(label) = label { - CommandKey::Label(label) - } else { - CommandKey::History - }; - - self.response_channels - .reply( - &CommandKey::Join(channel_name.clone()), - CommandResponse::Join(channel_name.clone()), - ) - .map_err(|e| { - anyhow!("Failed to reply to JOIN command {e:?}") - })?; - - self.response_channels - .reply(&key, CommandResponse::History(history.clone())) - .unwrap(); - } - - BatchData::Multiline { - target, - message: state_message, - } => { - let source = message.source_nickname().unwrap(); - - if self - .push_batch(target.to_string(), state_message.clone()) - .await - { - return Ok(()); - } - - if let Some(username) = tags.account.clone() { - let user = self.user_mut(source.to_string()).await; - user.username = Some(username); - } - - let channel = self.channel_mut(target.to_string()).await; - channel.messages.insert( - state_message.metadata.msgid.clone(), - state_message.clone(), - ); - - if source == self.state.me.as_ref().unwrap().nickname - && let Err(e) = self.response_channels.reply( - &CommandKey::Privmsg { - target: target.to_string(), - text: state_message - .text - .as_ref() - .unwrap() - .content - .clone(), - }, - CommandResponse::Privmsg(Box::new(state_message.clone())), - ) - { - error!("Failed to reply to PRIVMSG command {e:?}"); - } - - self.on_event(ServerEvent::Privmsg { - channel: target.to_string(), - message: state_message, - }) - .await?; - } - BatchData::Unhandled => (), - } - } - } - } - ChannelMODE(ref channel_name, ref mode) => { - let target = message.response_target().unwrap(); - let source = message.source_nickname().unwrap(); - - if !self.current_batches.iter().any(|b| b.is_chathistory()) { - dbg!(target, source, channel_name, mode); - } - } - TOPIC(ref channel_name, ref text) => { - let target = message.response_target().unwrap(); - let source = message.source_nickname().unwrap(); - - if !self.current_batches.iter().any(|b| b.is_chathistory()) { - dbg!(target, source, channel_name, text); - } - } - Raw(ref cmd, ref mut target) if cmd == "TAGMSG" => { - let target = target.remove(0); - - let mut tags = Tags::default(); - if let Some(ref t) = message.tags { - tags = Tags::parse(t); - } - - let is_unreact = tags.unreact.is_some(); - if let Some(react) = tags.react.or(tags.unreact) - && let Some(reply) = tags.reply - { - let channel = self.channel_mut(target.clone()).await; - - let nickname = message.source_nickname().unwrap().to_string(); - if let Some(message) = channel.messages.get_mut(&reply) { - let reactors = message - .text - .as_mut() - .unwrap() - .reactions - .entry(react.clone()) - .or_insert_with(Vec::new); - - if is_unreact { - reactors.retain(|v| *v != nickname); - } else { - reactors.push(nickname.clone()); - } - - // TODO: should it be sent if the message wasn't found? - self.on_event(ServerEvent::React { - target_message: reply, - user: nickname, - text: react, - is_unreact, - }) - .await?; - } - } - } - AUTHENTICATE(param) if param == "+" => { - if let SaslState::Requested { - nickname, - realname, - username, - password, - } = self.sasl_state.clone() - { - let credentials = - BASE64_STANDARD.encode(format!("\0{}\0{}", username, password).as_bytes()); - - // Chunk overly long credentials - let mut sending = credentials.as_str(); - while !sending.is_empty() { - let (chunk, rest) = sending.split_at(400.min(sending.len())); - self.sasl(chunk.to_string()) - .await - .context("Failed to send SASL chunk")?; - - if rest.is_empty() && chunk.len() == 400 { - self.sasl("+".to_string()) - .await - .context("Failed to send SASL end")?; - } - sending = rest; - } - self.nick(nickname.clone()) - .await - .context("Failed to send NICK")?; - self.user(username.clone(), String::from("0"), realname.clone()) - .await - .context("Failed to send USER")?; - - self.state.me = Some(User { - nickname, - username: Some(username), - realname: Some(realname), - display_name: None, - description: None, - profile_picture_url: None, - bot: false, - }); - } - } - ERROR(msg) => self.on_error(OrbitError::Generic(msg)).await?, - _ => { - warn!("unhandled message, {message:?}"); - } - } - - Ok(()) - } - - pub async fn push_batch(&mut self, target: String, state_message: Message) -> bool { + pub(crate) async fn push_batch(&mut self, target: String, state_message: Message) -> bool { if let Some(batch) = self.current_batches.iter_mut().find(|b| b.is_chathistory()) && let BatchData::History { channel, messages, .. @@ -882,275 +251,7 @@ impl IrcActor { } #[tracing::instrument(err, skip(self))] - pub async fn handle_caps( - &mut self, - sub: CapSubCommand, - param: Option, - caps: Option, - ) -> Result<(), OrbitError> { - match sub { - CapSubCommand::LS if let Some(caps) = caps => { - for cap in caps.split_whitespace() { - let cap = cap - .split('=') - .next() - .ok_or_else(|| anyhow!("Cap is empty: \"{}\"", cap))?; - self.state.capabilities.set_from_name(cap, None); - } - } - CapSubCommand::LS if let Some(param) = param => { - if param == "*" { - return Ok(()); - } - for cap in param.split_whitespace() { - let cap = cap - .split('=') - .next() - .ok_or_else(|| anyhow!("Cap is empty: \"{}\"", cap))?; - self.state.capabilities.set_from_name(cap, None); - } - } - CapSubCommand::ACK if let Some(param) = param => { - for cap in param.split_whitespace() { - self.state.capabilities.set_from_name(cap, Some(true)); - } - self.response_channels - .reply(&CommandKey::RequestCaps, CommandResponse::Capabilities) - .unwrap(); - } - _ => { - debug!("unhandled caps message"); - } - } - - Ok(()) - } - - #[tracing::instrument(err, skip(self))] - pub async fn handle_response( - &mut self, - rpl: Response, - params: Vec, - ) -> Result<(), OrbitError> { - match rpl { - Response::RPL_MOTDSTART => { - self.state.metadata.reset_motd(); - } - Response::RPL_MOTD => { - self.state.metadata.add_motd(¶ms[1]); - } - Response::RPL_ENDOFMOTD => self - .on_event(ServerEvent::ServerInfo(self.state.metadata.clone())) - .await - .map_err(|e| anyhow!("Failed to send server event {e:?}"))?, - Response::RPL_SASLSUCCESS => { - self.response_channels - .reply( - &CommandKey::SignIn, - CommandResponse::SignIn(Ok(SignedIn::User)), - ) - .map_err(|e| anyhow!("Failed to reply to sign in command {e:?}"))?; - - self.cap_end().await.context("Failed to send CAP END")?; - } - Response::RPL_WELCOME => { - self.response_channels - .reply( - &CommandKey::SignIn, - CommandResponse::SignIn(Ok(SignedIn::Guest)), - ) - .map_err(|e| anyhow!("Failed to reply to sign in command {e:?}"))?; - } - Response::RPL_LOGGEDIN => { - self.state.me.as_mut().unwrap().username = Some(params[2].clone()); - } - Response::ERR_SASLFAIL => { - self.response_channels - .reply( - &CommandKey::SignIn, - CommandResponse::SignIn(Err(OrbitError::SaslFailed(params[1].to_string()))), - ) - .map_err(|e| anyhow!("Failed to reply to sign in command {e:?}"))?; - } - Response::ERR_NICKNAMEINUSE => { - self.response_channels - .reply( - &CommandKey::SignIn, - CommandResponse::SignIn(Err(OrbitError::NickTaken)), - ) - .map_err(|e| anyhow!("Failed to reply to sign in command {e:?}"))?; - } - Response::RPL_TOPIC => { - let channel_name = params[1].to_string(); - let topic = params[2].to_string(); - let channel = self.channel_mut(channel_name).await; - - channel.metadata.topic = Some(topic); - - let metadata = channel.metadata.clone(); - self.on_event(ServerEvent::ChannelUpdated(metadata)) - .await - .unwrap(); - } - Response::RPL_NAMREPLY => { - let channel_name = params[2].to_string(); - let users: Vec<_> = params[3].split_whitespace().map(|u| u.to_owned()).collect(); - - let mut channel_users = Vec::new(); - for mut user in users { - if let Some(prefix) = &self.state.support.prefix - && let Some((role, _)) = - prefix.iter().find(|(_, p)| Some(*p) == user.chars().nth(0)) - { - user.remove(0); - let role = ChannelRole::from(*role); - channel_users.push(ChannelUser { - role, - nickname: user.clone(), - }); - } else { - channel_users.push(ChannelUser { - role: ChannelRole::Regular, - nickname: user.clone(), - }); - } - - self.state - .users - .entry(user.clone()) - .or_insert_with(|| User::new(user)); - } - - let channel = self.channel_mut(channel_name).await; - - channel.users = channel_users; - } - Response::RPL_ENDOFNAMES => self - .on_event(ServerEvent::UserList { - channel: params[1].to_string(), - users: self.state.channels.get(¶ms[1]).unwrap().users.clone(), - }) - .await - .map_err(|e| anyhow!("Failed to send server event {e:?}"))?, - Response::RPL_ISUPPORT => { - for option in ¶ms[1..(params.len() - 1)] { - let (key, value) = option.split_once('=').unzip(); - self.state.support.set(key.unwrap_or(option), value); - } - self.state.metadata.name = self.state.support.network.clone(); - } - Response::RPL_YOURHOST - | Response::RPL_CREATED - | Response::RPL_MYINFO - | Response::RPL_LUSERCLIENT - | Response::RPL_LUSEROP - | Response::RPL_LUSERUNKNOWN - | Response::RPL_LUSERCHANNELS - | Response::RPL_LUSERME - | Response::RPL_TOPICWHOTIME - | Response::RPL_LOCALUSERS - | Response::RPL_UMODEIS - | Response::RPL_GLOBALUSERS => (), - _ => { - warn!("unhandled response"); - } - } - - Ok(()) - } - - #[tracing::instrument(err, skip(self))] - pub async fn handle_command(&mut self, cmd: ActorMessage) -> Result<(), OrbitError> { - match cmd.command { - ActorCommand::GetState => cmd - .reply_tx - .unwrap() - .send(CommandResponse::GetState(Box::new(self.state.clone()))) - .unwrap(), - ActorCommand::GetChannelState(channel_name) => cmd - .reply_tx - .unwrap() - .send(CommandResponse::GetChannelState(Box::new( - self.state.channels.get(&channel_name).cloned(), - ))) - .unwrap(), - ActorCommand::SignIn { - nick, - user, - realname, - password, - } => { - self.response_channels - .register(CommandKey::SignIn, cmd.reply_tx.unwrap()); - self.sign_in(nick, user, realname, password).await?; - } - ActorCommand::SignInAnonymous { - nick, - user, - realname, - } => { - self.response_channels - .register(CommandKey::SignIn, cmd.reply_tx.unwrap()); - self.sign_in_anonymous(nick, user, realname).await?; - } - ActorCommand::Join { channel, password } => { - self.response_channels - .register(CommandKey::Join(channel.clone()), cmd.reply_tx.unwrap()); - self.join(channel, password).await.unwrap(); - } - ActorCommand::Privmsg { text, target } => { - self.response_channels.register( - CommandKey::Privmsg { - target: target.clone(), - text: text.clone(), - }, - cmd.reply_tx.unwrap(), - ); - self.privmsg(target, text).await.unwrap(); - } - ActorCommand::AddEventHandler { handler } => { - self.event_handlers.push(handler); - } - ActorCommand::AddErrorHandler { handler } => { - self.error_handlers.push(handler); - } - ActorCommand::AddDisconectHandler { handler } => { - self.disconnect_handlers.push(handler); - } - ActorCommand::RequestHistory { - channel, - before_msgid, - } => { - let label = if self.state.capabilities.labeled_response.enabled { - Some( - self.response_channels - .register_labeled(cmd.reply_tx.unwrap()), - ) - } else { - self.response_channels - .register(CommandKey::History, cmd.reply_tx.unwrap()); - - None - }; - - self.requested_history_batches.push(( - RequestedHistory { - channel: channel.clone(), - label: label.clone(), - }, - Instant::now(), - )); - self.history_before(channel, format!("msgid={before_msgid}"), 5, label) - .await - .context("Failed to send history before")?; - } - } - - Ok(()) - } - - #[tracing::instrument(err, skip(self))] - pub async fn on_event(&mut self, event: ServerEvent) -> Result<(), OrbitError> { + pub(crate) async fn on_event(&mut self, event: ServerEvent) -> Result<(), OrbitError> { for handler in &mut self.event_handlers { handler.send(event.clone()).await.unwrap(); } @@ -1159,7 +260,7 @@ impl IrcActor { } #[tracing::instrument(err, skip(self))] - pub async fn on_error(&mut self, error: OrbitError) -> Result<(), OrbitError> { + pub(crate) async fn on_error(&mut self, error: OrbitError) -> Result<(), OrbitError> { for handler in &mut self.error_handlers { handler.send(error.clone()).await.unwrap(); } @@ -1168,7 +269,7 @@ impl IrcActor { } #[tracing::instrument(err, skip(self))] - pub async fn on_disconnect(&mut self, reason: String) -> Result<(), OrbitError> { + pub(crate) async fn on_disconnect(&mut self, reason: String) -> Result<(), OrbitError> { for handler in &mut self.disconnect_handlers { handler.send(reason.clone()).await.unwrap(); } @@ -1177,7 +278,7 @@ impl IrcActor { } #[tracing::instrument(err, skip(self))] - async fn request_caps(&mut self) -> Result<(), OrbitError> { + pub(crate) async fn request_caps(&mut self) -> Result<(), OrbitError> { let irc_version = String::from("302"); self.cap_ls(irc_version) .await @@ -1203,7 +304,7 @@ impl IrcActor { } #[tracing::instrument(err, skip(self))] - async fn sign_in_anonymous( + pub(crate) async fn sign_in_anonymous( &mut self, nickname: String, username: String, @@ -1231,7 +332,7 @@ impl IrcActor { } #[tracing::instrument(err, skip(self))] - async fn sign_in( + pub(crate) async fn sign_in( &mut self, nickname: String, username: String, @@ -1256,14 +357,14 @@ impl IrcActor { Ok(()) } - async fn channel_mut(&mut self, name: String) -> &mut Channel { + pub(crate) async fn channel_mut(&mut self, name: String) -> &mut Channel { self.state .channels .entry(name.clone()) .or_insert_with(|| Channel::new(name)) } - async fn user_mut(&mut self, nickname: String) -> &mut User { + pub(crate) async fn user_mut(&mut self, nickname: String) -> &mut User { self.state .users .entry(nickname.clone()) @@ -1271,7 +372,7 @@ impl IrcActor { } } -impl SendCommand for IrcActor { +impl SendCommand for IrcActor { type Error = ::Error; async fn message(&mut self, command: IrcMessage) -> Result<(), Self::Error> { self.outgoing.message(command).await diff --git a/packages/core/core-shared/src/database.rs b/packages/core/core-shared/src/database.rs new file mode 100644 index 0000000..28450e6 --- /dev/null +++ b/packages/core/core-shared/src/database.rs @@ -0,0 +1,33 @@ +use std::fmt; + +use crate::state::{Message, OrbitError}; + +pub trait Database: fmt::Debug + Sized { + fn insert_message( + &mut self, + channel: &str, + message: Message, + ) -> impl Future>; + + fn message( + &mut self, + msgid: &str, + ) -> impl Future, OrbitError>>; + + fn messages(&mut self, channel: &str) + -> impl Future, OrbitError>>; + + fn add_reaction( + &mut self, + msgid: &str, + react: &str, + reactor: &str, + ) -> impl Future>; + + fn remove_reaction( + &mut self, + msgid: &str, + react: &str, + reactor: &str, + ) -> impl Future>; +} diff --git a/packages/core/core-shared/src/handlers.rs b/packages/core/core-shared/src/handlers.rs new file mode 100644 index 0000000..33aaa1a --- /dev/null +++ b/packages/core/core-shared/src/handlers.rs @@ -0,0 +1,898 @@ +use std::collections::HashMap; +#[cfg(not(feature = "web"))] +use std::time::Instant; +#[cfg(feature = "web")] +use web_time::Instant; + +#[cfg(feature = "web")] +use crate::dbg; +use crate::{ + SendCommand, + actor::{ + ActorCommand, ActorMessage, BatchData, CurrentBatch, IrcActor, IrcConnection, + RequestedHistory, SaslState, + }, + database::Database, + response_channels::{CommandKey, CommandResponse, generate_label}, + state::{ + Channel, ChannelRole, ChannelUser, History, Message, MessageMetadata, MessageReference, + MessageType, OrbitError, ServerEvent, SignedIn, Tags, TextMessage, User, + }, +}; +use anyhow::{Context, anyhow}; +use base64::prelude::*; +use irc_proto::{BatchSubCommand, CapSubCommand, Command::*, Message as IrcMessage, Response}; +use tracing::{debug, error, warn}; + +impl IrcActor { + #[tracing::instrument(err, skip(self))] + pub(crate) async fn handle_incoming( + &mut self, + mut message: IrcMessage, + ) -> Result<(), OrbitError> { + match message.command { + CAP(_, sub, param, caps) => self.handle_caps(sub, param, caps).await?, + PING(server1, server2) => { + self.outgoing + .pong(server1, server2) + .await + .context("Failed to send pong")?; + } + + Response(rpl, params) => self.handle_response(rpl, params).await?, + + JOIN(ref channel_name, _, _) => self.handle_join(&message, channel_name).await?, + PART(ref channel_name, ref comment) => { + self.handle_part(&message, channel_name, comment).await? + } + QUIT(ref comment) => self.handle_quit(&message, comment).await?, + PRIVMSG(ref target, ref text) => self.handle_privmsg(&message, target, text).await?, + BATCH(ref reference, ref typ, ref param) => { + self.handle_batch(&message, reference, typ, param).await? + } + + ChannelMODE(ref channel_name, ref mode) => { + let target = message.response_target().unwrap(); + let source = message.source_nickname().unwrap(); + + if !self.current_batches.iter().any(|b| b.is_chathistory()) { + dbg!(target, source, channel_name, mode); + } + } + TOPIC(ref channel_name, ref text) => { + let target = message.response_target().unwrap(); + let source = message.source_nickname().unwrap(); + + if !self.current_batches.iter().any(|b| b.is_chathistory()) { + dbg!(target, source, channel_name, text); + } + } + Raw(ref cmd, ref mut target) if cmd == "TAGMSG" => { + let target = target.remove(0); + self.handle_tagmsg(&message, target).await?; + } + AUTHENTICATE(ref param) if param == "+" => self.handle_authenticate(&message).await?, + ERROR(msg) => self.on_error(OrbitError::Generic(msg)).await?, + _ => { + warn!("unhandled message, {message:?}"); + } + } + + Ok(()) + } + + #[tracing::instrument(err, skip(self))] + pub(crate) async fn handle_part( + &mut self, + message: &IrcMessage, + target: &str, + comment: &Option, + ) -> Result<(), OrbitError> { + let mut tags = Tags::default(); + if let Some(ref t) = message.tags { + tags = Tags::parse(t); + } + let source = message.source_nickname().unwrap(); + + let state_message = Message { + text: None, + metadata: MessageMetadata { + msgid: tags.msgid_with_fallback(&["PART", source]), + server_time: tags.server_time_with_fallback() as f64, + message_type: MessageType::Part, + user: source.to_string(), + }, + }; + + if self + .push_batch(target.to_string(), state_message.clone()) + .await + { + return Ok(()); + } + + self.database + .insert_message(target, state_message.clone()) + .await?; + + self.on_event(ServerEvent::Privmsg { + channel: target.to_string(), + message: state_message, + }) + .await?; + + let channel = self.channel_mut(target.to_string()).await; + + channel.users.retain(|u| u.nickname != source); + + Ok(()) + } + + #[tracing::instrument(err, skip(self))] + pub(crate) async fn handle_quit( + &mut self, + message: &IrcMessage, + comment: &Option, + ) -> Result<(), OrbitError> { + let mut tags = Tags::default(); + if let Some(ref t) = message.tags { + tags = Tags::parse(t); + } + let source = message.source_nickname().unwrap(); + + let state_message = Message { + text: None, + metadata: MessageMetadata { + msgid: tags.msgid_with_fallback(&["QUIT", source]), + server_time: tags.server_time_with_fallback() as f64, + message_type: MessageType::Quit, + user: source.to_string(), + }, + }; + + for batch in &mut self.current_batches { + if let BatchData::History { messages, .. } = &mut batch.data { + messages.push(state_message); + return Ok(()); + } + } + + self.state.users.remove(source); + for channel in self.state.channels.values_mut() { + let before = channel.users.len(); + channel.users.retain(|u| u.nickname != source); + + if before > channel.users.len() { + self.database + .insert_message(&channel.metadata.name, state_message.clone()) + .await?; + } + } + + self.on_event(ServerEvent::Privmsg { + channel: String::new(), + message: state_message, + }) + .await?; + + Ok(()) + } + + #[tracing::instrument(err, skip(self))] + pub(crate) async fn handle_authenticate( + &mut self, + message: &IrcMessage, + ) -> Result<(), OrbitError> { + if let SaslState::Requested { + nickname, + realname, + username, + password, + } = self.sasl_state.clone() + { + let credentials = + BASE64_STANDARD.encode(format!("\0{}\0{}", username, password).as_bytes()); + + // Chunk overly long credentials + let mut sending = credentials.as_str(); + while !sending.is_empty() { + let (chunk, rest) = sending.split_at(400.min(sending.len())); + self.sasl(chunk.to_string()) + .await + .context("Failed to send SASL chunk")?; + + if rest.is_empty() && chunk.len() == 400 { + self.sasl("+".to_string()) + .await + .context("Failed to send SASL end")?; + } + sending = rest; + } + self.nick(nickname.clone()) + .await + .context("Failed to send NICK")?; + self.user(username.clone(), String::from("0"), realname.clone()) + .await + .context("Failed to send USER")?; + + self.state.me = Some(User { + nickname, + username: Some(username), + realname: Some(realname), + display_name: None, + description: None, + profile_picture_url: None, + bot: false, + }); + } + + Ok(()) + } + + #[tracing::instrument(err, skip(self))] + pub(crate) async fn handle_caps( + &mut self, + sub: CapSubCommand, + param: Option, + caps: Option, + ) -> Result<(), OrbitError> { + match sub { + CapSubCommand::LS if let Some(caps) = caps => { + for cap in caps.split_whitespace() { + let cap = cap + .split('=') + .next() + .ok_or_else(|| anyhow!("Cap is empty: \"{}\"", cap))?; + self.state.capabilities.set_from_name(cap, None); + } + } + CapSubCommand::LS if let Some(param) = param => { + if param == "*" { + return Ok(()); + } + for cap in param.split_whitespace() { + let cap = cap + .split('=') + .next() + .ok_or_else(|| anyhow!("Cap is empty: \"{}\"", cap))?; + self.state.capabilities.set_from_name(cap, None); + } + } + CapSubCommand::ACK if let Some(param) = param => { + for cap in param.split_whitespace() { + self.state.capabilities.set_from_name(cap, Some(true)); + } + self.response_channels + .reply(&CommandKey::RequestCaps, CommandResponse::Capabilities) + .unwrap(); + } + _ => { + debug!("unhandled caps message"); + } + } + + Ok(()) + } + + pub(crate) async fn handle_join( + &mut self, + message: &IrcMessage, + target: &str, + ) -> Result<(), OrbitError> { + let source = message.source_nickname().unwrap(); + + let mut tags = Tags::default(); + if let Some(ref t) = message.tags { + tags = Tags::parse(t); + } + + if self.current_batches.iter().any(|b| b.is_chathistory()) { + assert!(tags.server_time.is_some()); + } + + let state_message = Message { + text: None, + metadata: MessageMetadata { + msgid: tags.msgid_with_fallback(&["JOIN", source]), + server_time: tags.server_time_with_fallback() as f64, + message_type: MessageType::Join, + user: source.to_string(), + }, + }; + + if self + .push_batch(target.to_string(), state_message.clone()) + .await + { + return Ok(()); + } + + if source == self.state.me.as_ref().unwrap().nickname { + let channel = Channel::new(target.to_string()); + self.state + .channels + .insert(target.to_string(), channel.clone()); + + if !self.state.capabilities.history.enabled { + self.response_channels + .reply( + &CommandKey::Join(target.to_string()), + CommandResponse::Join(target.to_string()), + ) + .map_err(|e| anyhow!("Failed to reply to JOIN command {e:?}"))?; + } + + self.on_event(ServerEvent::Joined(channel)).await?; + + if self.state.capabilities.history.enabled { + let label = if self.state.capabilities.labeled_response.enabled { + Some(generate_label(&mut self.rng)) + } else { + None + }; + + self.requested_history_batches.push(( + RequestedHistory { + target: target.to_string(), + label: label.clone(), + }, + Instant::now(), + )); + self.history_latest(target.to_string(), None, 5, label) + .await + .context("Failed to request latest history")?; + } + } else { + let channel = self.channel_mut(target.to_string()).await; + + channel.users.push(ChannelUser { + nickname: source.to_string(), + role: ChannelRole::Regular, + }); + } + + self.database + .insert_message(target, state_message.clone()) + .await?; + + self.on_event(ServerEvent::Privmsg { + channel: target.to_string(), + message: state_message, + }) + .await?; + + Ok(()) + } + + pub(crate) async fn handle_privmsg( + &mut self, + message: &IrcMessage, + target: &String, + text: &str, + ) -> Result<(), OrbitError> { + let mut tags = Tags::default(); + if let Some(ref t) = message.tags { + tags = Tags::parse(t); + } + assert_eq!( + self.current_batches.iter().last().map(|b| b.id.as_str()), + tags.batch.as_deref(), + ); + + if let Some(batch) = self.current_batches.iter_mut().last() + && let BatchData::Multiline { + message, + target: channel, + } = &mut batch.data + && let Some(t) = message.text.as_mut() + { + if t.content.is_empty() { + *channel = target.to_string(); + t.content = text.to_string(); + } else { + t.content = format!("{}\n{}", t.content, text); + } + return Ok(()); + } + + let source = message.source_nickname().unwrap(); + let msgid = tags.msgid_with_fallback(&["PRIVMSG", source, target, text]); + let reply = self.reply_reference(&tags.reply).await?; + + let state_message = Message { + metadata: MessageMetadata { + msgid: msgid.clone(), + server_time: tags.server_time_with_fallback() as f64, + message_type: MessageType::Privmsg, + user: source.to_string(), + }, + text: Some(TextMessage { + content: text.to_string(), + reactions: HashMap::new(), + reply, + redacted: false, + edited: false, + relayed_by: tags.relayed_by, + }), + }; + + if self.push_batch(target.clone(), state_message.clone()).await { + return Ok(()); + } + + if let Some(username) = tags.account.clone() { + let user = self.user_mut(source.to_string()).await; + user.username = Some(username); + } + + self.database + .insert_message(target, state_message.clone()) + .await?; + + if source == self.state.me.as_ref().unwrap().nickname + && let Err(e) = self.response_channels.reply( + &CommandKey::Privmsg { + target: target.clone(), + text: text.to_string(), + }, + CommandResponse::Privmsg(Box::new(state_message.clone())), + ) + { + error!("Failed to reply to PRIVMSG command {e:?}"); + } + + self.on_event(ServerEvent::Privmsg { + channel: target.clone(), + message: state_message, + }) + .await?; + + Ok(()) + } + + async fn reply_reference( + &mut self, + reply: &Option, + ) -> Result, OrbitError> { + if let Some(r) = reply { + let rmsg = self.database.message(r).await?.map(|(_, m)| m); + + Ok(Some(MessageReference { + text: rmsg + .as_ref() + .and_then(|m| m.text.clone().map(|t| t.content)), + username: rmsg.map(|m| m.metadata.user), + })) + } else { + Ok(None) + } + } + + pub(crate) async fn handle_batch( + &mut self, + message: &IrcMessage, + reference: &str, + typ: &Option, + param: &Option>, + ) -> Result<(), OrbitError> { + let mut tags = Tags::default(); + if let Some(ref t) = message.tags { + tags = Tags::parse(t); + } + if let Some(id) = reference.strip_prefix('+') { + match typ { + Some(BatchSubCommand::CUSTOM(c)) if c.as_str() == "CHATHISTORY" => { + let idx = self + .requested_history_batches + .iter() + .position(|b| b.0.label == tags.label) + .expect("Chat history was requested"); + let channel = self.requested_history_batches.remove(idx).0.target; + + self.current_batches.push(CurrentBatch { + id: id.to_string(), + data: BatchData::History { + label: tags.label, + channel, + messages: Vec::new(), + }, + }); + } + Some(BatchSubCommand::CUSTOM(c)) if c.as_str() == "DRAFT/MULTILINE" => { + let source = message.source_nickname().unwrap(); + let target = message.response_target().unwrap(); + let msgid = tags.msgid_with_fallback(&["MULTILINE", source, target]); + + let reply = self.reply_reference(&tags.reply).await?; + + self.current_batches.push(CurrentBatch { + id: id.to_string(), + data: BatchData::Multiline { + target: String::new(), + message: Message { + metadata: MessageMetadata { + msgid, + message_type: MessageType::Privmsg, + server_time: tags.server_time_with_fallback() as f64, + user: source.to_string(), + }, + text: Some(TextMessage { + content: Default::default(), + reactions: Default::default(), + reply, + redacted: false, + edited: false, + relayed_by: tags.relayed_by, + }), + }, + }, + }); + } + _ => { + self.current_batches.push(CurrentBatch { + id: id.to_string(), + data: BatchData::Unhandled, + }); + warn!(?typ, ?param, "unhandled BATCH type"); + } + } + } else { + assert_eq!( + self.current_batches.iter().last().map(|b| b.id.as_str()), + Some(&reference[1..]) + ); + + if let Some(batch) = self.current_batches.pop() { + match batch.data { + BatchData::History { + label, + channel: channel_name, + messages, + } => { + for message in &messages { + self.database + .insert_message(&channel_name, message.clone()) + .await?; + } + + let history = History { + channel: channel_name.clone(), + messages, + }; + + let key = if let Some(label) = label { + CommandKey::Label(label) + } else { + CommandKey::History + }; + + self.response_channels + .reply( + &CommandKey::Join(channel_name.clone()), + CommandResponse::Join(channel_name.clone()), + ) + .map_err(|e| anyhow!("Failed to reply to JOIN command {e:?}"))?; + + self.response_channels + .reply(&key, CommandResponse::History(history.clone())) + .unwrap(); + } + + BatchData::Multiline { + target, + message: state_message, + } => { + let source = message.source_nickname().unwrap(); + + if self + .push_batch(target.to_string(), state_message.clone()) + .await + { + return Ok(()); + } + + if let Some(username) = tags.account.clone() { + let user = self.user_mut(source.to_string()).await; + user.username = Some(username); + } + + self.database + .insert_message(target.as_str(), state_message.clone()) + .await?; + + if source == self.state.me.as_ref().unwrap().nickname + && let Err(e) = self.response_channels.reply( + &CommandKey::Privmsg { + target: target.to_string(), + text: state_message.text.as_ref().unwrap().content.clone(), + }, + CommandResponse::Privmsg(Box::new(state_message.clone())), + ) + { + error!("Failed to reply to PRIVMSG command {e:?}"); + } + + self.on_event(ServerEvent::Privmsg { + channel: target.to_string(), + message: state_message, + }) + .await?; + } + BatchData::Unhandled => (), + } + } + } + + Ok(()) + } + + pub(crate) async fn handle_tagmsg( + &mut self, + message: &IrcMessage, + target: String, + ) -> Result<(), OrbitError> { + let mut tags = Tags::default(); + if let Some(ref t) = message.tags { + tags = Tags::parse(t); + } + + let is_unreact = tags.unreact.is_some(); + if let Some(react) = tags.react.or(tags.unreact) + && let Some(reply) = tags.reply + { + let reactor = message.source_nickname().unwrap().to_string(); + if is_unreact { + self.database + .remove_reaction(&reply, &react, &reactor) + .await?; + } else { + self.database.add_reaction(&reply, &react, &reactor).await?; + } + + self.on_event(ServerEvent::React { + target_message: reply, + user: reactor, + text: react, + is_unreact, + }) + .await?; + } + Ok(()) + } + + #[tracing::instrument(err, skip(self))] + pub(crate) async fn handle_response( + &mut self, + rpl: Response, + params: Vec, + ) -> Result<(), OrbitError> { + match rpl { + Response::RPL_MOTDSTART => { + self.state.metadata.reset_motd(); + } + Response::RPL_MOTD => { + self.state.metadata.add_motd(¶ms[1]); + } + Response::RPL_ENDOFMOTD => self + .on_event(ServerEvent::ServerInfo(self.state.metadata.clone())) + .await + .map_err(|e| anyhow!("Failed to send server event {e:?}"))?, + Response::RPL_SASLSUCCESS => { + self.response_channels + .reply( + &CommandKey::SignIn, + CommandResponse::SignIn(Ok(SignedIn::User)), + ) + .map_err(|e| anyhow!("Failed to reply to sign in command {e:?}"))?; + + self.cap_end().await.context("Failed to send CAP END")?; + } + Response::RPL_WELCOME => { + self.response_channels + .reply( + &CommandKey::SignIn, + CommandResponse::SignIn(Ok(SignedIn::Guest)), + ) + .map_err(|e| anyhow!("Failed to reply to sign in command {e:?}"))?; + } + Response::RPL_LOGGEDIN => { + self.state.me.as_mut().unwrap().username = Some(params[2].clone()); + } + Response::ERR_SASLFAIL => { + self.response_channels + .reply( + &CommandKey::SignIn, + CommandResponse::SignIn(Err(OrbitError::SaslFailed(params[1].to_string()))), + ) + .map_err(|e| anyhow!("Failed to reply to sign in command {e:?}"))?; + } + Response::ERR_NICKNAMEINUSE => { + self.response_channels + .reply( + &CommandKey::SignIn, + CommandResponse::SignIn(Err(OrbitError::NickTaken)), + ) + .map_err(|e| anyhow!("Failed to reply to sign in command {e:?}"))?; + } + Response::RPL_TOPIC => { + let channel_name = params[1].to_string(); + let topic = params[2].to_string(); + let channel = self.channel_mut(channel_name).await; + + channel.metadata.topic = Some(topic); + + let metadata = channel.metadata.clone(); + self.on_event(ServerEvent::ChannelUpdated(metadata)) + .await + .unwrap(); + } + Response::RPL_NAMREPLY => { + let channel_name = params[2].to_string(); + let users: Vec<_> = params[3].split_whitespace().map(|u| u.to_owned()).collect(); + + let mut channel_users = Vec::new(); + for mut user in users { + if let Some(prefix) = &self.state.support.prefix + && let Some((role, _)) = + prefix.iter().find(|(_, p)| Some(*p) == user.chars().nth(0)) + { + user.remove(0); + let role = ChannelRole::from(*role); + channel_users.push(ChannelUser { + role, + nickname: user.clone(), + }); + } else { + channel_users.push(ChannelUser { + role: ChannelRole::Regular, + nickname: user.clone(), + }); + } + + self.state + .users + .entry(user.clone()) + .or_insert_with(|| User::new(user)); + } + + let channel = self.channel_mut(channel_name).await; + + channel.users = channel_users; + } + Response::RPL_ENDOFNAMES => self + .on_event(ServerEvent::UserList { + channel: params[1].to_string(), + users: self.state.channels.get(¶ms[1]).unwrap().users.clone(), + }) + .await + .map_err(|e| anyhow!("Failed to send server event {e:?}"))?, + Response::RPL_ISUPPORT => { + for option in ¶ms[1..(params.len() - 1)] { + let (key, value) = option.split_once('=').unzip(); + self.state.support.set(key.unwrap_or(option), value); + } + self.state.metadata.name = self.state.support.network.clone(); + } + Response::RPL_YOURHOST + | Response::RPL_CREATED + | Response::RPL_MYINFO + | Response::RPL_LUSERCLIENT + | Response::RPL_LUSEROP + | Response::RPL_LUSERUNKNOWN + | Response::RPL_LUSERCHANNELS + | Response::RPL_LUSERME + | Response::RPL_TOPICWHOTIME + | Response::RPL_LOCALUSERS + | Response::RPL_UMODEIS + | Response::RPL_GLOBALUSERS => (), + _ => { + warn!("unhandled response"); + } + } + + Ok(()) + } + + #[tracing::instrument(err, skip(self))] + pub(crate) async fn handle_command(&mut self, cmd: ActorMessage) -> Result<(), OrbitError> { + match cmd.command { + ActorCommand::GetState => { + let mut state = self.state.clone(); + for (name, channel) in &mut state.channels { + channel.messages = self.database.messages(name).await?; + } + + cmd.reply_tx + .unwrap() + .send(CommandResponse::GetState(Box::new(state))) + .unwrap(); + } + ActorCommand::GetChannelState(channel_name) => { + let Some(mut channel) = self.state.channels.get(&channel_name).cloned() else { + cmd.reply_tx + .unwrap() + .send(CommandResponse::GetChannelState(Box::new(None))) + .unwrap(); + return Ok(()); + }; + channel.messages = self.database.messages(&channel_name).await?; + cmd.reply_tx + .unwrap() + .send(CommandResponse::GetChannelState(Box::new(Some(channel)))) + .unwrap(); + } + ActorCommand::SignIn { + nick, + user, + realname, + password, + } => { + self.response_channels + .register(CommandKey::SignIn, cmd.reply_tx.unwrap()); + self.sign_in(nick, user, realname, password).await?; + } + ActorCommand::SignInAnonymous { + nick, + user, + realname, + } => { + self.response_channels + .register(CommandKey::SignIn, cmd.reply_tx.unwrap()); + self.sign_in_anonymous(nick, user, realname).await?; + } + ActorCommand::Join { channel, password } => { + self.response_channels + .register(CommandKey::Join(channel.clone()), cmd.reply_tx.unwrap()); + self.join(channel, password).await.unwrap(); + } + ActorCommand::Privmsg { text, target } => { + self.response_channels.register( + CommandKey::Privmsg { + target: target.clone(), + text: text.clone(), + }, + cmd.reply_tx.unwrap(), + ); + self.privmsg(target, text).await.unwrap(); + } + ActorCommand::AddEventHandler { handler } => { + self.event_handlers.push(handler); + } + ActorCommand::AddErrorHandler { handler } => { + self.error_handlers.push(handler); + } + ActorCommand::AddDisconectHandler { handler } => { + self.disconnect_handlers.push(handler); + } + ActorCommand::RequestHistory { + channel, + before_msgid, + } => { + let label = if self.state.capabilities.labeled_response.enabled { + Some( + self.response_channels + .register_labeled(cmd.reply_tx.unwrap()), + ) + } else { + self.response_channels + .register(CommandKey::History, cmd.reply_tx.unwrap()); + + None + }; + + self.requested_history_batches.push(( + RequestedHistory { + target: channel.clone(), + label: label.clone(), + }, + Instant::now(), + )); + self.history_before(channel, format!("msgid={before_msgid}"), 5, label) + .await + .context("Failed to send history before")?; + } + } + + Ok(()) + } +} diff --git a/packages/core/core-shared/src/lib.rs b/packages/core/core-shared/src/lib.rs index 121e2dd..3e60e41 100644 --- a/packages/core/core-shared/src/lib.rs +++ b/packages/core/core-shared/src/lib.rs @@ -1,4 +1,7 @@ pub mod actor; +pub mod database; +mod handlers; +pub mod response_channels; mod send_command; pub mod state; diff --git a/packages/core/core-shared/src/response_channels.rs b/packages/core/core-shared/src/response_channels.rs new file mode 100644 index 0000000..2bf73a9 --- /dev/null +++ b/packages/core/core-shared/src/response_channels.rs @@ -0,0 +1,109 @@ +use std::time::Duration; + +use rand::{SeedableRng, rngs::SmallRng, seq::IndexedRandom}; +#[cfg(not(feature = "web"))] +use std::time::Instant; +#[cfg(feature = "web")] +use web_time::Instant; + +#[cfg(feature = "web")] +use crate::dbg; +use crate::state::{Channel, History, Message, OrbitError, Server, SignedIn}; +use futures::channel::oneshot; +use tracing::warn; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) enum CommandKey { + RequestCaps, + SignIn, + Join(String), + Privmsg { target: String, text: String }, + History, + Label(String), +} + +#[derive(Debug)] +pub enum CommandResponse { + GetState(Box), + GetChannelState(Box>), + Capabilities, + SignIn(Result), + Join(String), + Privmsg(Box), + History(History), +} + +const LABEL_CHARSET: &str = "abcdefghijklmnopqrstuvwxyz\ + ABCDEFGHIJKLMNOPQRSTUVWXYZ\ + 1234567890"; + +pub(crate) fn generate_label(rng: &mut SmallRng) -> String { + let char_vec = LABEL_CHARSET + .split("") + .filter(|c| !c.is_empty()) + .collect::>(); + + std::iter::repeat_with(|| char_vec.choose(rng).expect("CHARSET is not empty")) + .take(10) + .copied() + .collect::>() + .join("") +} + +#[derive(Debug)] +pub(crate) struct ResponseChannels { + channels: Vec<(CommandKey, Instant, oneshot::Sender)>, + rng: SmallRng, +} + +impl Default for ResponseChannels { + #[tracing::instrument] + fn default() -> Self { + Self { + channels: Vec::new(), + rng: SmallRng::from_seed([0; 32]), + } + } +} + +impl ResponseChannels { + #[tracing::instrument] + pub fn register(&mut self, key: CommandKey, os_tx: oneshot::Sender) { + self.channels.push((key, Instant::now(), os_tx)); + } + + #[tracing::instrument] + pub fn register_labeled(&mut self, os_tx: oneshot::Sender) -> String { + let label = generate_label(&mut self.rng); + + self.channels + .push((CommandKey::Label(label.clone()), Instant::now(), os_tx)); + + label + } + + #[tracing::instrument] + pub fn reply( + &mut self, + key: &CommandKey, + response: CommandResponse, + ) -> Result { + if let Some(idx) = self.channels.iter().position(|(rk, _, _)| rk == key) { + let (_, _, ch) = self.channels.remove(idx); + ch.send(response)?; + + Ok(true) + } else { + if let CommandKey::Label(label) = key { + warn!("Failed to find response channel for label {label:?}"); + } + // warn!("Failed to find response channel"); + Ok(false) + } + } + + pub fn check_timeouts(&mut self) { + self.channels + .retain(|(_, creation, _)| creation.elapsed() < Duration::from_secs(1)); + } +} diff --git a/packages/core/core-shared/src/state.rs b/packages/core/core-shared/src/state.rs index 27aee8c..cccb652 100644 --- a/packages/core/core-shared/src/state.rs +++ b/packages/core/core-shared/src/state.rs @@ -4,7 +4,7 @@ use std::str::FromStr; #[cfg(feature = "web")] use crate::dbg; use irc_proto::message::Tag; -use ordermap::OrderMap; +use serde::{Deserialize, Serialize}; use thiserror::Error; use time::OffsetDateTime; use time::format_description::well_known::Iso8601; @@ -78,7 +78,7 @@ impl ServerMetadata { #[derive(Debug, Clone, PartialEq, Eq)] pub struct Channel { pub metadata: ChannelMetadata, - pub messages: OrderMap, + pub messages: Vec, pub users: Vec, } @@ -493,7 +493,7 @@ pub struct ChannelUser { pub role: ChannelRole, } -#[derive(Debug, Clone)] +#[derive(Debug, Clone, Serialize, Deserialize)] pub struct Message { pub text: Option, pub metadata: MessageMetadata, @@ -507,7 +507,7 @@ impl PartialEq for Message { impl Eq for Message {} -#[derive(Debug, Clone)] +#[derive(Debug, Clone, Serialize, Deserialize)] #[cfg_attr(feature = "web", wasm_bindgen)] pub enum MessageType { Privmsg, @@ -518,17 +518,17 @@ pub enum MessageType { Quit, } -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct TextMessage { pub content: String, - pub reactions: OrderMap>, + pub reactions: HashMap>, pub reply: Option, pub redacted: bool, pub edited: bool, pub relayed_by: Option, } -#[derive(Debug, Clone)] +#[derive(Debug, Clone, Serialize, Deserialize)] #[cfg_attr(feature = "web", derive(Tsify))] #[cfg_attr(feature = "web", wasm_bindgen(getter_with_clone, inspectable))] pub struct MessageMetadata { @@ -546,7 +546,7 @@ impl PartialEq for MessageMetadata { impl Eq for MessageMetadata {} -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[cfg_attr(feature = "web", derive(Tsify))] #[cfg_attr(feature = "web", wasm_bindgen(getter_with_clone, inspectable))] pub struct MessageReference { diff --git a/packages/core/core-shared/tests/mock.rs b/packages/core/core-shared/tests/mock.rs index 365aec1..d500140 100644 --- a/packages/core/core-shared/tests/mock.rs +++ b/packages/core/core-shared/tests/mock.rs @@ -1,5 +1,7 @@ -use core_shared::actor::{CommandResponse, IrcConnection}; -use core_shared::state::{Capabilities, Capability, User}; +use core_shared::actor::IrcConnection; +use core_shared::database::Database; +use core_shared::response_channels::CommandResponse; +use core_shared::state::{Capabilities, Capability, OrbitError, User}; use futures::SinkExt; use futures::channel::mpsc::{UnboundedReceiver, UnboundedSender, unbounded}; use irc_proto::{CapSubCommand, Command, Message as IrcMessage}; @@ -42,6 +44,10 @@ impl IrcConnection for MockConn { Command::CAP(None, CapSubCommand::REQ, None, Some(_)) => { incoming_tx.send(Ok(IrcMessage::from_str("@time=2026-07-13T11:01:09.182Z :irc.hivecom.net CAP * ACK :echo-message message-tags sasl draft/message-redaction draft/metadata-2 draft/chathistory draft/event-playback draft/account-registration draft/multiline server-time").unwrap())).await.unwrap(); } + Command::AUTHENTICATE(param) if param == "PLAIN" => incoming_tx + .send(Ok(IrcMessage::from_str("AUTHENTICATE +").unwrap())) + .await + .unwrap(), Command::USER(_, _, _) => { incoming_tx.send(Ok(IrcMessage::from_str(":irc.hivecom.net 001 testnick :Welcome to the Hivecom IRC Network testnick").unwrap())).await.unwrap(); } @@ -56,9 +62,37 @@ impl IrcConnection for MockConn { } } +#[derive(Debug)] +struct MockDb; + +impl Database for MockDb { + async fn insert_message( + &mut self, + _: &str, + _: core_shared::state::Message, + ) -> Result<(), OrbitError> { + todo!() + } + async fn message( + &mut self, + _: &str, + ) -> Result, OrbitError> { + todo!() + } + async fn messages(&mut self, _: &str) -> Result, OrbitError> { + todo!() + } + async fn add_reaction(&mut self, _: &str, _: &str, _: &str) -> Result<(), OrbitError> { + todo!() + } + async fn remove_reaction(&mut self, _: &str, _: &str, _: &str) -> Result<(), OrbitError> { + todo!() + } +} + #[tokio::test] async fn test_irc_register_flow() { - let addr = IrcActor::::start(0, MockConn, |actor: IrcActor| { + let addr = IrcActor::start(0, MockConn, MockDb, |actor| { tokio::spawn(actor.run()); }) .await diff --git a/packages/core/core-wasm/Cargo.toml b/packages/core/core-wasm/Cargo.toml index 2459cb3..b55c1fc 100644 --- a/packages/core/core-wasm/Cargo.toml +++ b/packages/core/core-wasm/Cargo.toml @@ -23,7 +23,7 @@ gloo-net = "0.7.0" irc-proto = "1.1.0" core-shared = { version = "0.1.0", path = "../core-shared", features = ["web"], default-features = false } -indexed_db_futures = { version = "0.6.4", features = ["async-upgrade", "serde"] } +indexed_db_futures = { version = "0.6.4", features = ["async-upgrade", "cursors", "indices", "serde"] } serde = { version = "1.0.228", features = ["derive"] } anyhow = { version = "1.0.103", features = ["backtrace"] } tracing-wasm = "0.2.1" diff --git a/packages/core/core-wasm/src/database.rs b/packages/core/core-wasm/src/database.rs new file mode 100644 index 0000000..9187d8e --- /dev/null +++ b/packages/core/core-wasm/src/database.rs @@ -0,0 +1,200 @@ +use std::fmt; + +use anyhow::{Context, anyhow}; +use core_shared::{database::Database as ActorDatabase, state::OrbitError}; +use indexed_db_futures::prelude::QuerySource; +use indexed_db_futures::{ + Build, KeyPath, KeyPathSeq, database::Database as InnerDb, transaction::TransactionMode, +}; +use indexed_db_futures::{BuildSerde, KeyRange}; +use serde::{Deserialize, Serialize}; + +use crate::dbg; + +const MESSAGE_STORE: &str = "messages"; +const CHANNEL_TIME_INDEX: &str = "channel-timestamp"; + +pub struct IndexedDb { + inner: InnerDb, +} + +impl fmt::Debug for IndexedDb { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("IndexedDb").finish_non_exhaustive() + } +} + +impl IndexedDb { + pub async fn new(name: &str) -> Result { + let inner = InnerDb::open(name) + .with_on_upgrade_needed_fut(|event, db| async move { + let old_version = event.old_version() as u64; + let new_version = event.new_version().map(|v| v as u64); + + #[allow(clippy::single_match)] + match (old_version, new_version) { + (0, Some(1)) => { + let store = db.create_object_store(MESSAGE_STORE).build()?; + store + .create_index( + CHANNEL_TIME_INDEX, + KeyPath::Sequence(KeyPathSeq::from_slice(&[ + "channel", + "timestamp", + ])), + ) + .build()?; + } + _ => {} + } + + Ok(()) + }) + .await + .map_err(|e| anyhow!(e.to_string())) + .context("Failed to open IndexedDb")?; + + Ok(IndexedDb { inner }) + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct DbMessage { + pub message: core_shared::state::Message, + pub channel: String, + pub timestamp: f64, +} + +impl ActorDatabase for IndexedDb { + #[tracing::instrument(err, skip(self))] + async fn insert_message( + &mut self, + channel: &str, + message: core_shared::state::Message, + ) -> Result<(), OrbitError> { + let tx = self + .inner + .transaction(MESSAGE_STORE) + .with_mode(TransactionMode::Readwrite) + .build() + .map_err(|e| anyhow!(e.to_string())) + .context("Failed to create messages transaction")?; + let store = tx + .object_store(MESSAGE_STORE) + .map_err(|e| anyhow!(e.to_string())) + .context("Failed to open messages object store")?; + + let msgid = message.metadata.msgid.clone(); + let message = DbMessage { + timestamp: message.metadata.server_time, + channel: channel.to_string(), + message, + }; + + store + .put(message) + .with_key(msgid) + .with_key_type::() + .serde() + .map_err(|e| anyhow!(e.to_string())) + .context("Failed to put message")?; + + tx.commit() + .await + .map_err(|e| anyhow!(e.to_string())) + .context("Failed to commit transaction")?; + + Ok(()) + } + + #[tracing::instrument(err, skip(self))] + async fn message( + &mut self, + msgid: &str, + ) -> Result, OrbitError> { + let transaction = self.inner.transaction(MESSAGE_STORE).build().unwrap(); + let store = transaction.object_store(MESSAGE_STORE).unwrap(); + + let message: Option = store.get(msgid).serde().unwrap().await.unwrap(); + + Ok(message.map(|m| (m.channel, m.message))) + } + + #[tracing::instrument(err, skip(self))] + async fn messages( + &mut self, + channel: &str, + ) -> Result, OrbitError> { + let transaction = self.inner.transaction(MESSAGE_STORE).build().unwrap(); + let store = transaction.object_store(MESSAGE_STORE).unwrap(); + let index = store.index(CHANNEL_TIME_INDEX).unwrap(); + + // false = inclusive bound + let range = KeyRange::Bound((channel, 0.0), false, (channel, f64::MAX), false); + + let mut messages = Vec::new(); + for data in index + .get_all::() + .with_query(range) + .serde() + .unwrap() + .await + .unwrap() + { + let data = data.unwrap(); + messages.push(data.message); + } + + Ok(messages) + } + + #[tracing::instrument(err, skip(self))] + async fn add_reaction( + &mut self, + msgid: &str, + react: &str, + reactor: &str, + ) -> Result<(), OrbitError> { + let Some((channel, mut message)) = self.message(msgid).await? else { + return Ok(()); + }; + let Some(text) = &mut message.text else { + return Ok(()); + }; + let reactors = text.reactions.entry(react.to_string()).or_default(); + reactors.push(reactor.to_string()); + reactors.sort(); + reactors.dedup(); + + self.insert_message(&channel, message).await?; + + Ok(()) + } + + #[tracing::instrument(err, skip(self))] + async fn remove_reaction( + &mut self, + msgid: &str, + react: &str, + reactor: &str, + ) -> Result<(), OrbitError> { + let Some((channel, mut message)) = self.message(msgid).await? else { + return Ok(()); + }; + let Some(text) = &mut message.text else { + return Ok(()); + }; + if let Some(reactors) = text.reactions.get_mut(react) + && let Some(pos) = reactors.iter().position(|r| *r == reactor) + { + reactors.remove(pos); + if reactors.is_empty() { + text.reactions.remove(react); + } + } + + self.insert_message(&channel, message).await?; + + Ok(()) + } +} diff --git a/packages/core/core-wasm/src/lib.rs b/packages/core/core-wasm/src/lib.rs index 64a160d..0352413 100644 --- a/packages/core/core-wasm/src/lib.rs +++ b/packages/core/core-wasm/src/lib.rs @@ -3,7 +3,8 @@ use std::{fmt, str::FromStr}; use anyhow::{Context, bail}; use core_shared::{ SendCommand, - actor::{self, ActorCommand, ActorMessage, CommandResponse, IrcActor}, + actor::{self, ActorCommand, ActorMessage, IrcActor}, + response_channels::CommandResponse, state::{ self, Capabilities, ChannelMetadata, ChannelUser, MessageMetadata, MessageReference, ServerMetadata, SignedIn, User, @@ -52,6 +53,12 @@ macro_rules! dbg { use tracing_subscriber::prelude::*; use tracing_subscriber_wasm::MakeConsoleWriter; +use crate::database::IndexedDb; + +mod database; + +const DATABASE_NAME: &str = "obit-core"; + fn init_tracing() { let fmt_layer = tracing_subscriber::fmt::layer() .with_writer(MakeConsoleWriter::default()) @@ -118,7 +125,8 @@ pub struct IrcConnection { impl IrcConnection { async fn connect(id: i32, url: String) -> Result { let connection = WsConnection::new(url)?; - let address = IrcActor::start(id, connection, |actor| { + let database = IndexedDb::new(DATABASE_NAME).await?; + let address = IrcActor::start(id, connection, database, |actor| { spawn_local(async { actor.run().await }) }) .await?; @@ -504,7 +512,7 @@ impl From for Channel { fn from(channel: state::Channel) -> Self { Self { metadata: channel.metadata, - messages: channel.messages.into_values().map(Into::into).collect(), + messages: channel.messages.into_iter().map(Into::into).collect(), users: channel.users, } }