// IRC module. // // Maintains a persistent connection to an IRC server. Parses incoming // messages into notifications, supports sending messages and runtime // commands (join, leave, etc.). Config changes persist to daemon.toml. // // Runs as a spawned local task on the daemon's LocalSet. Notifications // flow through an mpsc channel into the main state. Reconnects // automatically with exponential backoff. use crate::config::{Config, IrcConfig}; use crate::notify::Notification; use crate::{home, now}; use std::cell::RefCell; use std::collections::VecDeque; use std::io; use std::rc::Rc; use std::sync::Arc; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::sync::mpsc; use tracing::{error, info, warn}; const MAX_LOG_LINES: usize = 200; const RECONNECT_BASE_SECS: u64 = 5; const RECONNECT_MAX_SECS: u64 = 300; const PING_INTERVAL_SECS: u64 = 120; const PING_TIMEOUT_SECS: u64 = 30; /// Parsed IRC message. struct IrcMessage { prefix: Option, // nick!user@host command: String, params: Vec, } impl IrcMessage { fn parse(line: &str) -> Option { let line = line.trim_end_matches(|c| c == '\r' || c == '\n'); if line.is_empty() { return None; } let (prefix, rest) = if line.starts_with(':') { let space = line.find(' ')?; (Some(line[1..space].to_string()), &line[space + 1..]) } else { (None, line) }; let (command_params, trailing) = if let Some(pos) = rest.find(" :") { (&rest[..pos], Some(rest[pos + 2..].to_string())) } else { (rest, None) }; let mut parts: Vec = command_params .split_whitespace() .map(String::from) .collect(); if parts.is_empty() { return None; } let command = parts.remove(0).to_uppercase(); let mut params = parts; if let Some(t) = trailing { params.push(t); } Some(IrcMessage { prefix, command, params, }) } /// Extract nick from prefix (nick!user@host → nick). fn nick(&self) -> Option<&str> { self.prefix .as_deref() .and_then(|p| p.split('!').next()) } } /// Shared IRC state, accessible from both the read task and RPC handlers. pub struct IrcState { pub config: IrcConfig, pub connected: bool, pub channels: Vec, pub log: VecDeque, writer: Option, } /// Type-erased writer handle so we can store it without generic params. type WriterHandle = Box; trait AsyncWriter { fn write_line(&mut self, line: &str) -> std::pin::Pin> + '_>>; } /// Writer over a TLS stream. struct TlsWriter { inner: tokio::io::WriteHalf>, } impl AsyncWriter for TlsWriter { fn write_line(&mut self, line: &str) -> std::pin::Pin> + '_>> { let data = format!("{line}\r\n"); Box::pin(async move { self.inner.write_all(data.as_bytes()).await }) } } /// Writer over a plain TCP stream. struct PlainWriter { inner: tokio::io::WriteHalf, } impl AsyncWriter for PlainWriter { fn write_line(&mut self, line: &str) -> std::pin::Pin> + '_>> { let data = format!("{line}\r\n"); Box::pin(async move { self.inner.write_all(data.as_bytes()).await }) } } impl IrcState { fn new(config: IrcConfig) -> Self { Self { channels: config.channels.clone(), config, connected: false, log: VecDeque::with_capacity(MAX_LOG_LINES), writer: None, } } fn push_log(&mut self, line: &str) { if self.log.len() >= MAX_LOG_LINES { self.log.pop_front(); } self.log.push_back(line.to_string()); } async fn send_raw(&mut self, line: &str) -> io::Result<()> { if let Some(ref mut w) = self.writer { w.write_line(line).await } else { Err(io::Error::new(io::ErrorKind::NotConnected, "not connected")) } } async fn send_privmsg(&mut self, target: &str, msg: &str) -> io::Result<()> { self.send_raw(&format!("PRIVMSG {target} :{msg}")).await } async fn join(&mut self, channel: &str) -> io::Result<()> { self.send_raw(&format!("JOIN {channel}")).await?; if !self.channels.iter().any(|c| c == channel) { self.channels.push(channel.to_string()); } Ok(()) } async fn part(&mut self, channel: &str) -> io::Result<()> { self.send_raw(&format!("PART {channel}")).await?; self.channels.retain(|c| c != channel); Ok(()) } } pub type SharedIrc = Rc>; /// Start the IRC module. Returns the shared state handle. pub fn start( config: IrcConfig, notify_tx: mpsc::UnboundedSender, daemon_config: Rc>, ) -> SharedIrc { let state = Rc::new(RefCell::new(IrcState::new(config))); let state_clone = state.clone(); tokio::task::spawn_local(async move { connection_loop(state_clone, notify_tx, daemon_config).await; }); state } async fn connection_loop( state: SharedIrc, notify_tx: mpsc::UnboundedSender, daemon_config: Rc>, ) { let mut backoff = RECONNECT_BASE_SECS; loop { let config = state.borrow().config.clone(); info!("irc: connecting to {}:{}", config.server, config.port); match connect_and_run(&state, &config, ¬ify_tx).await { Ok(()) => { info!("irc: connection closed cleanly"); } Err(e) => { error!("irc: connection error: {e}"); } } // Reset backoff if we had a working connection (registered // successfully before disconnecting) let was_connected = state.borrow().connected; state.borrow_mut().connected = false; state.borrow_mut().writer = None; if was_connected { backoff = RECONNECT_BASE_SECS; } // Persist current channel list to config { let channels = state.borrow().channels.clone(); let mut dc = daemon_config.borrow_mut(); dc.irc.channels = channels; dc.save(); } info!("irc: reconnecting in {backoff}s"); tokio::time::sleep(std::time::Duration::from_secs(backoff)).await; backoff = (backoff * 2).min(RECONNECT_MAX_SECS); } } async fn connect_and_run( state: &SharedIrc, config: &IrcConfig, notify_tx: &mpsc::UnboundedSender, ) -> io::Result<()> { let addr = format!("{}:{}", config.server, config.port); let tcp = tokio::net::TcpStream::connect(&addr).await?; if config.tls { let tls_config = rustls::ClientConfig::builder_with_provider( rustls::crypto::ring::default_provider().into(), ) .with_safe_default_protocol_versions() .map_err(|e| io::Error::new(io::ErrorKind::Other, e))? .with_root_certificates(root_certs()) .with_no_client_auth(); let connector = tokio_rustls::TlsConnector::from(Arc::new(tls_config)); let server_name = rustls::pki_types::ServerName::try_from(config.server.clone()) .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))?; let tls_stream = connector.connect(server_name, tcp).await?; let (reader, writer) = tokio::io::split(tls_stream); state.borrow_mut().writer = Some(Box::new(TlsWriter { inner: writer })); let buf_reader = BufReader::new(reader); register_and_read(state, config, buf_reader, notify_tx).await } else { let (reader, writer) = tokio::io::split(tcp); state.borrow_mut().writer = Some(Box::new(PlainWriter { inner: writer })); let buf_reader = BufReader::new(reader); register_and_read(state, config, buf_reader, notify_tx).await } } async fn register_and_read( state: &SharedIrc, config: &IrcConfig, mut reader: BufReader, notify_tx: &mpsc::UnboundedSender, ) -> io::Result<()> { // Register { let mut s = state.borrow_mut(); s.send_raw(&format!("NICK {}", config.nick)).await?; s.send_raw(&format!("USER {} 0 * :{}", config.user, config.realname)).await?; } let mut buf = Vec::new(); let mut ping_sent = false; let mut deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(PING_INTERVAL_SECS); loop { buf.clear(); let read_result = tokio::select! { result = reader.read_until(b'\n', &mut buf) => result, _ = tokio::time::sleep_until(deadline) => { if ping_sent { return Err(io::Error::new( io::ErrorKind::TimedOut, "ping timeout — no response from server", )); } info!("irc: no data for {}s, sending PING", PING_INTERVAL_SECS); state.borrow_mut().send_raw("PING :keepalive").await?; ping_sent = true; deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(PING_TIMEOUT_SECS); continue; } }; let n = read_result?; if n == 0 { break; } // Any data from server resets the ping timer ping_sent = false; deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(PING_INTERVAL_SECS); // IRC is not guaranteed UTF-8 — lossy conversion handles Latin-1 etc. let line = String::from_utf8_lossy(&buf).trim_end().to_string(); if line.is_empty() { continue; } let msg = match IrcMessage::parse(&line) { Some(m) => m, None => continue, }; match msg.command.as_str() { "PING" => { let arg = msg.params.first().map(|s| s.as_str()).unwrap_or(""); state.borrow_mut().send_raw(&format!("PONG :{arg}")).await?; } // RPL_WELCOME — registration complete "001" => { info!("irc: registered as {}", config.nick); state.borrow_mut().connected = true; // Join configured channels let channels = state.borrow().channels.clone(); for ch in &channels { if let Err(e) = state.borrow_mut().send_raw(&format!("JOIN {ch}")).await { warn!("irc: failed to join {ch}: {e}"); } } } "PRIVMSG" => { let target = msg.params.first().map(|s| s.as_str()).unwrap_or(""); let text = msg.params.get(1).map(|s| s.as_str()).unwrap_or(""); let nick = msg.nick().unwrap_or("unknown"); // Handle CTCP requests (wrapped in \x01) if text.starts_with('\x01') && text.ends_with('\x01') { let ctcp = &text[1..text.len()-1]; if ctcp.starts_with("VERSION") { let reply = format!( "NOTICE {nick} :\x01VERSION poc-daemon 0.4.0\x01" ); state.borrow_mut().send_raw(&reply).await.ok(); } // Don't generate notifications for CTCP continue; } // Log the message let log_line = if target.starts_with('#') { format!("[{}] <{}> {}", target, nick, text) } else { format!("[PM:{nick}] {text}") }; state.borrow_mut().push_log(&log_line); // Write to per-channel/per-user log file if target.starts_with('#') { append_log(target, nick, text); } else { append_log(&format!("pm-{nick}"), nick, text); } // Generate notification let (ntype, urgency) = classify_privmsg( nick, target, text, &config.nick, ); let _ = notify_tx.send(Notification { ntype, urgency, message: log_line, timestamp: now(), }); } // Nick in use "433" => { let alt = format!("{}_", config.nick); warn!("irc: nick in use, trying {alt}"); state.borrow_mut().send_raw(&format!("NICK {alt}")).await?; } "JOIN" | "PART" | "QUIT" | "KICK" | "MODE" | "TOPIC" | "NOTICE" => { // Could log these, but skip for now } _ => {} } } Ok(()) } /// Classify a PRIVMSG into notification type and urgency. fn classify_privmsg(nick: &str, target: &str, text: &str, my_nick: &str) -> (String, u8) { let my_nick_lower = my_nick.to_lowercase(); let text_lower = text.to_lowercase(); if !target.starts_with('#') { // Private message (format!("irc.pm.{nick}"), crate::notify::URGENT) } else if text_lower.contains(&my_nick_lower) { // Mentioned in channel (format!("irc.mention.{nick}"), crate::notify::NORMAL) } else { // Regular channel message let channel = target.trim_start_matches('#'); (format!("irc.channel.{channel}"), crate::notify::AMBIENT) } } /// Append a message to the per-channel or per-user log file. /// Logs go to ~/.claude/irc/logs/{target}.log (e.g. #bcachefs.log, pm-kent.log) fn append_log(target: &str, nick: &str, text: &str) { use std::io::Write; // Sanitize target for filename (strip leading #, lowercase) let filename = format!("{}.log", target.trim_start_matches('#').to_lowercase()); let dir = home().join(".claude/irc/logs"); let _ = std::fs::create_dir_all(&dir); if let Ok(mut f) = std::fs::OpenOptions::new() .create(true) .append(true) .open(dir.join(&filename)) { let secs = now() as u64; let _ = writeln!(f, "{secs} <{nick}> {text}"); } } fn root_certs() -> rustls::RootCertStore { let mut roots = rustls::RootCertStore::empty(); roots.extend(webpki_roots::TLS_SERVER_ROOTS.iter().cloned()); roots } /// Handle a runtime command from RPC. pub async fn handle_command( state: &SharedIrc, daemon_config: &Rc>, cmd: &str, args: &[String], ) -> Result { match cmd { "join" => { let channel = args.first().ok_or("usage: irc join ")?; let channel = if channel.starts_with('#') { channel.clone() } else { format!("#{channel}") }; state .borrow_mut() .join(&channel) .await .map_err(|e| e.to_string())?; // Persist let mut dc = daemon_config.borrow_mut(); if !dc.irc.channels.contains(&channel) { dc.irc.channels.push(channel.clone()); } dc.save(); Ok(format!("joined {channel}")) } "leave" | "part" => { let channel = args.first().ok_or("usage: irc leave ")?; let channel = if channel.starts_with('#') { channel.clone() } else { format!("#{channel}") }; state .borrow_mut() .part(&channel) .await .map_err(|e| e.to_string())?; // Persist let mut dc = daemon_config.borrow_mut(); dc.irc.channels.retain(|c| c != &channel); dc.save(); Ok(format!("left {channel}")) } "send" | "msg" => { if args.len() < 2 { return Err("usage: irc send ".into()); } let target = &args[0]; if target.starts_with('#') { let s = state.borrow(); if !s.channels.iter().any(|c| c == target) { return Err(format!( "not in channel {target} (joined: {})", s.channels.join(", ") )); } } let msg = args[1..].join(" "); let nick = state.borrow().config.nick.clone(); state .borrow_mut() .send_privmsg(target, &msg) .await .map_err(|e| e.to_string())?; append_log(target, &nick, &msg); Ok(format!("sent to {target}")) } "status" => { let s = state.borrow(); Ok(format!( "connected={} channels={} log_lines={} nick={}", s.connected, s.channels.join(","), s.log.len(), s.config.nick, )) } "log" => { let n: usize = args .first() .and_then(|s| s.parse().ok()) .unwrap_or(15); let s = state.borrow(); let lines: Vec<&String> = s.log.iter().rev().take(n).collect(); let mut lines: Vec<&str> = lines.iter().map(|s| s.as_str()).collect(); lines.reverse(); Ok(lines.join("\n")) } "nick" => { let new_nick = args.first().ok_or("usage: irc nick ")?; state .borrow_mut() .send_raw(&format!("NICK {new_nick}")) .await .map_err(|e| e.to_string())?; let mut dc = daemon_config.borrow_mut(); dc.irc.nick = new_nick.clone(); dc.save(); Ok(format!("nick → {new_nick}")) } _ => Err(format!( "unknown irc command: {cmd}\n\ commands: join, leave, send, status, log, nick" )), } }