client.rs 30 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733
  1. /* This file is part of DarkFi (https://dark.fi)
  2. *
  3. * Copyright (C) 2020-2026 Dyne.org foundation
  4. *
  5. * This program is free software: you can redistribute it and/or modify
  6. * it under the terms of the GNU Affero General Public License as
  7. * published by the Free Software Foundation, either version 3 of the
  8. * License, or (at your option) any later version.
  9. *
  10. * This program is distributed in the hope that it will be useful,
  11. * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12. * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  13. * GNU Affero General Public License for more details.
  14. *
  15. * You should have received a copy of the GNU Affero General Public License
  16. * along with this program. If not, see <https://www.gnu.org/licenses/>.
  17. */
  18. use std::{
  19. collections::{HashMap, HashSet, VecDeque},
  20. sync::{
  21. atomic::{AtomicBool, AtomicUsize, Ordering::SeqCst},
  22. Arc,
  23. },
  24. };
  25. use darkfi::{
  26. event_graph::{proto::EventPut, Event, NULL_ID},
  27. system::Subscription,
  28. Error, Result,
  29. };
  30. use darkfi_serial::{deserialize_async_partial, serialize_async};
  31. use futures::FutureExt;
  32. use sled_overlay::sled;
  33. use smol::{
  34. io::{self, AsyncBufRead, AsyncBufReadExt, AsyncWriteExt, BufReader},
  35. lock::{OnceCell, RwLock},
  36. net::SocketAddr,
  37. prelude::{AsyncRead, AsyncWrite},
  38. };
  39. use tracing::{debug, error, warn};
  40. use super::{
  41. server::{IrcServer, RlnMessageReservation, MAX_MSG_LEN},
  42. NickServ, SERVER_NAME,
  43. };
  44. use crate::Privmsg;
  45. const PENALTY_LIMIT: usize = 5;
  46. const MAX_IRC_LINE_LEN: usize = 1024;
  47. const MAX_PENDING_PRIVMSGS: usize = 128;
  48. /// Read one IRC line without allowing unbounded buffer growth.
  49. async fn read_bounded_line<R>(reader: &mut R, line: &mut String) -> Result<usize>
  50. where
  51. R: AsyncBufRead + Unpin,
  52. {
  53. line.clear();
  54. let mut bytes = Vec::new();
  55. loop {
  56. let (consumed, complete) = {
  57. let available = reader.fill_buf().await?;
  58. if available.is_empty() {
  59. if bytes.is_empty() {
  60. return Ok(0)
  61. }
  62. *line = String::from_utf8(bytes)?;
  63. return Ok(line.len())
  64. }
  65. let newline = available.iter().position(|b| *b == b'\n');
  66. let take = newline.map_or(available.len(), |idx| idx + 1);
  67. if bytes.len().saturating_add(take) > MAX_IRC_LINE_LEN {
  68. return Err(Error::ParseFailed("IRC line too long"))
  69. }
  70. bytes.extend_from_slice(&available[..take]);
  71. (take, newline.is_some())
  72. };
  73. reader.consume(consumed);
  74. if complete {
  75. *line = String::from_utf8(bytes)?;
  76. return Ok(line.len())
  77. }
  78. }
  79. }
  80. fn enqueue_pending_privmsg(args_queue: &mut VecDeque<Privmsg>, privmsg: Privmsg) -> bool {
  81. if args_queue.len() >= MAX_PENDING_PRIVMSGS {
  82. return false
  83. }
  84. args_queue.push_back(privmsg);
  85. true
  86. }
  87. /// Reply types, we can either send server replies, or client replies.
  88. pub enum ReplyType {
  89. /// Server reply, we have to use numerics
  90. Server((u16, String)),
  91. /// Client reply, message from someone to some{one,where}
  92. Client((String, String)),
  93. /// Pong reply, we just use server origin
  94. Pong(String),
  95. /// CAP reply
  96. Cap(String),
  97. /// NOTICE reply (from, to, what)
  98. Notice((String, String, String)),
  99. }
  100. /// Stateful IRC client handler, used for each client connection
  101. pub struct Client {
  102. /// Pointer to parent `IrcServer`
  103. pub server: Arc<IrcServer>,
  104. /// Subscription for incoming events
  105. pub incoming: Subscription<Event>,
  106. /// Subscription for incoming static events
  107. pub incoming_st: Subscription<Event>,
  108. /// Client socket addr
  109. pub addr: SocketAddr,
  110. /// ID of the last sent event
  111. pub last_sent: RwLock<blake3::Hash>,
  112. /// Active (joined) channels for this client
  113. pub channels: RwLock<HashSet<String>>,
  114. /// Penalty counter, when limit is reached, disconnect client
  115. pub penalty: AtomicUsize,
  116. /// Registration marker
  117. pub registered: AtomicBool,
  118. /// Registration pause marker
  119. pub reg_paused: AtomicBool,
  120. /// CAP END marker
  121. pub is_cap_end: AtomicBool,
  122. /// Password setup marker
  123. pub is_pass_set: AtomicBool,
  124. /// Client username
  125. pub username: Arc<RwLock<String>>,
  126. /// Client nickname
  127. pub nickname: Arc<RwLock<String>>,
  128. /// Client realname
  129. pub realname: RwLock<String>,
  130. /// Client caps
  131. pub caps: RwLock<HashMap<String, bool>>,
  132. /// Set of seen messages for the user
  133. /// TODO: It grows indefinitely, needs to be pruned.
  134. pub seen: OnceCell<sled::Tree>,
  135. /// NickServ instance
  136. pub nickserv: Arc<NickServ>,
  137. }
  138. impl Client {
  139. /// Instantiate a new Client.
  140. pub async fn new(
  141. server: Arc<IrcServer>,
  142. incoming: Subscription<Event>,
  143. incoming_st: Subscription<Event>,
  144. addr: SocketAddr,
  145. ) -> Result<Self> {
  146. let caps =
  147. HashMap::from([("no-history".to_string(), false), ("no-autojoin".to_string(), false)]);
  148. let username = Arc::new(RwLock::new(String::from("*")));
  149. let nickname = Arc::new(RwLock::new(String::from("*")));
  150. Ok(Self {
  151. server: server.clone(),
  152. incoming,
  153. incoming_st,
  154. addr,
  155. last_sent: RwLock::new(NULL_ID),
  156. channels: RwLock::new(HashSet::new()),
  157. penalty: AtomicUsize::new(0),
  158. registered: AtomicBool::new(false),
  159. reg_paused: AtomicBool::new(false),
  160. is_cap_end: AtomicBool::new(false),
  161. is_pass_set: AtomicBool::new(false),
  162. username: username.clone(),
  163. nickname: nickname.clone(),
  164. realname: RwLock::new(String::from("*")),
  165. caps: RwLock::new(caps),
  166. seen: OnceCell::new(),
  167. nickserv: Arc::new(
  168. NickServ::new(username.clone(), nickname.clone(), server.clone()).await?,
  169. ),
  170. })
  171. }
  172. /// This function handles a single IRC client. We listen to messages from the
  173. /// IRC client and relay them to the network, and we also get notified of
  174. /// incoming messages and relay them to the IRC client. The notifications come
  175. /// from events being inserted into the Event Graph.
  176. pub async fn multiplex_connection<S>(&self, stream: S) -> Result<()>
  177. where
  178. S: AsyncRead + AsyncWrite + Unpin + Send + 'static,
  179. {
  180. let (reader, mut writer) = io::split(stream);
  181. let mut reader = BufReader::new(reader);
  182. // Our buffer for the client line
  183. let mut line = String::new();
  184. let mut args_queue: VecDeque<_> = VecDeque::new();
  185. loop {
  186. futures::select! {
  187. // Process message from the IRC client
  188. r = read_bounded_line(&mut reader, &mut line).fuse() => {
  189. // If client closed unexpectedly, we disconnect.
  190. if let Ok(0) = r {
  191. error!("[IRC CLIENT] Read failed for {}: Client disconnected", self.addr);
  192. self.incoming.unsubscribe().await;
  193. self.incoming_st.unsubscribe().await;
  194. return Err(Error::ChannelStopped)
  195. }
  196. // If something failed during reading, we disconnect.
  197. if let Err(e) = r {
  198. error!("[IRC CLIENT] Read failed for {}: {e}", self.addr);
  199. self.incoming.unsubscribe().await;
  200. self.incoming_st.unsubscribe().await;
  201. return Err(Error::ChannelStopped)
  202. }
  203. // If the penalty limit is reached, disconnect the client.
  204. if self.penalty.load(SeqCst) == PENALTY_LIMIT {
  205. self.incoming.unsubscribe().await;
  206. self.incoming_st.unsubscribe().await;
  207. return Err(Error::ChannelStopped)
  208. }
  209. // We'll be strict here and disconnect the client
  210. // in case line processing failed in any way.
  211. match self.process_client_line(&line, &mut writer, &mut args_queue).await {
  212. // If we got an event back, we should broadcast it.
  213. // This means we add it to our DAG, and the DAG will
  214. // handle the rest of the propagation.
  215. Ok(Some(events)) => {
  216. for event in events {
  217. // Update the last sent event.
  218. let event_id = event.header.id();
  219. *self.last_sent.write().await = event_id;
  220. let current_genesis = self.server.darkirc.event_graph.current_genesis.read().await;
  221. let dag_name = current_genesis.header.timestamp.to_string();
  222. drop(current_genesis);
  223. // Build the RLN signal blob before touching the local
  224. // DAG when RLN is enabled. With RLN disabled, outbound
  225. // events deliberately carry no proof blob.
  226. let blob = if self.server.darkirc.event_graph.rln_enabled() {
  227. let (rln_identity, mid) = match self
  228. .server
  229. .reserve_rln_message_id(event.header.timestamp)
  230. .await?
  231. {
  232. RlnMessageReservation::Reserved {
  233. identity,
  234. message_id,
  235. } => (identity, message_id),
  236. RlnMessageReservation::MissingIdentity => {
  237. warn!(
  238. "[IRC CLIENT] No RLN identity registered; \
  239. refusing to send. Use \
  240. `/msg NickServ REGISTER ...` to register."
  241. );
  242. continue
  243. }
  244. RlnMessageReservation::BudgetExhausted => {
  245. warn!(
  246. "[IRC CLIENT] RLN message budget \
  247. exhausted for this epoch; dropping \
  248. message to avoid slash"
  249. );
  250. continue
  251. }
  252. };
  253. match rln_identity
  254. .create_signal(
  255. &event,
  256. mid,
  257. &self.server.darkirc.event_graph,
  258. )
  259. .await
  260. {
  261. Ok(blob) => serialize_async(&blob).await,
  262. Err(e) => {
  263. error!(
  264. "[IRC CLIENT] Failed creating RLN \
  265. signal proof: {e}"
  266. );
  267. return Err(e)
  268. }
  269. }
  270. } else {
  271. Vec::new()
  272. };
  273. // Commit our outbound signal through
  274. // the safe public API. It inserts the
  275. // header, verifies and stores the RLN
  276. // blob, then commits the event body.
  277. if let Err(e) = self
  278. .server
  279. .darkirc
  280. .event_graph
  281. .insert_signal_with_blob(&event, &blob, &dag_name)
  282. .await
  283. {
  284. error!(
  285. "[IRC CLIENT] Failed inserting verified \
  286. signal event: {e}"
  287. );
  288. continue
  289. }
  290. // We sent this, so it should be considered seen.
  291. if let Err(e) = self.mark_seen(&event_id).await {
  292. error!("[IRC CLIENT] (multiplex_connection) self.mark_seen({event_id}) failed: {e}");
  293. return Err(e)
  294. }
  295. if let Err(e) =
  296. self.server.darkirc.p2p.broadcast(&EventPut(event, blob)).await
  297. {
  298. error!("[IRC CLIENT] Event broadcast was not admitted: {e}");
  299. }
  300. }
  301. }
  302. // If we got nothing, we just pass.
  303. Ok(None) => {}
  304. // If we got an error, we disconnect the client.
  305. Err(e) => {
  306. self.incoming.unsubscribe().await;
  307. self.incoming_st.unsubscribe().await;
  308. return Err(e)
  309. }
  310. }
  311. // Clear the line buffer
  312. line = String::new();
  313. }
  314. // Process message from the network. These should only be PRIVMSG.
  315. //
  316. // N.b. handling "historical messages", i.e. outstanding messages
  317. // which have occured when darkirc is offline are handled in
  318. // <file:./command.rs::async fn get_history(&self, channels: &HashSet<String>) -> Result<Vec<ReplyType>> {>
  319. // for which the logic for delivery should be kept in sync
  320. r = self.incoming.receive().fuse() => {
  321. // We will skip this if it's our own message.
  322. let event_id = r.header.id();
  323. if *self.last_sent.read().await == event_id {
  324. continue
  325. }
  326. // If this event was seen, skip it
  327. match self.is_seen(&event_id).await {
  328. Ok(true) => continue,
  329. Ok(false) => {},
  330. Err(e) => {
  331. error!("[IRC CLIENT] (multiplex_connection) self.is_seen({event_id}) failed: {e}");
  332. return Err(e)
  333. }
  334. }
  335. // Try to deserialize the `Event`'s content into a `Privmsg`
  336. let mut privmsg = match deserialize_async_partial(r.content()).await {
  337. Ok((v, _)) => v,
  338. Err(e) => {
  339. error!(target: "irc::client", "[IRC CLIENT] Failed deserializing event: {e}");
  340. continue
  341. }
  342. };
  343. // If successful, potentially decrypt it:
  344. self.server.try_decrypt(&mut privmsg, self.nickname.read().await.as_ref()).await;
  345. // We should skip any attempts to contact services from the network.
  346. if ["nickserv", "chanserv"].contains(&privmsg.nick.to_lowercase().as_str()) {
  347. continue
  348. }
  349. // If the privmsg is not intented for any of the given
  350. // channels or contacts, ignore it
  351. // otherwise add it as a reply and mark it as seen
  352. // in the seen_events tree.
  353. let channels = self.channels.read().await;
  354. let contacts = self.server.contacts.read().await;
  355. if !channels.contains(&privmsg.channel) &&
  356. !contacts.contains_key(&privmsg.channel)
  357. {
  358. continue
  359. }
  360. // Add the nickname to the list of nicks on the channel, if it's a channel.
  361. let mut chans_lock = self.server.channels.write().await;
  362. if let Some(chan) = chans_lock.get_mut(&privmsg.channel) {
  363. chan.nicks.insert(privmsg.nick.clone());
  364. }
  365. drop(chans_lock);
  366. // Handle message lines individually
  367. for line in privmsg.msg.lines() {
  368. // Skip empty lines
  369. if line.is_empty() {
  370. continue
  371. }
  372. // Format the message
  373. let msg = format!("PRIVMSG {} :{line}", privmsg.channel);
  374. // Send it to the client
  375. let reply = ReplyType::Client((privmsg.nick.clone(), msg));
  376. if let Err(e) = self.reply(&mut writer, &reply).await {
  377. error!("[IRC CLIENT] Failed writing PRIVMSG to client: {e}");
  378. continue
  379. }
  380. }
  381. // Mark the message as seen for this USER
  382. if let Err(e) = self.mark_seen(&event_id).await {
  383. error!("[IRC CLIENT] (multiplex_connection) self.mark_seen({}) failed: {}", event_id, e);
  384. return Err(e)
  385. }
  386. }
  387. // Process message from the network. These should only be RLN identities.
  388. r = self.incoming_st.receive().fuse() => {
  389. // We will skip this if it's our own message.
  390. let event_id = r.header.id();
  391. if *self.last_sent.read().await == event_id {
  392. continue
  393. }
  394. // If this event was seen, skip it
  395. match self.is_seen(&event_id).await {
  396. Ok(true) => continue,
  397. Ok(false) => {},
  398. Err(e) => {
  399. error!("[IRC CLIENT] (multiplex_connection) self.is_seen({}) failed: {}", event_id, e);
  400. return Err(e)
  401. }
  402. }
  403. // Static-event arrival path. EventGraph notifies
  404. // `static_pub` only after `commit_verified_static_event`
  405. // has durably stored the event/blob and applied the RLN
  406. // state change. So all we need to do is bookkeeping for
  407. // this client's seen-set.
  408. // Mark the message as seen for this USER
  409. if let Err(e) = self.mark_seen(&event_id).await {
  410. error!("[IRC CLIENT] (multiplex_connection) self.mark_seen({event_id}) failed: {e}");
  411. return Err(e)
  412. }
  413. }
  414. }
  415. }
  416. }
  417. /// Send a reply to the IRC client. Matches on the reply type.
  418. async fn reply<W>(&self, writer: &mut W, reply: &ReplyType) -> Result<()>
  419. where
  420. W: AsyncWrite + Unpin,
  421. {
  422. let r = match reply {
  423. ReplyType::Server((rpl, msg)) => format!(":{SERVER_NAME} {rpl:03} {msg}"),
  424. ReplyType::Client((nick, msg)) => format!(":{nick}!~anon@darkirc {msg}"),
  425. ReplyType::Pong(origin) => format!(":{SERVER_NAME} PONG :{origin}"),
  426. ReplyType::Cap(msg) => format!(":{SERVER_NAME} {msg}"),
  427. ReplyType::Notice((src, dst, msg)) => {
  428. format!(":{src}!~anon@darkirc NOTICE {dst} :{msg}")
  429. }
  430. };
  431. debug!("[{}] <-- {r}", self.addr);
  432. writer.write(r.as_bytes()).await?;
  433. writer.write(b"\r\n").await?;
  434. writer.flush().await?;
  435. Ok(())
  436. }
  437. /// Handle the incoming line given sent by the IRC client
  438. async fn process_client_line<W>(
  439. &self,
  440. line: &str,
  441. writer: &mut W,
  442. args_queue: &mut VecDeque<Privmsg>,
  443. ) -> Result<Option<Vec<Event>>>
  444. where
  445. W: AsyncWrite + Unpin,
  446. {
  447. if line.trim().is_empty() {
  448. // Silently ignore empty commands
  449. return Ok(None)
  450. }
  451. let mut line = line.to_string();
  452. // Remove CRLF
  453. if line.ends_with("\r\n") {
  454. line.pop();
  455. line.pop();
  456. } else if line.ends_with("\n") {
  457. line.pop();
  458. } else {
  459. return Err(Error::ParseFailed("Line doesn't end with CR/LF"))
  460. }
  461. // Prefix the message part of PRIVMSG with ':' if is not already.
  462. // Or realname part of USER command.
  463. if let Some(index) = match line.split_whitespace().next() {
  464. Some("PRIVMSG") => Some(2),
  465. Some("USER") => Some(4),
  466. _ => None,
  467. } {
  468. let mut words: Vec<String> =
  469. line.splitn(index + 1, char::is_whitespace).map(|s| s.to_string()).collect();
  470. if words.len() > index && !words[index].starts_with(':') {
  471. words[index] = format!(":{}", words[index]);
  472. }
  473. line = words.join(" ");
  474. }
  475. // Parse the line
  476. let mut tokens = line.split_ascii_whitespace();
  477. // Commands can begin with :garbage, but we will reject clients
  478. // doing that for now to keep the protocol simple and focused.
  479. let cmd = tokens.next().ok_or(Error::ParseFailed("Invalid command line"))?;
  480. let args = line.replacen(cmd, "", 1);
  481. let cmd = cmd.to_uppercase();
  482. debug!("[{}] --> {cmd}{args}", self.addr);
  483. // Handle the command. These implementations are in `command.rs`.
  484. let replies: Vec<ReplyType> = match cmd.as_str() {
  485. "ADMIN" => self.handle_cmd_admin(&args).await?,
  486. "CAP" => self.handle_cmd_cap(&args).await?,
  487. "INFO" => self.handle_cmd_info(&args).await?,
  488. "JOIN" => self.handle_cmd_join(&args, true).await?,
  489. "LIST" => self.handle_cmd_list(&args).await?,
  490. "MODE" => self.handle_cmd_mode(&args).await?,
  491. "MOTD" => self.handle_cmd_motd(&args).await?,
  492. "NAMES" => self.handle_cmd_names(&args).await?,
  493. "NICK" => self.handle_cmd_nick(&args).await?,
  494. "PART" => self.handle_cmd_part(&args).await?,
  495. "PASS" => self.handle_cmd_pass(&args).await?,
  496. "PING" => self.handle_cmd_ping(&args).await?,
  497. "PRIVMSG" => self.handle_cmd_privmsg(&args).await?,
  498. "REHASH" => self.handle_cmd_rehash(&args).await?,
  499. "TOPIC" => self.handle_cmd_topic(&args).await?,
  500. "USER" => self.handle_cmd_user(&args).await?,
  501. "VERSION" => self.handle_cmd_version(&args).await?,
  502. "QUIT" => return Err(Error::ChannelStopped),
  503. _ => {
  504. warn!("[IRC CLIENT] Unimplemented \"{cmd}\" command");
  505. vec![]
  506. }
  507. };
  508. // Depending on the reply type, we send according messages.
  509. for reply in replies.iter() {
  510. self.reply(writer, reply).await?;
  511. }
  512. // If the command was a PRIVMSG the client sent, we need to encrypt it and
  513. // create an Event to broadcast and return it from this function. So let's try.
  514. // We also do not allow sending unencrypted DMs. In that case, we send a notice
  515. // to the client to inform them that the feature is not enabled.
  516. // NOTE: This is not the most performant way to do this, probably not even
  517. // TODO: the best place to do it. Patches welcome. It's also a bit fragile
  518. // since we assume that `handle_cmd_privmsg()` won't return any replies.
  519. if cmd.as_str() == "PRIVMSG" && replies.is_empty() {
  520. // If the DAG is not synced yet, queue client lines
  521. // Once synced, send queued lines and continue as normal
  522. if !self.server.darkirc.event_graph.is_synced() {
  523. debug!("DAG is still syncing, queuing and skipping...");
  524. let Some(privmsg) = self.args_to_privmsg(args).await else {
  525. self.penalty.fetch_add(1, SeqCst);
  526. return Ok(None)
  527. };
  528. if !enqueue_pending_privmsg(args_queue, privmsg) {
  529. self.penalty.fetch_add(1, SeqCst);
  530. let nick = self.nickname.read().await.to_string();
  531. let reply = ReplyType::Notice((
  532. SERVER_NAME.to_string(),
  533. nick,
  534. "PRIVMSG queue is full; wait for sync before sending more".to_string(),
  535. ));
  536. self.reply(writer, &reply).await?;
  537. }
  538. return Ok(None)
  539. }
  540. // Check if we have queued PRIVMSGs, if we do send all of them first.
  541. let mut pending_events = vec![];
  542. if !args_queue.is_empty() {
  543. for _ in 0..args_queue.len() {
  544. let privmsg = args_queue.pop_front().unwrap();
  545. pending_events.push(self.privmsg_to_event(privmsg).await?);
  546. }
  547. return Ok(Some(pending_events))
  548. }
  549. // If queue is empty, create an event and return it
  550. let Some(privmsg) = self.args_to_privmsg(args).await else {
  551. self.penalty.fetch_add(1, SeqCst);
  552. return Ok(None)
  553. };
  554. let event = self.privmsg_to_event(privmsg).await?;
  555. return Ok(Some(vec![event]))
  556. }
  557. Ok(None)
  558. }
  559. // Internal helper function that creates a PRIVMSG from IRC client arguments
  560. async fn args_to_privmsg(&self, args: String) -> Option<Privmsg> {
  561. let nick = self.nickname.read().await.to_string();
  562. let channel = args.split_ascii_whitespace().next()?.to_string();
  563. let msg_offset = args.find(':')? + 1;
  564. let (_, msg) = args.split_at(msg_offset);
  565. // Truncate messages longer than MAX_MSG_LEN
  566. let msg = if msg.len() > MAX_MSG_LEN { msg.split_at(MAX_MSG_LEN).0 } else { msg };
  567. Some(Privmsg { version: 0, msg_type: 0, channel, nick, msg: msg.to_string() })
  568. }
  569. // Internal helper function that creates an Event from PRIVMSG arguments
  570. async fn privmsg_to_event(&self, mut privmsg: Privmsg) -> Result<Event> {
  571. // Encrypt the Privmsg if an encryption method is available.
  572. self.server.try_encrypt(&mut privmsg).await;
  573. // Build a DAG event and return it.
  574. Event::new(serialize_async(&privmsg).await, &self.server.darkirc.event_graph).await
  575. }
  576. /// Atomically mark a message as seen for this client.
  577. pub async fn mark_seen(&self, event_id: &blake3::Hash) -> Result<()> {
  578. let db = self
  579. .seen
  580. .get_or_init(|| async {
  581. let u = self.username.read().await.to_string();
  582. self.server.darkirc.sled.open_tree(format!("darkirc_user_{u}")).unwrap()
  583. })
  584. .await;
  585. debug!("Marking event {event_id} as seen");
  586. let mut batch = sled::Batch::default();
  587. batch.insert(event_id.as_bytes(), &[]);
  588. Ok(db.apply_batch(batch)?)
  589. }
  590. /// Check if a message was already marked seen for this client.
  591. pub async fn is_seen(&self, event_id: &blake3::Hash) -> Result<bool> {
  592. let db = self
  593. .seen
  594. .get_or_init(|| async {
  595. let u = self.username.read().await.to_string();
  596. self.server.darkirc.sled.open_tree(format!("darkirc_user_{u}")).unwrap()
  597. })
  598. .await;
  599. Ok(db.contains_key(event_id.as_bytes())?)
  600. }
  601. }
  602. #[cfg(test)]
  603. mod tests {
  604. use std::collections::VecDeque;
  605. use smol::io::{BufReader, Cursor};
  606. use super::{
  607. enqueue_pending_privmsg, read_bounded_line, MAX_IRC_LINE_LEN, MAX_PENDING_PRIVMSGS,
  608. };
  609. use crate::Privmsg;
  610. #[test]
  611. fn read_bounded_line_accepts_line_within_limit() {
  612. smol::block_on(async {
  613. let input = format!("{}\n", "a".repeat(MAX_IRC_LINE_LEN - 1));
  614. let mut reader = BufReader::new(Cursor::new(input.into_bytes()));
  615. let mut line = String::new();
  616. let read = read_bounded_line(&mut reader, &mut line).await.unwrap();
  617. assert_eq!(read, MAX_IRC_LINE_LEN);
  618. assert!(line.ends_with('\n'));
  619. });
  620. }
  621. #[test]
  622. fn read_bounded_line_rejects_oversized_line() {
  623. smol::block_on(async {
  624. let input = format!("{}\n", "a".repeat(MAX_IRC_LINE_LEN));
  625. let mut reader = BufReader::new(Cursor::new(input.into_bytes()));
  626. let mut line = String::new();
  627. assert!(read_bounded_line(&mut reader, &mut line).await.is_err());
  628. });
  629. }
  630. #[test]
  631. fn pending_privmsg_queue_has_fixed_capacity() {
  632. let mut queue = VecDeque::new();
  633. for _ in 0..MAX_PENDING_PRIVMSGS {
  634. assert!(enqueue_pending_privmsg(&mut queue, privmsg()));
  635. }
  636. assert!(!enqueue_pending_privmsg(&mut queue, privmsg()));
  637. assert_eq!(queue.len(), MAX_PENDING_PRIVMSGS);
  638. }
  639. fn privmsg() -> Privmsg {
  640. Privmsg {
  641. version: 0,
  642. msg_type: 0,
  643. channel: "#chan".to_string(),
  644. nick: "nick".to_string(),
  645. msg: "msg".to_string(),
  646. }
  647. }
  648. }