From e3bb0f6222b3f21ab7fb99d68b07cda9d6702efa Mon Sep 17 00:00:00 2001 From: Christopher Nelson Date: Sat, 18 Jul 2026 18:39:32 -0400 Subject: [PATCH 1/4] feat(discord): add rich presence support --- Cargo.lock | 27 +- Cargo.toml | 1 + src/app/discord/imp.rs | 168 +++++ src/app/discord/ipc.rs | 507 ++++++++++++++ src/app/discord/mod.rs | 48 ++ src/app/discord/worker.rs | 1331 +++++++++++++++++++++++++++++++++++++ src/app/imp.rs | 33 + src/app/ipc/event.rs | 14 + src/app/ipc/request.rs | 215 +++++- src/app/ipc/response.rs | 45 ++ src/app/mod.rs | 1 + 11 files changed, 2388 insertions(+), 2 deletions(-) create mode 100644 src/app/discord/imp.rs create mode 100644 src/app/discord/ipc.rs create mode 100644 src/app/discord/mod.rs create mode 100644 src/app/discord/worker.rs diff --git a/Cargo.lock b/Cargo.lock index c9e14ca..880825d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -612,6 +612,21 @@ dependencies = [ "windows-sys", ] +[[package]] +name = "discord-rich-presence" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "90c55d69cab17c19677ce3a5f8face993a9e6eaf847fecac3547f3a3ff4a2494" +dependencies = [ + "log", + "serde", + "serde_derive", + "serde_json", + "serde_repr", + "thiserror", + "uuid 0.8.2", +] + [[package]] name = "displaydoc" version = "0.2.6" @@ -2594,6 +2609,7 @@ dependencies = [ "chrono", "clap", "dirs", + "discord-rich-presence", "epoxy", "flume", "gdk4-wayland", @@ -2923,6 +2939,15 @@ version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" +[[package]] +name = "uuid" +version = "0.8.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bc5cf98d8186244414c848017f0e2676b3fcb46807f6668a97dfe67359a3c4b7" +dependencies = [ + "getrandom 0.2.17", +] + [[package]] name = "uuid" version = "1.23.4" @@ -3304,7 +3329,7 @@ dependencies = [ "tokio", "tracing", "uds_windows", - "uuid", + "uuid 1.23.4", "windows-sys", "winnow", "zbus_macros", diff --git a/Cargo.toml b/Cargo.toml index 7488da1..ef22fb8 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -19,6 +19,7 @@ webkit = { package = "webkit6", version = "0.6.1", features = ["v2_52"] } ashpd = { version = "0.13.12", features = ["background", "gtk4", "inhibit", "open_uri"] } ksni = "0.3.5" mpris-server = "0.10.0" +discord-rich-presence = "1.1" clap = { version = "4.6.1", features = ["derive"] } gettext-rs = { version = "0.7.7", features = ["gettext-system"] } tokio = { version = "1.52.3", features = ["full"] } diff --git a/src/app/discord/imp.rs b/src/app/discord/imp.rs new file mode 100644 index 0000000..1273cf9 --- /dev/null +++ b/src/app/discord/imp.rs @@ -0,0 +1,168 @@ +use std::{ + cell::RefCell, + rc::Rc, + sync::mpsc::{self, Sender}, + thread::JoinHandle, +}; + +use gtk::glib::{self, subclass::prelude::*}; +use tracing::error; + +use super::worker::{self, DiscordCommand, RichPresenceClient}; +use crate::{app::ipc::event::DiscordActivity, spawn_local}; + +type StatusCallback = Box; + +struct WorkerHandle { + commands: Sender, + _join: JoinHandle<()>, +} + +#[derive(Default)] +pub struct Discord { + worker: RefCell>, + status_callback: Rc>>, +} + +#[glib::object_subclass] +impl ObjectSubclass for Discord { + const NAME: &'static str = "Discord"; + type Type = super::Discord; + type ParentType = glib::Object; +} + +impl Discord { + pub fn start(&self) { + if self.worker.borrow().is_some() { + return; + } + + let (commands, receiver) = mpsc::channel::(); + let (status_sender, status_receiver) = flume::unbounded::(); + + let join = std::thread::spawn(move || { + worker::run( + receiver, + move |connected| { + status_sender.send(connected).ok(); + }, + RichPresenceClient::new, + ); + }); + + let status_callback = self.status_callback.clone(); + spawn_local!(async move { + while let Ok(connected) = status_receiver.recv_async().await { + if let Some(callback) = &*status_callback.borrow() { + callback(connected); + } + } + }); + + *self.worker.borrow_mut() = Some(WorkerHandle { + commands, + _join: join, + }); + } + + pub fn stop(&self) { + // Dropping the command sender ends the worker's command loop, which + // closes the Discord connection and lets the thread terminate on + // its own without blocking the GTK main thread. + let handle = self.worker.borrow_mut().take(); + drop(handle); + } + + pub fn connect(&self) { + if let Err(e) = self.send_command(DiscordCommand::Connect) { + error!("Failed to queue Discord connect: {e}"); + self.emit_status(false); + } + } + + pub fn disconnect(&self) { + if let Err(e) = self.send_command(DiscordCommand::Disconnect) { + error!("Failed to queue Discord disconnect: {e}"); + self.emit_status(false); + } + } + + pub fn set_activity(&self, activity: DiscordActivity) { + if let Err(e) = self.send_command(DiscordCommand::SetActivity(activity)) { + error!("Failed to queue Discord set activity: {e}"); + self.emit_status(false); + } + } + + pub fn clear_activity(&self) { + if let Err(e) = self.send_command(DiscordCommand::ClearActivity) { + error!("Failed to queue Discord clear activity: {e}"); + self.emit_status(false); + } + } + + pub fn set_status_callback(&self, callback: F) { + self.status_callback + .borrow_mut() + .replace(Box::new(callback)); + } + + fn send_command(&self, command: DiscordCommand) -> Result<(), String> { + let worker = self.worker.borrow(); + let worker = worker.as_ref().ok_or("Discord service not started")?; + + worker + .commands + .send(command) + .map_err(|e| format!("Discord worker unavailable: {e}")) + } + + fn emit_status(&self, connected: bool) { + if let Some(callback) = &*self.status_callback.borrow() { + callback(connected); + } + } +} + +impl ObjectImpl for Discord {} + +#[cfg(test)] +mod tests { + use std::{cell::RefCell, rc::Rc}; + + use crate::app::ipc::event::DiscordActivity; + + fn activity() -> DiscordActivity { + DiscordActivity { + state: "Watching".to_owned(), + details: "Movie".to_owned(), + image: None, + start_timestamp: None, + end_timestamp: None, + } + } + + #[test] + fn set_activity_with_unavailable_worker_reports_disconnected() { + let discord = super::super::Discord::default(); + let statuses = Rc::new(RefCell::new(Vec::new())); + let collected = Rc::clone(&statuses); + discord.connect_status(move |connected| collected.borrow_mut().push(connected)); + + discord.set_activity(activity()); + + assert_eq!(*statuses.borrow(), [false]); + } + + #[test] + fn clear_activity_with_unavailable_worker_reports_disconnected() { + let discord = super::super::Discord::default(); + let statuses = Rc::new(RefCell::new(Vec::new())); + let collected = Rc::clone(&statuses); + discord.connect_status(move |connected| collected.borrow_mut().push(connected)); + + discord.clear_activity(); + + assert_eq!(*statuses.borrow(), [false]); + } +} diff --git a/src/app/discord/ipc.rs b/src/app/discord/ipc.rs new file mode 100644 index 0000000..67fd61d --- /dev/null +++ b/src/app/discord/ipc.rs @@ -0,0 +1,507 @@ +use std::{ + collections::HashSet, + env, + fs::Metadata, + io::{self, Read, Write}, + net::Shutdown, + os::unix::{fs::MetadataExt, net::UnixStream}, + path::{Path, PathBuf}, + sync::atomic::{AtomicU64, Ordering}, + time::{Duration, Instant}, +}; + +use serde_json::{Value, json}; +use tracing::warn; + +const ENV_KEYS: [&str; 4] = ["XDG_RUNTIME_DIR", "TMPDIR", "TMP", "TEMP"]; +const APP_SUBPATHS: [&str; 7] = [ + "", + "app/com.discordapp.Discord/", + "app/dev.vencord.Vesktop/", + ".flatpak/com.discordapp.Discord/xdg-run/", + ".flatpak/dev.vencord.Vesktop/xdg-run/", + "snap.discord-canary/", + "snap.discord/", +]; +const MAX_FRAME_LENGTH: usize = 1024 * 1024; +const MAX_PENDING_NONCES: usize = 64; +const HANDSHAKE_READ_SLICE: Duration = Duration::from_millis(100); +const LIVENESS_READ_TIMEOUT: Duration = Duration::from_millis(10); +const WRITE_TIMEOUT: Duration = Duration::from_secs(1); + +static NEXT_NONCE: AtomicU64 = AtomicU64::new(1); + +pub struct DiscordIpcTransport { + client_id: String, + connection: Option, + configured_path: Option, + handshake_timeout: Duration, +} + +struct Connection { + socket: UnixStream, + path: PathBuf, + identity: SocketIdentity, + received: Vec, + pending_nonces: HashSet, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +struct SocketIdentity { + device: u64, + inode: u64, +} + +impl SocketIdentity { + fn from_metadata(metadata: &Metadata) -> Self { + Self { + device: metadata.dev(), + inode: metadata.ino(), + } + } +} + +impl DiscordIpcTransport { + pub fn new(client_id: &str, handshake_timeout: Duration) -> Self { + Self { + client_id: client_id.to_owned(), + connection: None, + configured_path: None, + handshake_timeout, + } + } + + #[cfg(test)] + pub fn with_path(client_id: &str, path: PathBuf, handshake_timeout: Duration) -> Self { + let mut transport = Self::new(client_id, handshake_timeout); + transport.configured_path = Some(path); + transport + } + + pub fn connect(&mut self) -> Result<(), String> { + self.connection = None; + + let paths = self + .configured_path + .clone() + .map_or_else(find_pipes, |path| vec![path]); + if paths.is_empty() { + return Err("failed to find Discord IPC socket".to_owned()); + } + + let mut last_error = None; + for path in paths { + match Connection::connect(path.clone()) { + Ok(mut connection) => { + if let Err(error) = + connection.handshake(&self.client_id, self.handshake_timeout) + { + last_error = Some(format!("{}: {error}", path.display())); + continue; + } + self.connection = Some(connection); + return Ok(()); + } + Err(error) => last_error = Some(format!("{}: {error}", path.display())), + } + } + + Err(last_error.unwrap_or_else(|| "failed to connect to Discord IPC".to_owned())) + } + + pub fn set_activity(&mut self, activity: Value) -> Result<(), String> { + self.send_command(Some(activity)) + } + + pub fn clear_activity(&mut self) -> Result<(), String> { + self.send_command(None) + } + + fn send_command(&mut self, activity: Option) -> Result<(), String> { + let connection = self + .connection + .as_mut() + .ok_or_else(|| "Discord IPC is not connected".to_owned())?; + let nonce = format!( + "{}-{}", + std::process::id(), + NEXT_NONCE.fetch_add(1, Ordering::Relaxed) + ); + let payload = json!({ + "cmd": "SET_ACTIVITY", + "args": { + "pid": std::process::id(), + "activity": activity, + }, + "nonce": nonce, + }); + + connection.send(1, &payload)?; + connection.track_nonce(nonce); + Ok(()) + } + + pub fn poll_liveness(&mut self) -> Result<(), String> { + self.connection + .as_mut() + .ok_or_else(|| "Discord IPC is not connected".to_owned())? + .poll_liveness() + } + + pub fn close(&mut self) -> Result<(), String> { + let Some(mut connection) = self.connection.take() else { + return Ok(()); + }; + + let _ = connection.send(2, &json!({})); + match connection.socket.shutdown(Shutdown::Both) { + Ok(()) => Ok(()), + Err(error) if is_closed_socket(&error) => Ok(()), + Err(error) => Err(format!("Discord IPC socket shutdown failed: {error}")), + } + } +} + +impl Connection { + /// Tracks responses for draining and correlation, not synchronous command success. + fn track_nonce(&mut self, nonce: String) { + if self.pending_nonces.len() >= MAX_PENDING_NONCES { + self.pending_nonces.clear(); + } + self.pending_nonces.insert(nonce); + } + + fn connect(path: PathBuf) -> Result { + let socket = UnixStream::connect(&path) + .map_err(|error| format!("failed to connect to socket: {error}"))?; + socket + .set_write_timeout(Some(WRITE_TIMEOUT)) + .map_err(|error| format!("failed to set Discord IPC write timeout: {error}"))?; + let metadata = path + .metadata() + .map_err(|error| format!("failed to inspect connected socket: {error}"))?; + + Ok(Self { + socket, + path, + identity: SocketIdentity::from_metadata(&metadata), + received: Vec::new(), + pending_nonces: HashSet::new(), + }) + } + + fn handshake(&mut self, client_id: &str, timeout: Duration) -> Result<(), String> { + self.send(0, &json!({ "v": 1, "client_id": client_id }))?; + let deadline = Instant::now() + timeout; + + loop { + let now = Instant::now(); + if now >= deadline { + return Err(format!( + "Discord IPC handshake timed out after {} ms", + timeout.as_millis() + )); + } + + self.socket + .set_read_timeout(Some(HANDSHAKE_READ_SLICE.min(deadline - now))) + .map_err(|error| format!("failed to set handshake read timeout: {error}"))?; + + match self.read_once() { + Ok(frames) => { + for (opcode, payload) in frames { + if opcode == 2 { + return Err("Discord closed IPC during handshake".to_owned()); + } + if opcode == 1 + && payload.get("evt").and_then(Value::as_str) == Some("READY") + { + self.socket + .set_read_timeout(Some(LIVENESS_READ_TIMEOUT)) + .map_err(|error| { + format!("failed to set liveness read timeout: {error}") + })?; + return Ok(()); + } + } + } + Err(error) if is_transient_read_error(&error) => {} + Err(error) => return Err(format!("Discord IPC handshake read failed: {error}")), + } + } + } + + fn poll_liveness(&mut self) -> Result<(), String> { + let metadata = self + .path + .metadata() + .map_err(|error| format!("Discord IPC socket disappeared: {error}"))?; + if SocketIdentity::from_metadata(&metadata) != self.identity { + return Err("Discord IPC socket was replaced".to_owned()); + } + + match self.read_once() { + Ok(frames) => self.process_frames(frames), + Err(error) if is_transient_read_error(&error) => Ok(()), + Err(error) => Err(format!("Discord IPC read failed: {error}")), + } + } + + fn process_frames(&mut self, frames: Vec<(u32, Value)>) -> Result<(), String> { + for (opcode, payload) in frames { + if opcode == 2 { + return Err("Discord closed the IPC connection".to_owned()); + } + if let Some(nonce) = payload.get("nonce").and_then(Value::as_str) { + self.pending_nonces.remove(nonce); + } + } + Ok(()) + } + + fn send(&mut self, opcode: u32, payload: &Value) -> Result<(), String> { + let data = serde_json::to_vec(payload) + .map_err(|error| format!("failed to serialize Discord IPC payload: {error}"))?; + let length = + u32::try_from(data.len()).map_err(|_| "Discord IPC payload is too large".to_owned())?; + let mut header = [0_u8; 8]; + header[..4].copy_from_slice(&opcode.to_le_bytes()); + header[4..].copy_from_slice(&length.to_le_bytes()); + + self.socket + .write_all(&header) + .and_then(|()| self.socket.write_all(&data)) + .map_err(|error| format!("Discord IPC write failed: {error}")) + } + + fn read_once(&mut self) -> io::Result> { + let mut buffer = [0_u8; 8192]; + match self.socket.read(&mut buffer) { + Ok(0) => { + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + "Discord IPC EOF", + )); + } + Ok(length) => self.received.extend_from_slice(&buffer[..length]), + Err(error) => return Err(error), + } + + let mut frames = Vec::new(); + loop { + if self.received.len() < 8 { + break; + } + let opcode = u32::from_le_bytes( + self.received[..4] + .try_into() + .expect("four-byte opcode slice"), + ); + let length = u32::from_le_bytes( + self.received[4..8] + .try_into() + .expect("four-byte length slice"), + ) as usize; + if length > MAX_FRAME_LENGTH { + warn!("Discord IPC frame exceeds size limit"); + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "Discord IPC frame exceeds size limit", + )); + } + if self.received.len() < 8 + length { + break; + } + + let payload = + serde_json::from_slice(&self.received[8..8 + length]).map_err(|error| { + warn!("Discord IPC frame contains malformed JSON"); + io::Error::new( + io::ErrorKind::InvalidData, + format!("invalid Discord IPC JSON: {error}"), + ) + })?; + self.received.drain(..8 + length); + frames.push((opcode, payload)); + } + Ok(frames) + } +} + +fn is_transient_read_error(error: &io::Error) -> bool { + matches!( + error.kind(), + io::ErrorKind::Interrupted | io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut + ) +} + +fn is_closed_socket(error: &io::Error) -> bool { + error.kind() == io::ErrorKind::NotConnected +} + +fn find_pipes() -> Vec { + let snap = env::var_os("SNAP").is_some(); + let mut paths = Vec::new(); + + for key in ENV_KEYS { + let Some(value) = env::var_os(key) else { + continue; + }; + let mut base = PathBuf::from(value); + if snap && key == "XDG_RUNTIME_DIR" { + base.pop(); + } + if !base.is_dir() { + continue; + } + + for index in 0..10 { + for subpath in APP_SUBPATHS { + let path = base.join(subpath).join(format!("discord-ipc-{index}")); + if Path::new(&path).exists() { + paths.push(path); + } + } + } + } + + paths +} + +#[cfg(test)] +mod tests { + use std::os::unix::net::UnixStream; + + use super::*; + + fn connection() -> (Connection, UnixStream) { + let (socket, peer) = UnixStream::pair().expect("test socket pair should be created"); + ( + Connection { + socket, + path: PathBuf::new(), + identity: SocketIdentity { + device: 0, + inode: 0, + }, + received: Vec::new(), + pending_nonces: HashSet::new(), + }, + peer, + ) + } + + #[test] + fn interrupted_reads_are_transient() { + assert!(is_transient_read_error(&io::Error::from( + io::ErrorKind::Interrupted + ))); + } + + #[test] + fn not_connected_shutdown_is_success() { + assert!(is_closed_socket(&io::Error::from( + io::ErrorKind::NotConnected + ))); + assert!(!is_closed_socket(&io::Error::from( + io::ErrorKind::PermissionDenied + ))); + } + + #[test] + fn pending_nonces_never_exceed_the_bound() { + let (mut connection, _peer) = connection(); + + for nonce in 0..(MAX_PENDING_NONCES * 3) { + connection.track_nonce(nonce.to_string()); + assert!(connection.pending_nonces.len() <= MAX_PENDING_NONCES); + } + } + + #[test] + fn matching_response_removes_pending_nonce() { + let (mut connection, mut peer) = connection(); + connection.track_nonce("expected".to_owned()); + let payload = serde_json::to_vec(&json!({ "nonce": "expected" })).unwrap(); + peer.write_all(&1_u32.to_le_bytes()).unwrap(); + peer.write_all(&(payload.len() as u32).to_le_bytes()) + .unwrap(); + peer.write_all(&payload).unwrap(); + + let frames = connection.read_once().expect("response should be readable"); + connection.process_frames(frames).unwrap(); + + assert!(connection.pending_nonces.is_empty()); + } + + #[test] + fn unknown_response_nonce_is_harmless() { + let (mut connection, mut peer) = connection(); + connection.track_nonce("expected".to_owned()); + let payload = serde_json::to_vec(&json!({ "nonce": "unknown" })).unwrap(); + peer.write_all(&1_u32.to_le_bytes()).unwrap(); + peer.write_all(&(payload.len() as u32).to_le_bytes()) + .unwrap(); + peer.write_all(&payload).unwrap(); + + let frames = connection.read_once().expect("response should be readable"); + connection.process_frames(frames).unwrap(); + + assert!(connection.pending_nonces.contains("expected")); + } + + #[test] + fn close_opcode_fails_liveness_frame_processing() { + let (mut connection, _peer) = connection(); + + let error = connection + .process_frames(vec![(2, json!({}))]) + .expect_err("close opcode should fail liveness"); + + assert!(error.contains("closed")); + } + + #[test] + fn oversized_frame_is_rejected() { + let (mut connection, mut peer) = connection(); + peer.write_all(&1_u32.to_le_bytes()).unwrap(); + peer.write_all(&((MAX_FRAME_LENGTH + 1) as u32).to_le_bytes()) + .unwrap(); + + let error = connection + .read_once() + .expect_err("oversized frame should fail"); + + assert_eq!(error.kind(), io::ErrorKind::InvalidData); + } + + #[test] + fn malformed_json_frame_is_rejected() { + let (mut connection, mut peer) = connection(); + let payload = b"{"; + peer.write_all(&1_u32.to_le_bytes()).unwrap(); + peer.write_all(&(payload.len() as u32).to_le_bytes()) + .unwrap(); + peer.write_all(payload).unwrap(); + + let error = connection + .read_once() + .expect_err("malformed JSON should fail"); + + assert_eq!(error.kind(), io::ErrorKind::InvalidData); + } + + #[test] + fn partial_header_followed_by_eof_is_rejected() { + let (mut connection, mut peer) = connection(); + peer.write_all(&1_u32.to_le_bytes()).unwrap(); + + assert!(connection.read_once().unwrap().is_empty()); + peer.shutdown(Shutdown::Write).unwrap(); + + let error = connection + .read_once() + .expect_err("EOF after partial header should fail"); + assert_eq!(error.kind(), io::ErrorKind::UnexpectedEof); + } +} diff --git a/src/app/discord/mod.rs b/src/app/discord/mod.rs new file mode 100644 index 0000000..45a454b --- /dev/null +++ b/src/app/discord/mod.rs @@ -0,0 +1,48 @@ +mod imp; +mod ipc; +mod worker; + +use adw::subclass::prelude::ObjectSubclassIsExt; +use gtk::glib; + +use crate::app::ipc::event::DiscordActivity; + +glib::wrapper! { + pub struct Discord(ObjectSubclass); +} + +impl Default for Discord { + fn default() -> Self { + glib::Object::builder().build() + } +} + +impl Discord { + pub fn start(&self) { + self.imp().start(); + } + + pub fn stop(&self) { + self.imp().stop(); + } + + pub fn connect(&self) { + self.imp().connect(); + } + + pub fn disconnect(&self) { + self.imp().disconnect(); + } + + pub fn set_activity(&self, activity: DiscordActivity) { + self.imp().set_activity(activity); + } + + pub fn clear_activity(&self) { + self.imp().clear_activity(); + } + + pub fn connect_status(&self, callback: F) { + self.imp().set_status_callback(callback); + } +} diff --git a/src/app/discord/worker.rs b/src/app/discord/worker.rs new file mode 100644 index 0000000..8fa46b8 --- /dev/null +++ b/src/app/discord/worker.rs @@ -0,0 +1,1331 @@ +use std::{ + sync::mpsc::{Receiver, RecvTimeoutError}, + time::{Duration, Instant}, +}; + +use discord_rich_presence::activity; +use tracing::{debug, warn}; + +use super::ipc::DiscordIpcTransport; +use crate::app::ipc::event::DiscordActivity; + +const DISCORD_APP_ID: &str = "1452620752263319665"; +const FALLBACK_LARGE_IMAGE: &str = "stremio_logo"; +const LARGE_IMAGE_TEXT: &str = "Stremio"; +const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(3); +const LIVENESS_INTERVAL: Duration = Duration::from_secs(1); + +/// Maximum length of the user-visible `state` and `details` fields in +/// UTF-16 code units, matching how Discord's JavaScript client measures +/// string length. +const MAX_FIELD_LENGTH: usize = 128; + +pub enum DiscordCommand { + Connect, + Disconnect, + SetActivity(DiscordActivity), + ClearActivity, +} + +/// Minimal seam over the Discord IPC client so the command handling +/// state machine can be tested without a running Discord client. +pub trait DiscordClient { + fn connect(&mut self) -> Result<(), String>; + fn set_activity(&mut self, activity: &DiscordActivity) -> Result<(), String>; + fn clear_activity(&mut self) -> Result<(), String>; + fn poll_liveness(&mut self) -> Result<(), String>; + fn close(&mut self) -> Result<(), String>; +} + +pub struct RichPresenceClient { + client: DiscordIpcTransport, +} + +impl RichPresenceClient { + pub fn new() -> Self { + Self { + client: DiscordIpcTransport::new(DISCORD_APP_ID, HANDSHAKE_TIMEOUT), + } + } + + #[cfg(test)] + fn with_path(path: std::path::PathBuf, handshake_timeout: Duration) -> Self { + Self { + client: DiscordIpcTransport::with_path(DISCORD_APP_ID, path, handshake_timeout), + } + } +} + +impl DiscordClient for RichPresenceClient { + fn connect(&mut self) -> Result<(), String> { + self.client.connect() + } + + fn set_activity(&mut self, activity: &DiscordActivity) -> Result<(), String> { + let payload = serde_json::to_value(build_activity(activity)) + .map_err(|error| format!("failed to serialize Discord activity: {error}"))?; + self.client.set_activity(payload) + } + + fn clear_activity(&mut self) -> Result<(), String> { + self.client.clear_activity() + } + + fn poll_liveness(&mut self) -> Result<(), String> { + self.client.poll_liveness() + } + + fn close(&mut self) -> Result<(), String> { + self.client.close() + } +} + +/// Runs the Discord command loop until all command senders are dropped, +/// then closes the connection. Blocking: must run on a dedicated thread. +pub fn run(receiver: Receiver, status: S, make_client: F) +where + C: DiscordClient, + S: FnMut(bool), + F: Fn() -> C, +{ + run_with_liveness_interval(receiver, status, make_client, LIVENESS_INTERVAL); +} + +fn run_with_liveness_interval( + receiver: Receiver, + mut status: S, + make_client: F, + liveness_interval: Duration, +) where + C: DiscordClient, + S: FnMut(bool), + F: Fn() -> C, +{ + debug!("Discord worker thread started"); + + let mut client: Option = None; + let mut cached_activity: Option = None; + let mut next_liveness_check = Instant::now() + liveness_interval; + + loop { + let wait = next_liveness_check.saturating_duration_since(Instant::now()); + match receiver.recv_timeout(wait) { + Ok(command) => handle_command( + &mut client, + &mut cached_activity, + &make_client, + command, + &mut status, + ), + Err(RecvTimeoutError::Timeout) => {} + Err(RecvTimeoutError::Disconnected) => break, + } + + if Instant::now() >= next_liveness_check { + next_liveness_check = Instant::now() + liveness_interval; + if let Some(current_client) = client.as_mut() + && let Err(e) = current_client.poll_liveness() + { + debug!("Discord transport lost; liveness failure dropped client: {e}"); + client = None; + status(false); + } + } + } + + if let Some(mut current_client) = client { + let _ = current_client.close(); + } + + debug!("Discord worker thread exiting"); +} + +fn handle_command( + client: &mut Option, + cached_activity: &mut Option, + make_client: &F, + command: DiscordCommand, + status: &mut S, +) where + C: DiscordClient, + F: Fn() -> C, + S: FnMut(bool) + ?Sized, +{ + match command { + DiscordCommand::Connect => { + if client.is_some() { + status(true); + return; + } + + let mut next_client = make_client(); + match next_client.connect() { + Ok(()) => { + debug!("Discord connect succeeded"); + *client = Some(next_client); + status(true); + + if let Some(activity) = cached_activity.as_ref() { + debug!("Discord cached activity replay started"); + if let Err(e) = client + .as_mut() + .expect("client was just connected") + .set_activity(activity) + { + warn!("Discord cached activity replay failed: {e}"); + *client = None; + status(false); + } else { + debug!("Discord cached activity replay succeeded"); + } + } + } + Err(e) => { + debug!("Discord connect failed: {e}"); + status(false); + } + } + } + DiscordCommand::Disconnect => { + *cached_activity = None; + + if let Some(mut current_client) = client.take() + && let Err(e) = current_client.close() + { + debug!("Discord transport unavailable during disconnect: {e}"); + } + + status(false); + } + DiscordCommand::SetActivity(activity) => { + *cached_activity = Some(activity); + + let Some(current_client) = client.as_mut() else { + return; + }; + + if let Err(e) = current_client + .set_activity(cached_activity.as_ref().expect("activity was just cached")) + { + warn!("Failed to set Discord activity after transport loss: {e}"); + *client = None; + status(false); + } + } + DiscordCommand::ClearActivity => { + *cached_activity = None; + + let Some(current_client) = client.as_mut() else { + return; + }; + + if let Err(e) = current_client.clear_activity() { + warn!("Failed to clear Discord activity after transport loss: {e}"); + *client = None; + status(false); + } + } + } +} + +/// Builds the Discord rich-presence payload for a web-UI activity, +/// clamping the user-visible fields to Discord's length limit. +fn build_activity(activity: &DiscordActivity) -> activity::Activity<'_> { + let state = clamp_field(&activity.state); + let details = clamp_field(&activity.details); + + let mut payload = activity::Activity::new() + .activity_type(activity::ActivityType::Watching) + .assets( + activity::Assets::new() + .large_image(activity.image.as_deref().unwrap_or(FALLBACK_LARGE_IMAGE)) + .large_text(LARGE_IMAGE_TEXT), + ); + + // Omit empty optional fields instead of sending empty strings. + if !state.is_empty() { + payload = payload.state(state); + } + if !details.is_empty() { + payload = payload.details(details); + } + + let timestamps = match (activity.start_timestamp, activity.end_timestamp) { + (Some(start), Some(end)) => Some(activity::Timestamps::new().start(start).end(end)), + (Some(start), None) => Some(activity::Timestamps::new().start(start)), + (None, Some(end)) => Some(activity::Timestamps::new().end(end)), + (None, None) => None, + }; + if let Some(timestamps) = timestamps { + payload = payload.timestamps(timestamps); + } + + payload +} + +/// Limits a user-visible activity field to Discord's length limit, measured +/// in UTF-16 code units (as Discord's JavaScript client counts), keeping as +/// much of the input as fits without splitting a Unicode scalar value. +fn clamp_field(value: &str) -> String { + let mut units = 0; + let mut end = value.len(); + + for (index, character) in value.char_indices() { + units += character.len_utf16(); + if units > MAX_FIELD_LENGTH { + end = index; + break; + } + } + + value[..end].to_string() +} + +#[cfg(test)] +mod tests { + use std::{ + cell::Cell, + fs, + io::{Read, Write}, + os::unix::net::{UnixListener, UnixStream}, + path::PathBuf, + sync::{ + Arc, Mutex, + atomic::{AtomicU64, AtomicUsize, Ordering}, + mpsc, + }, + time::{Duration, Instant}, + }; + + use super::*; + + fn activity() -> DiscordActivity { + DiscordActivity { + state: "Watching".to_owned(), + details: "Movie".to_owned(), + image: Some("https://example.com/poster.jpg".to_owned()), + start_timestamp: Some(1_752_700_000), + end_timestamp: Some(1_752_707_200), + } + } + + #[derive(Default)] + struct MockClient { + closed: usize, + activities: usize, + cleared: usize, + fail_connect: bool, + fail_set_activity: bool, + fail_clear_activity: bool, + fail_liveness: bool, + } + + impl DiscordClient for MockClient { + fn connect(&mut self) -> Result<(), String> { + if self.fail_connect { + return Err("connect failed".to_owned()); + } + + Ok(()) + } + + fn set_activity(&mut self, _activity: &DiscordActivity) -> Result<(), String> { + if self.fail_set_activity { + return Err("set activity failed".to_owned()); + } + + self.activities += 1; + Ok(()) + } + + fn clear_activity(&mut self) -> Result<(), String> { + if self.fail_clear_activity { + return Err("clear activity failed".to_owned()); + } + + self.cleared += 1; + Ok(()) + } + + fn poll_liveness(&mut self) -> Result<(), String> { + if self.fail_liveness { + return Err("connection reset".to_owned()); + } + + Ok(()) + } + + fn close(&mut self) -> Result<(), String> { + self.closed += 1; + Ok(()) + } + } + + fn make_client() -> MockClient { + MockClient::default() + } + + fn make_failing_client() -> MockClient { + MockClient { + fail_connect: true, + ..Default::default() + } + } + + #[test] + fn connect_success_reports_connected() { + let mut client: Option = None; + let mut cached_activity = None; + let mut statuses = vec![]; + + handle_command( + &mut client, + &mut cached_activity, + &make_client, + DiscordCommand::Connect, + &mut |c| statuses.push(c), + ); + + assert_eq!(statuses, [true]); + assert!(client.is_some()); + } + + #[test] + fn connect_is_idempotent() { + let mut client: Option = None; + let mut cached_activity = None; + let constructed = Cell::new(0); + let make_counting_client = || { + constructed.set(constructed.get() + 1); + MockClient::default() + }; + let mut statuses = vec![]; + + handle_command( + &mut client, + &mut cached_activity, + &make_counting_client, + DiscordCommand::Connect, + &mut |c| statuses.push(c), + ); + handle_command( + &mut client, + &mut cached_activity, + &make_counting_client, + DiscordCommand::Connect, + &mut |c| statuses.push(c), + ); + + assert_eq!(statuses, [true, true]); + assert_eq!(constructed.get(), 1); + } + + #[test] + fn connect_failure_reports_disconnected() { + let mut client: Option = None; + let mut cached_activity = None; + let mut statuses = vec![]; + + handle_command( + &mut client, + &mut cached_activity, + &make_failing_client, + DiscordCommand::Connect, + &mut |c| statuses.push(c), + ); + + assert_eq!(statuses, [false]); + assert!(client.is_none()); + } + + #[test] + fn disconnect_closes_client_and_reports_disconnected() { + let mut client: Option = None; + let mut cached_activity = None; + let mut statuses = vec![]; + + handle_command( + &mut client, + &mut cached_activity, + &make_client, + DiscordCommand::Connect, + &mut |c| statuses.push(c), + ); + handle_command( + &mut client, + &mut cached_activity, + &make_client, + DiscordCommand::Disconnect, + &mut |c| statuses.push(c), + ); + + assert_eq!(statuses, [true, false]); + assert!(client.is_none()); + } + + #[test] + fn disconnect_without_client_still_reports_disconnected() { + let mut client: Option = None; + let mut cached_activity = None; + let mut statuses = vec![]; + + handle_command( + &mut client, + &mut cached_activity, + &make_client, + DiscordCommand::Disconnect, + &mut |c| statuses.push(c), + ); + + assert_eq!(statuses, [false]); + } + + #[test] + fn set_activity_while_disconnected_is_replayed_on_connect() { + let mut client: Option = None; + let mut cached_activity = None; + let mut statuses = vec![]; + + handle_command( + &mut client, + &mut cached_activity, + &make_client, + DiscordCommand::SetActivity(activity()), + &mut |c| statuses.push(c), + ); + handle_command( + &mut client, + &mut cached_activity, + &make_client, + DiscordCommand::Connect, + &mut |c| statuses.push(c), + ); + + assert_eq!(statuses, [true]); + assert_eq!(cached_activity, Some(activity())); + assert_eq!( + client + .as_ref() + .expect("client should be connected") + .activities, + 1 + ); + } + + #[test] + fn set_activity_is_forwarded_to_client() { + let mut client: Option = None; + let mut cached_activity = None; + let mut statuses = vec![]; + + handle_command( + &mut client, + &mut cached_activity, + &make_client, + DiscordCommand::Connect, + &mut |c| statuses.push(c), + ); + handle_command( + &mut client, + &mut cached_activity, + &make_client, + DiscordCommand::SetActivity(activity()), + &mut |c| statuses.push(c), + ); + + assert_eq!(statuses, [true]); + assert_eq!( + client + .as_ref() + .expect("client should be connected") + .activities, + 1 + ); + } + + #[test] + fn clear_activity_is_forwarded_to_client() { + let mut client: Option = None; + let mut cached_activity = None; + let mut statuses = vec![]; + + handle_command( + &mut client, + &mut cached_activity, + &make_client, + DiscordCommand::Connect, + &mut |c| statuses.push(c), + ); + handle_command( + &mut client, + &mut cached_activity, + &make_client, + DiscordCommand::ClearActivity, + &mut |c| statuses.push(c), + ); + + assert_eq!(statuses, [true]); + assert_eq!( + client.as_ref().expect("client should be connected").cleared, + 1 + ); + } + + #[test] + fn set_activity_failure_drops_client_and_reports_disconnected() { + let mut client: Option = None; + let mut cached_activity = None; + let mut statuses = vec![]; + + handle_command( + &mut client, + &mut cached_activity, + &make_client, + DiscordCommand::Connect, + &mut |c| statuses.push(c), + ); + client + .as_mut() + .expect("client should be connected") + .fail_set_activity = true; + handle_command( + &mut client, + &mut cached_activity, + &make_client, + DiscordCommand::SetActivity(activity()), + &mut |c| statuses.push(c), + ); + + assert_eq!(statuses, [true, false]); + assert!(client.is_none()); + assert_eq!(cached_activity, Some(activity())); + } + + #[test] + fn transport_failure_preserves_activity_for_reconnect() { + let mut client: Option = None; + let mut cached_activity = None; + let mut statuses = vec![]; + + handle_command( + &mut client, + &mut cached_activity, + &make_client, + DiscordCommand::Connect, + &mut |c| statuses.push(c), + ); + handle_command( + &mut client, + &mut cached_activity, + &make_client, + DiscordCommand::SetActivity(activity()), + &mut |c| statuses.push(c), + ); + client + .as_mut() + .expect("client should be connected") + .fail_set_activity = true; + handle_command( + &mut client, + &mut cached_activity, + &make_client, + DiscordCommand::SetActivity(activity()), + &mut |c| statuses.push(c), + ); + handle_command( + &mut client, + &mut cached_activity, + &make_client, + DiscordCommand::Connect, + &mut |c| statuses.push(c), + ); + + assert_eq!(statuses, [true, false, true]); + assert_eq!(cached_activity, Some(activity())); + assert_eq!( + client.as_ref().expect("client should reconnect").activities, + 1 + ); + } + + #[test] + fn liveness_failure_reports_disconnected_and_preserves_activity() { + let (commands, receiver) = mpsc::channel(); + let (status_sender, status_receiver) = mpsc::channel(); + let constructed = Arc::new(AtomicUsize::new(0)); + let client_count = Arc::clone(&constructed); + + let worker = std::thread::spawn(move || { + run_with_liveness_interval( + receiver, + move |connected| { + status_sender.send(connected).ok(); + }, + move || MockClient { + fail_liveness: client_count.fetch_add(1, Ordering::SeqCst) == 0, + ..Default::default() + }, + Duration::from_millis(10), + ); + }); + + commands.send(DiscordCommand::Connect).unwrap(); + commands + .send(DiscordCommand::SetActivity(activity())) + .unwrap(); + assert_eq!( + status_receiver.recv_timeout(Duration::from_secs(1)), + Ok(true) + ); + assert_eq!( + status_receiver.recv_timeout(Duration::from_secs(1)), + Ok(false) + ); + + commands.send(DiscordCommand::Connect).unwrap(); + assert_eq!( + status_receiver.recv_timeout(Duration::from_secs(1)), + Ok(true) + ); + drop(commands); + worker.join().expect("worker thread should terminate"); + assert_eq!(constructed.load(Ordering::SeqCst), 2); + } + + static NEXT_TEST_SOCKET: AtomicU64 = AtomicU64::new(0); + + struct TestSocketDir(PathBuf); + + impl TestSocketDir { + fn new() -> Self { + let path = std::env::temp_dir().join(format!( + "stremio-discord-test-{}-{}", + std::process::id(), + NEXT_TEST_SOCKET.fetch_add(1, Ordering::Relaxed) + )); + fs::create_dir(&path).expect("test socket directory should be created"); + Self(path) + } + + fn socket_path(&self) -> PathBuf { + self.0.join("discord-ipc-0") + } + } + + impl Drop for TestSocketDir { + fn drop(&mut self) { + let _ = fs::remove_dir_all(&self.0); + } + } + + fn read_frame(stream: &mut UnixStream) -> (u32, serde_json::Value) { + stream + .set_read_timeout(Some(Duration::from_secs(2))) + .expect("server read timeout should be set"); + let mut header = [0_u8; 8]; + stream + .read_exact(&mut header) + .expect("server should receive frame header"); + let opcode = u32::from_le_bytes(header[..4].try_into().unwrap()); + let length = u32::from_le_bytes(header[4..].try_into().unwrap()) as usize; + let mut payload = vec![0_u8; length]; + stream + .read_exact(&mut payload) + .expect("server should receive frame payload"); + ( + opcode, + serde_json::from_slice(&payload).expect("client frame should contain JSON"), + ) + } + + fn write_frame(stream: &mut UnixStream, opcode: u32, payload: serde_json::Value) { + let payload = serde_json::to_vec(&payload).expect("server payload should serialize"); + stream + .write_all(&opcode.to_le_bytes()) + .and_then(|()| stream.write_all(&(payload.len() as u32).to_le_bytes())) + .and_then(|()| stream.write_all(&payload)) + .expect("server should send frame"); + } + + fn accept_handshake(listener: &UnixListener) -> UnixStream { + let (mut stream, _) = listener.accept().expect("server should accept client"); + let (opcode, handshake) = read_frame(&mut stream); + assert_eq!(opcode, 0); + assert_eq!(handshake["v"], 1); + assert_eq!(handshake["client_id"], DISCORD_APP_ID); + write_frame( + &mut stream, + 1, + serde_json::json!({ "cmd": "DISPATCH", "evt": "READY", "data": {} }), + ); + stream + } + + #[test] + fn eof_drops_client_and_reconnect_replays_cached_activity() { + let socket_dir = TestSocketDir::new(); + let socket_path = socket_dir.socket_path(); + let listener = UnixListener::bind(&socket_path).expect("fake Discord should bind"); + let (captured_sender, captured_receiver) = mpsc::channel(); + let (release_sender, release_receiver) = mpsc::channel(); + + let server = std::thread::spawn(move || { + let mut first = accept_handshake(&listener); + let (opcode, first_activity) = read_frame(&mut first); + assert_eq!(opcode, 1); + assert!(first_activity["nonce"].as_str().is_some()); + captured_sender.send(first_activity).unwrap(); + drop(first); + + let mut second = accept_handshake(&listener); + let (opcode, replayed_activity) = read_frame(&mut second); + assert_eq!(opcode, 1); + assert!(replayed_activity["nonce"].as_str().is_some()); + captured_sender.send(replayed_activity).unwrap(); + release_receiver + .recv_timeout(Duration::from_secs(2)) + .expect("test should release second connection"); + }); + + let (commands, receiver) = mpsc::channel(); + let (status_sender, status_receiver) = mpsc::channel(); + let client_path = socket_path.clone(); + let worker = std::thread::spawn(move || { + run_with_liveness_interval( + receiver, + move |connected| { + status_sender.send(connected).ok(); + }, + move || { + RichPresenceClient::with_path(client_path.clone(), Duration::from_millis(250)) + }, + Duration::from_millis(10), + ); + }); + + commands + .send(DiscordCommand::SetActivity(activity())) + .unwrap(); + commands.send(DiscordCommand::Connect).unwrap(); + assert_eq!( + status_receiver.recv_timeout(Duration::from_secs(1)), + Ok(true) + ); + let initially_sent = captured_receiver + .recv_timeout(Duration::from_secs(1)) + .expect("first activity should arrive"); + assert_eq!( + status_receiver.recv_timeout(Duration::from_secs(1)), + Ok(false) + ); + + commands.send(DiscordCommand::Connect).unwrap(); + assert_eq!( + status_receiver.recv_timeout(Duration::from_secs(1)), + Ok(true) + ); + let replayed = captured_receiver + .recv_timeout(Duration::from_secs(1)) + .expect("cached activity should be replayed"); + assert_eq!( + initially_sent["args"]["activity"], + replayed["args"]["activity"] + ); + assert!( + status_receiver + .recv_timeout(Duration::from_millis(50)) + .is_err(), + "a second liveness monitor must not survive reconnect" + ); + + commands.send(DiscordCommand::Disconnect).unwrap(); + assert_eq!( + status_receiver.recv_timeout(Duration::from_secs(1)), + Ok(false) + ); + drop(commands); + release_sender.send(()).unwrap(); + worker.join().expect("worker should terminate"); + server.join().expect("fake Discord should terminate"); + } + + #[test] + fn socket_replacement_is_detected_while_old_stream_remains_open() { + let socket_dir = TestSocketDir::new(); + let socket_path = socket_dir.socket_path(); + let listener = UnixListener::bind(&socket_path).expect("fake Discord should bind"); + let (connected_sender, connected_receiver) = mpsc::channel(); + let (release_sender, release_receiver) = mpsc::channel(); + let server = std::thread::spawn(move || { + let _stream = accept_handshake(&listener); + connected_sender.send(()).unwrap(); + release_receiver + .recv_timeout(Duration::from_secs(2)) + .expect("test should release stale connection"); + }); + + let mut client = + RichPresenceClient::with_path(socket_path.clone(), Duration::from_millis(250)); + client.connect().expect("client should handshake"); + connected_receiver + .recv_timeout(Duration::from_secs(1)) + .expect("server should hold old stream open"); + + fs::remove_file(&socket_path).expect("old socket path should be removed"); + let replacement = UnixListener::bind(&socket_path).expect("replacement should bind"); + let error = client + .poll_liveness() + .expect_err("replaced socket should fail liveness"); + assert!(error.contains("replaced")); + + drop(replacement); + release_sender.send(()).unwrap(); + server.join().expect("fake Discord should terminate"); + } + + #[test] + fn worker_shutdown_finishes_after_handshake_timeout() { + let socket_dir = TestSocketDir::new(); + let socket_path = socket_dir.socket_path(); + let listener = UnixListener::bind(&socket_path).expect("fake Discord should bind"); + let server = std::thread::spawn(move || { + let (mut stream, _) = listener.accept().expect("server should accept client"); + let _ = read_frame(&mut stream); + std::thread::sleep(Duration::from_millis(300)); + }); + + let (commands, receiver) = mpsc::channel(); + let path = socket_path.clone(); + let worker = std::thread::spawn(move || { + run_with_liveness_interval( + receiver, + |_| {}, + move || RichPresenceClient::with_path(path.clone(), Duration::from_millis(100)), + Duration::from_millis(10), + ); + }); + + let started = Instant::now(); + commands.send(DiscordCommand::Connect).unwrap(); + drop(commands); + worker.join().expect("worker should terminate"); + assert!(started.elapsed() < Duration::from_secs(1)); + server.join().expect("fake Discord should terminate"); + } + + #[test] + fn clear_activity_failure_drops_client_and_reports_disconnected() { + let mut client: Option = None; + let mut cached_activity = None; + let mut statuses = vec![]; + + handle_command( + &mut client, + &mut cached_activity, + &make_client, + DiscordCommand::Connect, + &mut |c| statuses.push(c), + ); + client + .as_mut() + .expect("client should be connected") + .fail_clear_activity = true; + handle_command( + &mut client, + &mut cached_activity, + &make_client, + DiscordCommand::ClearActivity, + &mut |c| statuses.push(c), + ); + + assert_eq!(statuses, [true, false]); + assert!(client.is_none()); + assert!(cached_activity.is_none()); + } + + #[test] + fn clear_activity_prevents_replay_after_reconnect() { + let mut client: Option = None; + let mut cached_activity = None; + let mut statuses = vec![]; + + handle_command( + &mut client, + &mut cached_activity, + &make_client, + DiscordCommand::Connect, + &mut |c| statuses.push(c), + ); + handle_command( + &mut client, + &mut cached_activity, + &make_client, + DiscordCommand::SetActivity(activity()), + &mut |c| statuses.push(c), + ); + client + .as_mut() + .expect("client should be connected") + .fail_clear_activity = true; + handle_command( + &mut client, + &mut cached_activity, + &make_client, + DiscordCommand::ClearActivity, + &mut |c| statuses.push(c), + ); + handle_command( + &mut client, + &mut cached_activity, + &make_client, + DiscordCommand::Connect, + &mut |c| statuses.push(c), + ); + + assert_eq!(statuses, [true, false, true]); + assert!(cached_activity.is_none()); + assert_eq!( + client.as_ref().expect("client should reconnect").activities, + 0 + ); + } + + #[test] + fn failed_replay_retains_activity_for_later_reconnect() { + let mut client: Option = None; + let mut cached_activity = None; + let constructed = Cell::new(0); + let make_replay_failing_client = || { + let fail_set_activity = constructed.get() == 0; + constructed.set(constructed.get() + 1); + MockClient { + fail_set_activity, + ..Default::default() + } + }; + let mut statuses = vec![]; + + handle_command( + &mut client, + &mut cached_activity, + &make_replay_failing_client, + DiscordCommand::SetActivity(activity()), + &mut |c| statuses.push(c), + ); + handle_command( + &mut client, + &mut cached_activity, + &make_replay_failing_client, + DiscordCommand::Connect, + &mut |c| statuses.push(c), + ); + + assert_eq!(statuses, [true, false]); + assert!(client.is_none()); + assert_eq!(cached_activity, Some(activity())); + + handle_command( + &mut client, + &mut cached_activity, + &make_replay_failing_client, + DiscordCommand::Connect, + &mut |c| statuses.push(c), + ); + + assert_eq!(statuses, [true, false, true]); + assert_eq!( + client.as_ref().expect("client should reconnect").activities, + 1 + ); + } + + #[test] + fn run_closes_client_when_channel_closes() { + let (commands, receiver) = mpsc::channel(); + let (status_sender, status_receiver) = mpsc::channel(); + + let worker = std::thread::spawn(move || { + run( + receiver, + move |connected| { + status_sender.send(connected).ok(); + }, + make_client, + ); + }); + + commands.send(DiscordCommand::Connect).ok(); + commands.send(DiscordCommand::SetActivity(activity())).ok(); + drop(commands); + + worker.join().expect("worker thread should terminate"); + + let statuses: Vec = status_receiver.try_iter().collect(); + assert_eq!(statuses, [true]); + } + + #[test] + fn command_ordering_is_deterministic() { + struct RecordingClient(Arc>>); + + impl DiscordClient for RecordingClient { + fn connect(&mut self) -> Result<(), String> { + self.0.lock().unwrap().push("connect"); + Ok(()) + } + + fn set_activity(&mut self, _activity: &DiscordActivity) -> Result<(), String> { + self.0.lock().unwrap().push("set"); + Ok(()) + } + + fn clear_activity(&mut self) -> Result<(), String> { + self.0.lock().unwrap().push("clear"); + Ok(()) + } + + fn poll_liveness(&mut self) -> Result<(), String> { + self.0.lock().unwrap().push("poll"); + Ok(()) + } + + fn close(&mut self) -> Result<(), String> { + self.0.lock().unwrap().push("close"); + Ok(()) + } + } + + let calls = Arc::new(Mutex::new(Vec::new())); + let client_calls = Arc::clone(&calls); + let (commands, receiver) = mpsc::channel(); + let worker = std::thread::spawn(move || { + run_with_liveness_interval( + receiver, + |_| {}, + move || RecordingClient(Arc::clone(&client_calls)), + Duration::from_secs(1), + ); + }); + + commands.send(DiscordCommand::Connect).unwrap(); + commands + .send(DiscordCommand::SetActivity(activity())) + .unwrap(); + commands.send(DiscordCommand::ClearActivity).unwrap(); + commands.send(DiscordCommand::Disconnect).unwrap(); + drop(commands); + worker.join().expect("worker should terminate"); + + assert_eq!(*calls.lock().unwrap(), ["connect", "set", "clear", "close"]); + } + + #[test] + fn explicit_disconnect_clears_activity_before_reenable() { + let mut client: Option = None; + let mut cached_activity = None; + let mut statuses = vec![]; + + handle_command( + &mut client, + &mut cached_activity, + &make_client, + DiscordCommand::Connect, + &mut |c| statuses.push(c), + ); + handle_command( + &mut client, + &mut cached_activity, + &make_client, + DiscordCommand::SetActivity(activity()), + &mut |c| statuses.push(c), + ); + handle_command( + &mut client, + &mut cached_activity, + &make_client, + DiscordCommand::Disconnect, + &mut |c| statuses.push(c), + ); + handle_command( + &mut client, + &mut cached_activity, + &make_client, + DiscordCommand::Connect, + &mut |c| statuses.push(c), + ); + + assert_eq!(statuses, [true, false, true]); + assert!(cached_activity.is_none()); + assert_eq!( + client.as_ref().expect("client should reconnect").activities, + 0 + ); + } + + fn activity_json(activity: &DiscordActivity) -> serde_json::Value { + serde_json::to_value(build_activity(activity)).expect("activity should serialize") + } + + #[test] + fn build_activity_populates_state_details_and_type() { + let json = activity_json(&activity()); + + assert_eq!(json["state"], "Watching"); + assert_eq!(json["details"], "Movie"); + assert_eq!(json["type"], 3); + assert_eq!(json["assets"]["large_text"], "Stremio"); + } + + #[test] + fn build_activity_omits_empty_state() { + let json = activity_json(&DiscordActivity { + state: String::new(), + ..activity() + }); + + assert!(json.get("state").is_none()); + assert_eq!(json["details"], "Movie"); + } + + #[test] + fn build_activity_omits_empty_details() { + let json = activity_json(&DiscordActivity { + details: String::new(), + ..activity() + }); + + assert_eq!(json["state"], "Watching"); + assert!(json.get("details").is_none()); + } + + #[test] + fn build_activity_uses_provided_image() { + let json = activity_json(&activity()); + + assert_eq!( + json["assets"]["large_image"], + "https://example.com/poster.jpg" + ); + } + + #[test] + fn build_activity_without_image_uses_fallback_asset() { + let json = activity_json(&DiscordActivity { + image: None, + ..activity() + }); + + assert_eq!(json["assets"]["large_image"], "stremio_logo"); + assert_eq!(json["assets"]["large_text"], "Stremio"); + } + + #[test] + fn build_activity_without_timestamps_omits_them() { + let json = activity_json(&DiscordActivity { + start_timestamp: None, + end_timestamp: None, + ..activity() + }); + + assert!(json.get("timestamps").is_none()); + } + + #[test] + fn build_activity_with_only_start_timestamp() { + let json = activity_json(&DiscordActivity { + end_timestamp: None, + ..activity() + }); + + assert_eq!(json["timestamps"]["start"], 1_752_700_000); + assert!(json["timestamps"].get("end").is_none()); + } + + #[test] + fn build_activity_with_only_end_timestamp() { + let json = activity_json(&DiscordActivity { + start_timestamp: None, + ..activity() + }); + + assert_eq!(json["timestamps"]["end"], 1_752_707_200); + assert!(json["timestamps"].get("start").is_none()); + } + + #[test] + fn build_activity_with_both_timestamps() { + let json = activity_json(&activity()); + + assert_eq!(json["timestamps"]["start"], 1_752_700_000); + assert_eq!(json["timestamps"]["end"], 1_752_707_200); + } + + #[test] + fn build_activity_clamps_fields_to_utf16_limit() { + let json = activity_json(&DiscordActivity { + state: "s".repeat(MAX_FIELD_LENGTH + 50), + details: "🎬".repeat(MAX_FIELD_LENGTH + 1), + ..activity() + }); + + let state = json["state"].as_str().expect("state should be present"); + assert_eq!(utf16_len(state), MAX_FIELD_LENGTH); + let details = json["details"].as_str().expect("details should be present"); + assert_eq!(details, "🎬".repeat(MAX_FIELD_LENGTH / 2)); + assert_eq!(utf16_len(details), MAX_FIELD_LENGTH); + } + + fn utf16_len(value: &str) -> usize { + value.chars().map(char::len_utf16).sum() + } + + #[test] + fn clamp_field_keeps_strings_below_limit() { + assert_eq!(clamp_field("Watching"), "Watching"); + assert_eq!(clamp_field(""), ""); + } + + #[test] + fn clamp_field_keeps_strings_exactly_at_limit() { + let value = "b".repeat(MAX_FIELD_LENGTH); + assert_eq!(clamp_field(&value), value); + } + + #[test] + fn clamp_field_limits_overlong_bmp_strings() { + let value = "a".repeat(MAX_FIELD_LENGTH + 50); + let clamped = clamp_field(&value); + + assert_eq!(utf16_len(&clamped), MAX_FIELD_LENGTH); + assert!(value.starts_with(&clamped)); + } + + #[test] + fn clamp_field_counts_multibyte_bmp_chars_as_one_unit() { + // 'á' is one UTF-16 code unit but two UTF-8 bytes. + let value = "á".repeat(MAX_FIELD_LENGTH + 1); + let clamped = clamp_field(&value); + + assert_eq!(clamped, "á".repeat(MAX_FIELD_LENGTH)); + } + + #[test] + fn clamp_field_limits_non_bmp_emoji_by_utf16_units() { + // '🎬' is outside the BMP and takes two UTF-16 code units. + let value = "🎬".repeat(MAX_FIELD_LENGTH / 2 + 1); + let clamped = clamp_field(&value); + + assert_eq!(clamped, "🎬".repeat(MAX_FIELD_LENGTH / 2)); + assert_eq!(utf16_len(&clamped), MAX_FIELD_LENGTH); + } + + #[test] + fn clamp_field_never_splits_a_surrogate_pair() { + // With a single UTF-16 unit left, the two-unit emoji is dropped whole. + let value = format!("{}🎬", "a".repeat(MAX_FIELD_LENGTH - 1)); + let clamped = clamp_field(&value); + + assert_eq!(clamped, "a".repeat(MAX_FIELD_LENGTH - 1)); + assert_eq!(utf16_len(&clamped), MAX_FIELD_LENGTH - 1); + } +} diff --git a/src/app/imp.rs b/src/app/imp.rs index 65f108b..93931c9 100644 --- a/src/app/imp.rs +++ b/src/app/imp.rs @@ -7,6 +7,7 @@ use tracing::error; use crate::{ app::{ config::{APP_ID, APP_NAME, URI_SCHEME}, + discord::Discord, ipc::{ self, event::{IpcEvent, IpcEventMpv}, @@ -33,6 +34,7 @@ pub struct Application { decorations: Cell, tray: RefCell>, mpris: RefCell>, + discord: RefCell>, window: RefCell>, webview: RefCell>, deeplink: RefCell>, @@ -71,6 +73,7 @@ impl ApplicationImpl for Application { let tray = Tray::default(); let video = Video::default(); let mpris = Mpris::default(); + let discord = Discord::default(); let startup_url = self.startup_url.borrow(); let dev_mode = self.dev_mode.get(); @@ -132,6 +135,8 @@ impl ApplicationImpl for Application { video, #[weak] mpris, + #[weak] + discord, move |webview: WebView, message: &str| { if let Ok(event) = ipc::parse_request(message) { match event { @@ -161,6 +166,18 @@ impl ApplicationImpl for Application { IpcEvent::Quit => { app.quit(); } + IpcEvent::DiscordConnect => { + discord.connect(); + } + IpcEvent::DiscordDisconnect => { + discord.disconnect(); + } + IpcEvent::DiscordSetActivity(activity) => { + discord.set_activity(activity); + } + IpcEvent::DiscordClearActivity => { + discord.clear_activity(); + } IpcEvent::Mpv(event) => match event { IpcEventMpv::Observe(name) => video.observe_mpv_property(name), IpcEventMpv::Command((name, args)) => { @@ -255,10 +272,22 @@ impl ApplicationImpl for Application { mpris.start(APP_ID, APP_NAME); + discord.connect_status(clone!( + #[weak] + webview, + move |connected| { + let message = ipc::create_response(IpcEvent::DiscordStatus(connected)); + webview.send(&message); + } + )); + + discord.start(); + window.present(); *self.tray.borrow_mut() = Some(tray); *self.mpris.borrow_mut() = Some(mpris); + *self.discord.borrow_mut() = Some(discord); *self.window.borrow_mut() = Some(window); *self.webview.borrow_mut() = Some(webview); } @@ -283,6 +312,10 @@ impl ApplicationImpl for Application { } fn shutdown(&self) { + if let Some(discord) = self.discord.take() { + discord.stop(); + } + if let Some(window) = self.window.take() { window.destroy(); } diff --git a/src/app/ipc/event.rs b/src/app/ipc/event.rs index 3cc5967..938c7b8 100644 --- a/src/app/ipc/event.rs +++ b/src/app/ipc/event.rs @@ -12,6 +12,15 @@ pub enum IpcEventMpv { Ended((String, Option)), } +#[derive(Deserialize, Debug, Clone, PartialEq, Eq)] +pub struct DiscordActivity { + pub state: String, + pub details: String, + pub image: Option, + pub start_timestamp: Option, + pub end_timestamp: Option, +} + #[derive(Deserialize, Debug)] pub enum IpcEvent { Init, @@ -23,6 +32,11 @@ pub enum IpcEvent { Mpv(IpcEventMpv), MediaMetadata((String, Option, Option)), MediaStatus(bool), + DiscordConnect, + DiscordDisconnect, + DiscordSetActivity(DiscordActivity), + DiscordClearActivity, + DiscordStatus(bool), } impl TryFrom<&str> for IpcEvent { diff --git a/src/app/ipc/request.rs b/src/app/ipc/request.rs index 2b85ed8..da34311 100644 --- a/src/app/ipc/request.rs +++ b/src/app/ipc/request.rs @@ -1,7 +1,7 @@ use serde::Deserialize; use serde_json::Value; -use super::event::{IpcEvent, IpcEventMpv}; +use super::event::{DiscordActivity, IpcEvent, IpcEventMpv}; #[derive(Deserialize, Debug)] pub struct IpcMessageRequest { @@ -28,6 +28,28 @@ pub struct IpcMessageRequestMediaStatus { paused: bool, } +#[derive(Deserialize, Debug)] +#[serde(rename_all = "camelCase")] +pub struct IpcMessageRequestDiscordSetActivity { + state: String, + details: String, + image: Option, + start_timestamp: Option, + end_timestamp: Option, +} + +impl From for DiscordActivity { + fn from(value: IpcMessageRequestDiscordSetActivity) -> Self { + Self { + state: value.state, + details: value.details, + image: value.image, + start_timestamp: value.start_timestamp, + end_timestamp: value.end_timestamp, + } + } +} + impl TryFrom for IpcEvent { type Error = String; @@ -100,6 +122,16 @@ impl TryFrom for IpcEvent { Ok(IpcEvent::MediaStatus(data.paused)) } + "discord-connect" => Ok(IpcEvent::DiscordConnect), + "discord-disconnect" => Ok(IpcEvent::DiscordDisconnect), + "discord-set-activity" => { + let data: IpcMessageRequestDiscordSetActivity = + serde_json::from_value(data) + .map_err(|_| "Invalid discord-set-activity object")?; + + Ok(IpcEvent::DiscordSetActivity(data.into())) + } + "discord-clear-activity" => Ok(IpcEvent::DiscordClearActivity), method => Err(format!("Invalid method: {method}")), }, None => match name { @@ -115,3 +147,184 @@ impl TryFrom for IpcEvent { } } } + +#[cfg(test)] +mod tests { + use super::*; + + fn parse(message: &str) -> Result { + IpcEvent::try_from(message) + } + + #[test] + fn parses_discord_connect() { + let event = parse(r#"{"type":6,"args":["discord-connect",{}]}"#); + + assert!(matches!(event, Ok(IpcEvent::DiscordConnect))); + } + + #[test] + fn parses_discord_disconnect() { + let event = parse(r#"{"type":6,"args":["discord-disconnect",{}]}"#); + + assert!(matches!(event, Ok(IpcEvent::DiscordDisconnect))); + } + + #[test] + fn parses_discord_clear_activity() { + let event = parse(r#"{"type":6,"args":["discord-clear-activity",{}]}"#); + + assert!(matches!(event, Ok(IpcEvent::DiscordClearActivity))); + } + + #[test] + fn parses_discord_set_activity() { + let event = parse( + r#"{"type":6,"args":["discord-set-activity",{ + "state": "Watching", + "details": "Movie", + "image": "https://example.com/poster.jpg", + "startTimestamp": 1752700000, + "endTimestamp": 1752707200 + }]}"#, + ); + + let Ok(IpcEvent::DiscordSetActivity(activity)) = event else { + panic!("Expected DiscordSetActivity, got {event:?}"); + }; + + assert_eq!(activity.state, "Watching"); + assert_eq!(activity.details, "Movie"); + assert_eq!( + activity.image.as_deref(), + Some("https://example.com/poster.jpg") + ); + assert_eq!(activity.start_timestamp, Some(1752700000)); + assert_eq!(activity.end_timestamp, Some(1752707200)); + } + + #[test] + fn parses_discord_set_activity_with_null_optional_fields() { + let event = parse( + r#"{"type":6,"args":["discord-set-activity",{ + "state": "Paused", + "details": "Episode", + "image": null, + "startTimestamp": null, + "endTimestamp": null + }]}"#, + ); + + let Ok(IpcEvent::DiscordSetActivity(activity)) = event else { + panic!("Expected DiscordSetActivity, got {event:?}"); + }; + + assert_eq!(activity.state, "Paused"); + assert_eq!(activity.details, "Episode"); + assert_eq!(activity.image, None); + assert_eq!(activity.start_timestamp, None); + assert_eq!(activity.end_timestamp, None); + } + + #[test] + fn parses_discord_set_activity_with_missing_optional_fields() { + let event = parse( + r#"{"type":6,"args":["discord-set-activity",{ + "state": "Watching", + "details": "Movie" + }]}"#, + ); + + let Ok(IpcEvent::DiscordSetActivity(activity)) = event else { + panic!("Expected DiscordSetActivity, got {event:?}"); + }; + + assert_eq!(activity.image, None); + assert_eq!(activity.start_timestamp, None); + assert_eq!(activity.end_timestamp, None); + } + + #[test] + fn rejects_discord_set_activity_with_missing_required_fields() { + let without_state = + parse(r#"{"type":6,"args":["discord-set-activity",{"details": "Movie"}]}"#); + let without_details = + parse(r#"{"type":6,"args":["discord-set-activity",{"state": "Watching"}]}"#); + + assert_eq!( + without_state.err(), + Some( + "Failed to convert IpcEvent to IpcMessageRequest: Invalid discord-set-activity object" + .to_owned() + ) + ); + assert_eq!( + without_details.err(), + Some( + "Failed to convert IpcEvent to IpcMessageRequest: Invalid discord-set-activity object" + .to_owned() + ) + ); + } + + #[test] + fn rejects_discord_set_activity_with_invalid_field_types() { + let invalid_state = parse( + r#"{"type":6,"args":["discord-set-activity",{"state": 42, "details": "Movie"}]}"#, + ); + let invalid_image = parse( + r#"{"type":6,"args":["discord-set-activity",{"state": "Watching", "details": "Movie", "image": 42}]}"#, + ); + let invalid_timestamp = parse( + r#"{"type":6,"args":["discord-set-activity",{"state": "Watching", "details": "Movie", "startTimestamp": "now"}]}"#, + ); + + assert_eq!( + invalid_state.err(), + Some( + "Failed to convert IpcEvent to IpcMessageRequest: Invalid discord-set-activity object" + .to_owned() + ) + ); + assert_eq!( + invalid_image.err(), + Some( + "Failed to convert IpcEvent to IpcMessageRequest: Invalid discord-set-activity object" + .to_owned() + ) + ); + assert_eq!( + invalid_timestamp.err(), + Some( + "Failed to convert IpcEvent to IpcMessageRequest: Invalid discord-set-activity object" + .to_owned() + ) + ); + } + + #[test] + fn rejects_discord_methods_without_payload() { + let event = parse(r#"{"type":6,"args":["discord-connect"]}"#); + + assert_eq!( + event.err(), + Some( + "Failed to convert IpcEvent to IpcMessageRequest: Invalid method: discord-connect" + .to_owned() + ) + ); + } + + #[test] + fn rejects_unknown_discord_method() { + let event = parse(r#"{"type":6,"args":["discord-status",{}]}"#); + + assert_eq!( + event.err(), + Some( + "Failed to convert IpcEvent to IpcMessageRequest: Invalid method: discord-status" + .to_owned() + ) + ); + } +} diff --git a/src/app/ipc/response.rs b/src/app/ipc/response.rs index 76803be..f45a033 100644 --- a/src/app/ipc/response.rs +++ b/src/app/ipc/response.rs @@ -102,7 +102,52 @@ impl TryFrom for IpcMessageResponse { "paused": paused }])), }), + IpcEvent::DiscordStatus(connected) => Ok(IpcMessageResponse { + id: 1, + r#type: 1, + object: TRANSPORT_NAME.to_owned(), + data: None, + args: Some(json!(["discord-status", { + "connected": connected + }])), + }), _ => Err("Failed to convert IpcEvent to IpcMessageResponse"), } } } + +#[cfg(test)] +mod tests { + use serde_json::Value; + + use super::*; + + fn serialize(event: IpcEvent) -> Value { + let response = IpcMessageResponse::try_from(event).expect("Failed to create response"); + serde_json::to_value(response).expect("Failed to serialize response") + } + + #[test] + fn serializes_discord_status_connected() { + let value = serialize(IpcEvent::DiscordStatus(true)); + + assert_eq!(value.get("type"), Some(&json!(1))); + assert_eq!(value.get("object"), Some(&json!("transport"))); + assert_eq!( + value.get("args"), + Some(&json!(["discord-status", { "connected": true }])) + ); + } + + #[test] + fn serializes_discord_status_disconnected() { + let value = serialize(IpcEvent::DiscordStatus(false)); + + assert_eq!(value.get("type"), Some(&json!(1))); + assert_eq!(value.get("object"), Some(&json!("transport"))); + assert_eq!( + value.get("args"), + Some(&json!(["discord-status", { "connected": false }])) + ); + } +} diff --git a/src/app/mod.rs b/src/app/mod.rs index a923107..6b21333 100644 --- a/src/app/mod.rs +++ b/src/app/mod.rs @@ -1,5 +1,6 @@ mod config; mod dialogs; +mod discord; mod imp; mod ipc; mod mpris; From 2ed69d61b57ddb27cec320859365b5a1dfe08e01 Mon Sep 17 00:00:00 2001 From: Christopher Nelson Date: Sat, 18 Jul 2026 18:39:43 -0400 Subject: [PATCH 2/4] build(flatpak): allow Discord IPC access --- flatpak/com.stremio.Stremio.Devel.json | 3 +++ 1 file changed, 3 insertions(+) diff --git a/flatpak/com.stremio.Stremio.Devel.json b/flatpak/com.stremio.Stremio.Devel.json index 0009e5c..d2563e4 100644 --- a/flatpak/com.stremio.Stremio.Devel.json +++ b/flatpak/com.stremio.Stremio.Devel.json @@ -31,6 +31,9 @@ "--persist=.stremio-server", "--filesystem=xdg-download", "--filesystem=xdg-run/pipewire-0:ro", + "--filesystem=xdg-run/discord-ipc-0", + "--filesystem=xdg-run/app/com.discordapp.Discord:create", + "--filesystem=xdg-run/app/dev.vencord.Vesktop:create", "--talk-name=org.kde.StatusNotifierWatcher", "--own-name=com.stremio.Stremio.Devel", "--own-name=org.mpris.MediaPlayer2.Stremio", From e13af63e2471837c7148594d3a1e5a34e3f06cbf Mon Sep 17 00:00:00 2001 From: Christopher Nelson Date: Sat, 18 Jul 2026 18:39:47 -0400 Subject: [PATCH 3/4] ci: run Rust tests --- .github/workflows/build.yml | 3 +++ 1 file changed, 3 insertions(+) diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 2c1f1f1..8d1863c 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -33,5 +33,8 @@ jobs: - name: Lint clippy run: cargo clippy --all --no-deps -- -D warnings + - name: Test + run: cargo test + - name: Build run: cargo build \ No newline at end of file From 5093194d62deb309a8db998a0bc8a833f8ca58af Mon Sep 17 00:00:00 2001 From: Christopher Nelson Date: Sat, 18 Jul 2026 18:39:52 -0400 Subject: [PATCH 4/4] build(flatpak): stop build script on failure --- flatpak/build.sh | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/flatpak/build.sh b/flatpak/build.sh index 949925f..276e899 100755 --- a/flatpak/build.sh +++ b/flatpak/build.sh @@ -1,9 +1,11 @@ #!/bin/sh +set -eu + app_id="com.stremio.Stremio.Devel" cwd="flatpak" python3 $cwd/flatpak-builder-tools/cargo/flatpak-cargo-generator.py Cargo.lock -o $cwd/cargo-sources.json flatpak-builder --repo=$cwd/repo --force-clean $cwd/build $cwd/$app_id.json -flatpak build-bundle $cwd/repo $cwd/$app_id.flatpak $app_id \ No newline at end of file +flatpak build-bundle $cwd/repo $cwd/$app_id.flatpak $app_id