1use std::collections::HashMap;
11use std::net::TcpStream;
12use std::os::fd::RawFd;
13use std::path::Path;
14use std::sync::atomic::{AtomicU64, Ordering};
15use std::sync::{Arc, LazyLock};
16use std::time::{Duration, Instant};
17
18use parking_lot::{Condvar, Mutex};
19use serde::{Deserialize, Serialize};
20use tungstenite::stream::MaybeTlsStream;
21
22use crate::config::{self, Config};
23use crate::helpers::{subsonic_auth, subsonic_client};
24use crate::player::state::QueueMode;
25use crate::remote::client::SubsonicAuth;
26pub use crate::remote::outputs::{LinkOutput, LinkOutputs, OutputChoice};
27use crate::remote::profile;
28use crate::remote::wire::{self, Waker};
29
30#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
32#[serde(tag = "type", rename_all = "camelCase")]
33pub enum LinkCommand {
34 #[serde(rename_all = "camelCase")]
37 Play {
38 track_ids: Vec<String>,
39 #[serde(default)]
40 start_at: u32,
41 #[serde(default, skip_serializing_if = "is_zero")]
42 position_ms: u64,
43 #[serde(default, skip_serializing_if = "std::ops::Not::not")]
44 paused: bool,
45 #[serde(default, skip_serializing_if = "std::ops::Not::not")]
51 handoff: bool,
52 #[serde(default, skip_serializing_if = "QueueMode::is_keep")]
55 mode: QueueMode,
56 },
57 #[serde(rename_all = "camelCase")]
60 Enqueue {
61 track_ids: Vec<String>,
62 #[serde(default, skip_serializing_if = "std::ops::Not::not")]
63 play: bool,
64 },
65 #[serde(rename_all = "camelCase")]
67 PlayNext {
68 track_ids: Vec<String>,
69 },
70 #[serde(rename_all = "camelCase")]
72 Remove {
73 track_ids: Vec<String>,
74 },
75 Clear,
76 Sync {
80 #[serde(default)]
81 full: bool,
82 },
83 #[serde(rename_all = "camelCase")]
86 Evict {
87 track_ids: Vec<String>,
88 },
89 #[serde(rename_all = "camelCase")]
92 JumpTo {
93 track_id: String,
94 },
95 #[serde(rename_all = "camelCase")]
96 Seek {
97 position_ms: u64,
98 },
99 Pause,
100 Resume,
101 Next,
102 Previous,
103 PlayItem {
107 id: String,
108 },
109 RemoveItems {
110 ids: Vec<String>,
111 },
112 MoveItems {
114 ids: Vec<String>,
115 target: String,
116 after: bool,
117 },
118 #[serde(rename_all = "camelCase")]
120 Insert {
121 track_ids: Vec<String>,
122 after: String,
123 },
124 Undo,
125 Redo,
126 Shuffle {
129 on: bool,
130 },
131 Repeat {
133 mode: crate::player::state::Repeat,
134 },
135 SleepTimer {
137 timer: Option<crate::player::state::SleepTimer>,
138 },
139 HandOff {
143 to: String,
144 },
145 Devices {
148 devices: Vec<LinkDevice>,
149 },
150 DeviceKeys {
156 keys: Vec<LinkDeviceKey>,
157 },
158 Shares {
161 grantees: Vec<String>,
162 #[serde(default, skip_serializing_if = "Option::is_none")]
164 error: Option<String>,
165 #[serde(default, skip_serializing_if = "Vec::is_empty")]
167 accounts: Vec<String>,
168 },
169 Shared {
172 command: Box<LinkCommand>,
173 },
174 Forgotten {
177 device: String,
178 },
179 HistoryChanged,
183 DspProfilesChanged,
187 SetOutput {
190 output: OutputChoice,
191 },
192 RefreshOutputs,
195 SetRendererVolume {
197 volume: u8,
198 },
199 SetPreset {
202 device: String,
203 profile: Option<String>,
204 },
205 WatchLevels {
209 on: bool,
210 },
211 Levels {
214 from: String,
215 f: crate::remote::levels::Frame,
216 },
217 Acked {
220 from: String,
221 ack: u64,
222 outcome: crate::remote::acks::AckOutcome,
223 },
224}
225
226fn is_zero(n: &u64) -> bool {
227 *n == 0
228}
229
230impl LinkCommand {
231 pub fn allowed_playback(&self) -> bool {
242 match self {
243 Self::Play { .. }
244 | Self::Enqueue { .. }
245 | Self::PlayNext { .. }
246 | Self::Remove { .. }
247 | Self::Clear
248 | Self::JumpTo { .. }
249 | Self::Seek { .. }
250 | Self::Pause
251 | Self::Resume
252 | Self::Next
253 | Self::Previous
254 | Self::PlayItem { .. }
255 | Self::RemoveItems { .. }
256 | Self::MoveItems { .. }
257 | Self::Insert { .. }
258 | Self::Undo
259 | Self::Redo
260 | Self::Shuffle { .. }
261 | Self::Repeat { .. }
262 | Self::SleepTimer { .. }
263 | Self::HandOff { .. }
264 | Self::SetOutput { .. }
265 | Self::RefreshOutputs
266 | Self::SetRendererVolume { .. }
267 | Self::SetPreset { .. }
268 | Self::WatchLevels { .. } => true,
269 Self::Sync { .. }
271 | Self::Evict { .. }
272 | Self::Devices { .. }
273 | Self::DeviceKeys { .. }
274 | Self::Shares { .. }
275 | Self::Shared { .. }
276 | Self::Forgotten { .. }
277 | Self::HistoryChanged
278 | Self::DspProfilesChanged
279 | Self::Levels { .. }
280 | Self::Acked { .. } => false,
281 }
282 }
283
284 pub fn from_the_network(&self, full: bool) -> Option<CommandSource> {
288 if full {
289 self.allowed_playback().then_some(CommandSource::Nearby)
290 } else {
291 self.allowed_nearby().then_some(CommandSource::Stranger)
292 }
293 }
294
295 pub fn relayable(&self) -> bool {
302 match self {
303 Self::Play { .. }
304 | Self::Enqueue { .. }
305 | Self::PlayNext { .. }
306 | Self::Remove { .. }
307 | Self::Clear
308 | Self::Sync { .. }
309 | Self::Evict { .. }
310 | Self::JumpTo { .. }
311 | Self::Seek { .. }
312 | Self::Pause
313 | Self::Resume
314 | Self::Next
315 | Self::Previous
316 | Self::PlayItem { .. }
317 | Self::RemoveItems { .. }
318 | Self::MoveItems { .. }
319 | Self::Insert { .. }
320 | Self::Undo
321 | Self::Redo
322 | Self::Shuffle { .. }
323 | Self::Repeat { .. }
324 | Self::SleepTimer { .. }
325 | Self::HandOff { .. }
326 | Self::SetOutput { .. }
327 | Self::RefreshOutputs
328 | Self::SetRendererVolume { .. }
329 | Self::SetPreset { .. } => true,
330 Self::Devices { .. }
331 | Self::DeviceKeys { .. }
332 | Self::Acked { .. }
333 | Self::Shares { .. }
334 | Self::Shared { .. }
335 | Self::Forgotten { .. }
336 | Self::HistoryChanged
337 | Self::DspProfilesChanged
338 | Self::WatchLevels { .. }
339 | Self::Levels { .. } => false,
340 }
341 }
342
343 pub fn live_only(&self) -> bool {
347 match self {
348 Self::Shared { command } => command.live_only(),
349 cmd => matches!(cmd, Self::WatchLevels { .. } | Self::RefreshOutputs),
350 }
351 }
352
353 pub fn allowed_nearby(&self) -> bool {
363 match self {
364 Self::Play { .. }
365 | Self::Enqueue { .. }
366 | Self::PlayNext { .. }
367 | Self::Remove { .. }
368 | Self::Clear
369 | Self::JumpTo { .. }
370 | Self::Seek { .. }
371 | Self::Pause
372 | Self::Resume
373 | Self::Next
374 | Self::Previous
375 | Self::PlayItem { .. }
376 | Self::RemoveItems { .. }
377 | Self::MoveItems { .. }
378 | Self::Insert { .. }
379 | Self::Undo
380 | Self::Redo
381 | Self::Shuffle { .. }
382 | Self::Repeat { .. }
383 | Self::SleepTimer { .. }
384 | Self::HandOff { .. }
385 | Self::WatchLevels { .. } => true,
386 Self::Sync { .. }
387 | Self::Evict { .. }
388 | Self::Devices { .. }
389 | Self::DeviceKeys { .. }
390 | Self::Forgotten { .. }
391 | Self::HistoryChanged
392 | Self::DspProfilesChanged
393 | Self::Levels { .. }
394 | Self::Acked { .. }
395 | Self::SetOutput { .. }
396 | Self::RefreshOutputs
397 | Self::SetRendererVolume { .. }
398 | Self::SetPreset { .. }
399 | Self::Shares { .. }
400 | Self::Shared { .. } => false,
401 }
402 }
403
404 pub fn track_ids(&self) -> &[String] {
406 match self {
407 Self::Play { track_ids, .. }
408 | Self::Enqueue { track_ids, .. }
409 | Self::PlayNext { track_ids }
410 | Self::Remove { track_ids }
411 | Self::Evict { track_ids }
412 | Self::Insert { track_ids, .. } => track_ids,
413 Self::JumpTo { track_id } => std::slice::from_ref(track_id),
414 Self::Shared { command } => command.track_ids(),
415 _ => &[],
416 }
417 }
418}
419
420#[derive(Debug, Clone, Copy, PartialEq, Eq)]
422pub enum CommandSource {
423 Account,
426 Shared,
429 Nearby,
432 Stranger,
436}
437
438#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
440pub struct LinkDeviceKey {
441 pub id: String,
443 pub key: String,
445 #[serde(default, skip_serializing_if = "Option::is_none")]
448 pub owner: Option<String>,
449}
450
451pub fn valid_device_key(key: &str) -> bool {
454 use base64::Engine as _;
455 base64::engine::general_purpose::STANDARD
456 .decode(key)
457 .is_ok_and(|bytes| bytes.len() == 32)
458}
459
460#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
462#[serde(rename_all = "camelCase")]
463pub struct LinkDevice {
464 pub id: String,
466 pub name: String,
467 pub platform: String,
468 pub linked: bool,
471 pub state: Option<LinkState>,
474 #[serde(default, skip_serializing_if = "Option::is_none")]
476 pub last_seen: Option<i64>,
477 #[serde(default, skip_serializing_if = "Option::is_none")]
481 pub wakeable: Option<bool>,
482 #[serde(default, skip_serializing_if = "Option::is_none")]
485 pub owner: Option<String>,
486 #[serde(default, skip_serializing_if = "std::ops::Not::not")]
489 pub acks: bool,
490}
491
492impl LinkCommand {
493 pub fn track_ids_mut(&mut self) -> Vec<&mut String> {
496 match self {
497 Self::Play { track_ids, .. }
498 | Self::Enqueue { track_ids, .. }
499 | Self::PlayNext { track_ids }
500 | Self::Remove { track_ids }
501 | Self::Evict { track_ids }
502 | Self::Insert { track_ids, .. } => track_ids.iter_mut().collect(),
503 Self::JumpTo { track_id } => vec![track_id],
504 Self::Shared { command } => command.track_ids_mut(),
505 Self::PlayItem { .. }
506 | Self::RemoveItems { .. }
507 | Self::MoveItems { .. }
508 | Self::Undo
509 | Self::Redo
510 | Self::Shuffle { .. }
511 | Self::Repeat { .. }
512 | Self::SleepTimer { .. }
513 | Self::HandOff { .. }
514 | Self::Devices { .. }
515 | Self::DeviceKeys { .. }
516 | Self::Shares { .. }
517 | Self::Forgotten { .. }
518 | Self::HistoryChanged
519 | Self::DspProfilesChanged
520 | Self::WatchLevels { .. }
521 | Self::Levels { .. }
522 | Self::Acked { .. }
523 | Self::SetOutput { .. }
524 | Self::RefreshOutputs
525 | Self::SetRendererVolume { .. }
526 | Self::SetPreset { .. }
527 | Self::Clear
528 | Self::Sync { .. }
529 | Self::Seek { .. }
530 | Self::Pause
531 | Self::Resume
532 | Self::Next
533 | Self::Previous => Vec::new(),
534 }
535 }
536}
537
538#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
540#[serde(rename_all = "camelCase")]
541pub struct LinkState {
542 pub playing: bool,
543 pub title: Option<String>,
544 pub artist: Option<String>,
545 #[serde(default)]
546 pub album: Option<String>,
547 #[serde(default)]
549 pub position_ms: u64,
550 #[serde(default)]
551 pub duration_ms: u64,
552 #[serde(default)]
554 pub queue: Vec<LinkQueueEntry>,
555 #[serde(default, skip_serializing_if = "Option::is_none")]
557 pub outputs: Option<LinkOutputs>,
558 #[serde(default, skip_serializing_if = "std::ops::Not::not")]
559 pub shuffle: bool,
560 #[serde(default, skip_serializing_if = "crate::player::state::Repeat::is_off")]
561 pub repeat: crate::player::state::Repeat,
562 #[serde(default, skip_serializing_if = "Option::is_none")]
563 pub sleep: Option<crate::player::state::Sleep>,
564 #[serde(default, skip_serializing_if = "std::ops::Not::not")]
566 pub sleep_fading: bool,
567}
568
569#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
570#[serde(rename_all = "camelCase")]
571pub struct LinkQueueEntry {
572 #[serde(default)]
574 pub id: Option<String>,
575 pub track_id: Option<String>,
577 pub title: String,
578 pub artist: String,
579 #[serde(default)]
580 pub album: String,
581 #[serde(default)]
582 pub duration_ms: u64,
583 pub current: bool,
584}
585
586impl LinkState {
587 pub fn differs(&self, sent: &LinkState, elapsed: Duration) -> bool {
591 let strip = |s: &LinkState| LinkState {
592 position_ms: 0,
593 ..s.clone()
594 };
595 if strip(self) != strip(sent) {
596 return true;
597 }
598 let expected = if sent.playing {
599 sent.position_ms + elapsed.as_millis() as u64
600 } else {
601 sent.position_ms
602 };
603 self.position_ms.abs_diff(expected) > 3000
604 }
605}
606
607#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
609#[serde(tag = "type", rename_all = "camelCase")]
610pub enum LinkReport {
611 State(LinkState),
612 Push {
616 token: String,
617 sandbox: bool,
618 },
619 Command {
623 to: String,
624 command: LinkCommand,
625 #[serde(default, skip_serializing_if = "Option::is_none")]
626 ack: Option<u64>,
627 },
628 Received {
632 ack: u64,
633 },
634 Ack {
636 ack: u64,
637 outcome: crate::remote::acks::AckOutcome,
638 },
639 Activity {
642 token: Option<String>,
643 device: Option<String>,
644 #[serde(default)]
645 sandbox: bool,
646 },
647 Hello(LinkHello),
649 Wake {
652 to: String,
653 #[serde(default)]
654 notify: bool,
655 },
656 Share {
659 grantee: String,
660 allow: bool,
661 },
662 Forget {
665 device: String,
666 },
667 Levels {
670 f: crate::remote::levels::Frame,
671 },
672}
673
674#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
677#[serde(rename_all = "camelCase")]
678pub struct LinkHello {
679 pub id: String,
680 pub name: String,
681 pub platform: String,
682 pub library: Option<String>,
686 #[serde(default)]
689 pub acks: bool,
690 #[serde(default, skip_serializing_if = "Option::is_none")]
693 pub nonce: Option<String>,
694}
695
696pub fn library_fingerprint(cfg: &Config) -> Option<String> {
699 let auth = subsonic_auth(cfg)?;
700 let url = auth.base_url.trim_end_matches('/').to_ascii_lowercase();
701 Some(format!("{:x}", md5::compute(url.as_bytes())))
702}
703
704static PUSH_TOKEN: Mutex<Option<(String, bool)>> = Mutex::new(None);
707
708type ActivityToken = (String, String, bool);
712static ACTIVITY: Mutex<Option<Option<ActivityToken>>> = Mutex::new(None);
713
714pub fn set_activity(activity: Option<ActivityToken>) {
717 *ACTIVITY.lock() = Some(activity);
718 if let Some(up) = LINK.lock().as_ref() {
719 up.waker.wake();
720 }
721}
722
723pub fn parse_command(json: &str) -> Result<LinkCommand, String> {
726 serde_json::from_str(json).map_err(|e| e.to_string())
727}
728
729pub fn set_push_token(token: String, sandbox: bool) {
731 *PUSH_TOKEN.lock() = Some((token, sandbox));
732 nudge();
733 if let Some(up) = LINK.lock().as_ref() {
734 up.waker.wake();
735 }
736}
737
738#[derive(Debug, Clone)]
740pub struct LinkIdentity {
741 pub name: String,
743 pub platform: String,
745 pub device_id: String,
748 pub files: &'static str,
752}
753
754impl LinkIdentity {
755 pub fn this_device(name: Option<String>) -> Self {
757 let (platform, label) = if cfg!(target_os = "ios") {
758 ("ios", "iPhone")
759 } else if cfg!(target_os = "tvos") {
760 ("tvos", "Apple TV")
761 } else if cfg!(target_os = "macos") {
762 ("macos", "Mac")
763 } else if cfg!(target_os = "android") {
764 ("android", "Android")
765 } else {
766 ("linux", "Linux")
767 };
768 Self {
769 name: name
770 .filter(|n| !n.trim().is_empty())
771 .or_else(hostname)
772 .unwrap_or_else(|| label.to_string()),
773 platform: platform.to_string(),
774 device_id: device_id(&config::config_dir()),
775 files: "",
776 }
777 }
778
779 pub fn terminal() -> Option<Self> {
785 const FILES: &str = "terminal-";
786 static HELD: std::sync::OnceLock<std::fs::File> = std::sync::OnceLock::new();
789 let dir = config::config_dir();
790 if HELD.get().is_none() {
791 let lock = std::fs::OpenOptions::new()
792 .create(true)
793 .truncate(false)
794 .write(true)
795 .open(dir.join(format!("{FILES}device.lock")))
796 .ok()?;
797 lock.try_lock().ok()?;
798 let _ = HELD.set(lock);
799 }
800 let app = Self::this_device(None);
801 Some(Self {
802 name: format!("{} (terminal)", app.name),
803 device_id: file_device_id(&dir.join(format!("{FILES}device-id"))),
804 files: FILES,
805 ..app
806 })
807 }
808}
809
810#[derive(Clone)]
814pub struct Local {
815 pub identity: LinkIdentity,
816 pub state: Arc<dyn Fn() -> LinkState + Send + Sync>,
817 pub on_command:
820 Arc<dyn Fn(LinkCommand, CommandSource, Option<crate::remote::acks::Pending>) + Send + Sync>,
821}
822
823pub fn spawn(local: Local) {
832 std::thread::Builder::new()
833 .name("koan-link".into())
834 .spawn(move || run(local))
835 .expect("failed to spawn the link thread");
836}
837
838const RETRY_MIN: Duration = Duration::from_secs(2);
839const RETRY_MAX: Duration = Duration::from_secs(60);
840
841static SIGN_IN: AtomicU64 = AtomicU64::new(0);
844
845pub fn relink() {
849 SIGN_IN.fetch_add(1, Ordering::SeqCst);
850 if let Some(up) = LINK.lock().as_ref() {
851 up.waker.wake();
852 }
853 nudge();
854}
855
856fn run(local: Local) {
857 let mut wait = RETRY_MIN;
858 loop {
859 crate::quiet::wait_until_awake();
860 let signed_in = SIGN_IN.load(Ordering::SeqCst);
863 let cfg = Config::load().unwrap_or_default();
864 let account = crate::remote::proof::account_of(&cfg);
865 let Some(auth) = subsonic_auth(&cfg) else {
866 rest(RETRY_MAX);
867 continue;
868 };
869 match profile::for_auth(&auth) {
870 Some(p) if p.links() => {}
871 Some(_) => {
873 rest(RETRY_MAX);
874 continue;
875 }
876 None => {
877 rest(wait);
878 wait = (wait * 2).min(RETRY_MAX);
879 continue;
880 }
881 }
882
883 match connect(&auth, &local.identity) {
884 Ok((mut socket, fd)) => {
885 log::info!("link: connected to {}", auth.base_url);
886 if let Some(client) = crate::helpers::subsonic_client(&cfg) {
889 client.outage().retry_now();
890 }
891 wait = RETRY_MIN;
892 if let Err(e) = serve(&mut socket, fd, &local, signed_in, account) {
893 log::info!("link: closed: {e}");
894 }
895 *LINK.lock() = None;
896 crate::remote::devices::set_linked(false);
897 }
898 Err(e) => {
899 log::warn!("link: {e}");
900 profile::forget();
906 }
907 }
908 if rest(wait) {
909 wait = RETRY_MIN;
910 continue;
911 }
912 wait = (wait * 2).min(RETRY_MAX);
913 }
914}
915
916static NUDGE: (Mutex<bool>, Condvar) = (Mutex::new(false), Condvar::new());
917
918pub fn nudge() {
924 *NUDGE.0.lock() = true;
925 NUDGE.1.notify_all();
926}
927
928fn rest(d: Duration) -> bool {
930 let mut nudged = NUDGE.0.lock();
931 if !*nudged {
932 NUDGE.1.wait_for(&mut nudged, d);
933 }
934 std::mem::replace(&mut *nudged, false)
935}
936
937pub fn hang_up() {
939 if let Some(up) = LINK.lock().as_ref() {
940 up.waker.wake();
941 }
942}
943
944struct Up {
946 waker: Arc<Waker>,
947 outbox: Vec<LinkReport>,
948}
949
950static LINK: Mutex<Option<Up>> = Mutex::new(None);
951
952pub fn report(report: LinkReport) -> bool {
954 let mut link = LINK.lock();
955 let Some(up) = link.as_mut() else {
956 return false;
957 };
958 up.outbox.push(report);
959 up.waker.wake();
960 true
961}
962
963type Socket = tungstenite::WebSocket<MaybeTlsStream<TcpStream>>;
964
965fn connect(auth: &SubsonicAuth, identity: &LinkIdentity) -> Result<(Socket, RawFd), String> {
966 let url = link_url(auth, identity)?;
967 let (socket, _) = tungstenite::connect(url).map_err(|e| e.to_string())?;
968 let fd = wire::prepare(socket.get_ref())?;
969 Ok((socket, fd))
970}
971
972fn link_url(auth: &SubsonicAuth, identity: &LinkIdentity) -> Result<String, String> {
974 let base = if let Some(rest) = auth.base_url.strip_prefix("https://") {
975 format!("wss://{rest}")
976 } else if let Some(rest) = auth.base_url.strip_prefix("http://") {
977 format!("ws://{rest}")
978 } else {
979 return Err(format!("not an http(s) server: {}", auth.base_url));
980 };
981 let mut query = auth.query().map_err(|e| e.to_string())?;
982 let device_key = crate::remote::proof::public_key().unwrap_or_default();
986 for (k, v) in [
987 ("client", identity.name.as_str()),
988 ("platform", identity.platform.as_str()),
989 ("device", identity.device_id.as_str()),
990 ("devices", "1"),
993 ("acks", "1"),
996 ("deviceKey", device_key.as_str()),
997 ] {
998 if v.is_empty() {
999 continue;
1000 }
1001 query.push('&');
1002 query.push_str(k);
1003 query.push('=');
1004 query.push_str(&percent_encode(v));
1005 }
1006 Ok(format!("{base}/rest/koanLink?{query}"))
1007}
1008
1009fn serve(
1010 socket: &mut Socket,
1011 fd: RawFd,
1012 local: &Local,
1013 signed_in: u64,
1014 account: Option<String>,
1015) -> Result<(), String> {
1016 let waker = Waker::new().map_err(|e| e.to_string())?;
1017 let watcher = waker.clone();
1018 wire::wake_on_engine_change(&waker);
1019 *LINK.lock() = Some(Up {
1020 waker: waker.clone(),
1021 outbox: Vec::new(),
1022 });
1023 crate::remote::devices::set_linked(true);
1024 let mut session = LinkSession {
1025 local,
1026 sent: None,
1027 sent_push: None,
1028 sent_activity: None,
1029 waker: watcher,
1030 levels: None,
1031 signed_in,
1032 account,
1033 };
1034 wire::drive(socket, fd, &waker, &mut session)
1035}
1036
1037struct LinkSession<'a> {
1038 local: &'a Local,
1039 sent: Option<(LinkState, Instant)>,
1040 sent_push: Option<(String, bool)>,
1041 sent_activity: Option<Option<ActivityToken>>,
1042 waker: Arc<Waker>,
1043 levels: Option<crate::remote::levels::Watch>,
1046 signed_in: u64,
1048 account: Option<String>,
1051}
1052
1053impl wire::Session for LinkSession<'_> {
1054 fn outgoing(&mut self) -> Vec<String> {
1055 let mut out = Vec::new();
1056 let push = PUSH_TOKEN.lock().clone();
1057 if let Some((token, sandbox)) = push.clone()
1058 && push != self.sent_push
1059 {
1060 out.push(LinkReport::Push { token, sandbox });
1061 self.sent_push = push;
1062 }
1063 let activity = ACTIVITY.lock().clone();
1064 if activity.is_some() && activity != self.sent_activity {
1065 let (token, device, sandbox) = match activity.clone().flatten() {
1066 Some((t, d, s)) => (Some(t), Some(d), s),
1067 None => (None, None, false),
1068 };
1069 out.push(LinkReport::Activity {
1070 token,
1071 device,
1072 sandbox,
1073 });
1074 self.sent_activity = activity;
1075 }
1076 if let Some(up) = LINK.lock().as_mut() {
1077 out.append(&mut up.outbox);
1078 }
1079 let now = (self.local.state)();
1080 if self
1081 .sent
1082 .as_ref()
1083 .is_none_or(|(s, at)| now.differs(s, at.elapsed()))
1084 {
1085 out.push(LinkReport::State(now.clone()));
1086 self.sent = Some((now, Instant::now()));
1087 }
1088 if let Some(f) = self.levels.as_mut().and_then(|w| w.take()) {
1089 out.push(LinkReport::Levels { f });
1090 }
1091 out.iter()
1092 .filter_map(|r| serde_json::to_string(r).ok())
1093 .collect()
1094 }
1095
1096 fn incoming(&mut self, text: &str) {
1097 let envelope = match serde_json::from_str::<crate::remote::acks::Envelope>(text) {
1098 Ok(envelope) => envelope,
1099 Err(e) => {
1100 log::warn!("link: not a command ({e}): {text}");
1101 return;
1102 }
1103 };
1104 if let Some(ack) = envelope.ack {
1105 report(LinkReport::Received { ack });
1106 }
1107 let Some((command, pending)) = crate::remote::acks::take(envelope, |ack, outcome| {
1109 report(LinkReport::Ack { ack, outcome });
1110 }) else {
1111 return;
1112 };
1113 match command {
1114 LinkCommand::Acked { from, ack, outcome } => {
1115 crate::remote::acks::resolve(ack, &from, outcome);
1116 }
1117 LinkCommand::Devices { devices } => {
1118 crate::remote::devices::set_account(devices);
1119 }
1120 LinkCommand::DeviceKeys { keys } => {
1121 crate::remote::proof::keep(keys, self.account.clone());
1122 }
1123 LinkCommand::Shares {
1124 grantees,
1125 error,
1126 accounts,
1127 } => {
1128 crate::remote::devices::set_shares(grantees, error, accounts);
1129 }
1130 LinkCommand::Shared { command } => {
1131 if command.allowed_playback() {
1134 (self.local.on_command)(*command, CommandSource::Shared, pending);
1135 } else {
1136 log::warn!("link: refused from a shared account: {command:?}");
1137 if let Some(pending) = pending {
1138 pending.finish(crate::remote::acks::AckOutcome::Refused {
1139 reason: "not allowed from another account".into(),
1140 });
1141 }
1142 }
1143 }
1144 LinkCommand::Forgotten { device } => {
1145 crate::remote::devices::forgotten(&device);
1146 }
1147 LinkCommand::WatchLevels { on } => {
1148 self.levels = on.then(|| crate::remote::levels::feed().watch(&self.waker));
1149 }
1150 LinkCommand::Levels { from, f } => {
1151 crate::remote::levels::remote().received(&from, f);
1152 }
1153 cmd => (self.local.on_command)(cmd, CommandSource::Account, pending),
1154 }
1155 }
1156
1157 fn done(&self) -> bool {
1158 !crate::quiet::awake() || SIGN_IN.load(Ordering::SeqCst) != self.signed_in
1159 }
1160}
1161
1162pub fn resolve_tracks(
1173 db: &crate::db::connection::Database,
1174 remote_ids: &[String],
1175 may_sync: bool,
1176) -> (Vec<i64>, bool) {
1177 let (found, synced) = resolve_each(db, remote_ids, may_sync);
1178 (found.into_iter().flatten().collect(), synced)
1179}
1180
1181pub fn resolve_each(
1184 db: &crate::db::connection::Database,
1185 remote_ids: &[String],
1186 may_sync: bool,
1187) -> (Vec<Option<i64>>, bool) {
1188 let lookup = |db: &crate::db::connection::Database| {
1189 let mut stmt = db
1190 .conn
1191 .prepare_cached(
1192 "SELECT id FROM tracks WHERE uid = ?1
1193 UNION ALL SELECT id FROM tracks WHERE remote_id = ?1 LIMIT 1",
1194 )
1195 .ok();
1196 remote_ids
1197 .iter()
1198 .map(|rid| {
1199 stmt.as_mut()
1200 .and_then(|s| s.query_row([rid], |r| r.get::<_, i64>(0)).ok())
1201 })
1202 .collect::<Vec<_>>()
1203 };
1204 let found = lookup(db);
1205 let missing: Vec<&String> = remote_ids
1206 .iter()
1207 .zip(&found)
1208 .filter(|(_, f)| f.is_none())
1209 .map(|(id, _)| id)
1210 .collect();
1211 if missing.is_empty() || !may_sync || missing.iter().all(|id| recently_missed(id)) {
1212 return (found, false);
1213 }
1214 sync(db, crate::helpers::Walk::IfChanged);
1215 let found = lookup(db);
1216 let mut missed = MISSED.lock();
1217 let now = Instant::now();
1218 missed.retain(|_, at| now.duration_since(*at) < MISS_TTL);
1219 for (id, _) in remote_ids.iter().zip(&found).filter(|(_, f)| f.is_none()) {
1220 missed.insert(id.clone(), now);
1221 }
1222 (found, true)
1223}
1224
1225static MISSED: LazyLock<Mutex<HashMap<String, Instant>>> = LazyLock::new(Default::default);
1227const MISS_TTL: Duration = Duration::from_secs(300);
1228
1229fn recently_missed(id: &str) -> bool {
1230 MISSED
1231 .lock()
1232 .get(id)
1233 .is_some_and(|at| at.elapsed() < MISS_TTL)
1234}
1235
1236pub fn sync(db: &crate::db::connection::Database, walk: crate::helpers::Walk) {
1239 let cfg = Config::load().unwrap_or_default();
1240 if let Some(client) = subsonic_client(&cfg)
1241 && let Err(e) = crate::helpers::sync_remote(
1242 db,
1243 &client,
1244 walk,
1245 &cfg.remote.url,
1246 &cfg.remote.username,
1247 &|_| {},
1248 )
1249 {
1250 log::warn!("link: sync failed: {e}");
1251 }
1252}
1253
1254fn device_id(dir: &Path) -> String {
1258 #[cfg(any(target_os = "ios", target_os = "tvos"))]
1259 {
1260 use security_framework::passwords::{get_generic_password, set_generic_password};
1261 const SERVICE: &str = "cc.blit.koan.link";
1262 if let Some(id) = get_generic_password(SERVICE, "device-id")
1263 .ok()
1264 .and_then(|b| String::from_utf8(b).ok())
1265 .filter(|id| !id.trim().is_empty())
1266 {
1267 return id;
1268 }
1269 let id = file_device_id(&dir.join("device-id"));
1270 let _ = set_generic_password(SERVICE, "device-id", id.as_bytes());
1271 id
1272 }
1273 #[cfg(not(any(target_os = "ios", target_os = "tvos")))]
1274 file_device_id(&dir.join("device-id"))
1275}
1276
1277fn file_device_id(path: &Path) -> String {
1278 if let Ok(id) = std::fs::read_to_string(path) {
1279 let id = id.trim();
1280 if !id.is_empty() {
1281 return id.to_string();
1282 }
1283 }
1284 let id = uuid::Uuid::now_v7().to_string();
1285 if let Some(dir) = path.parent() {
1286 let _ = std::fs::create_dir_all(dir);
1287 }
1288 let _ = std::fs::write(path, &id);
1289 id
1290}
1291
1292fn hostname() -> Option<String> {
1293 let mut buf = [0u8; 256];
1294 let ok = unsafe { libc::gethostname(buf.as_mut_ptr().cast(), buf.len()) } == 0;
1297 if !ok {
1298 return None;
1299 }
1300 let end = buf.iter().position(|&b| b == 0).unwrap_or(buf.len());
1301 let name = String::from_utf8_lossy(&buf[..end]);
1302 let name = name.trim_end_matches(".local").trim();
1303 (!name.is_empty() && name != "localhost").then(|| name.to_string())
1304}
1305
1306fn percent_encode(s: &str) -> String {
1307 let mut out = String::with_capacity(s.len());
1308 for b in s.bytes() {
1309 if b.is_ascii_alphanumeric() || matches!(b, b'-' | b'_' | b'.' | b'~') {
1310 out.push(b as char);
1311 } else {
1312 out.push_str(&format!("%{b:02X}"));
1313 }
1314 }
1315 out
1316}
1317
1318#[cfg(test)]
1319mod tests {
1320
1321 #[test]
1322 fn levels_cross_the_link_compactly() {
1323 use crate::remote::levels::Frame;
1324 let f = Frame(61_250, 512, 300, 40);
1325
1326 let report = serde_json::to_string(&LinkReport::Levels { f }).unwrap();
1327 assert_eq!(report, r#"{"type":"levels","f":[61250,512,300,40]}"#);
1328 assert_eq!(
1329 serde_json::from_str::<LinkReport>(&report).unwrap(),
1330 LinkReport::Levels { f }
1331 );
1332
1333 let relayed = LinkCommand::Levels {
1334 from: "dev-phone".into(),
1335 f,
1336 };
1337 let text = serde_json::to_string(&relayed).unwrap();
1338 assert!(text.len() < 80, "{} bytes: {text}", text.len());
1339 assert_eq!(serde_json::from_str::<LinkCommand>(&text).unwrap(), relayed);
1340
1341 let watch = LinkCommand::WatchLevels { on: true };
1342 let text = serde_json::to_string(&watch).unwrap();
1343 assert_eq!(serde_json::from_str::<LinkCommand>(&text).unwrap(), watch);
1344 assert!(watch.allowed_nearby(), "a stranger may watch the bars");
1345 assert!(!relayed.allowed_nearby(), "only the server relays frames");
1346 }
1347 use super::*;
1348
1349 #[test]
1350 fn a_sleep_timer_travels_as_playback_and_comes_back_in_the_state() {
1351 use crate::player::state::{Sleep, SleepTimer};
1352 let cmd: LinkCommand =
1353 serde_json::from_str(r#"{"type":"sleepTimer","timer":{"kind":"after","minutes":30}}"#)
1354 .unwrap();
1355 assert_eq!(
1356 cmd,
1357 LinkCommand::SleepTimer {
1358 timer: Some(SleepTimer::After { minutes: 30 })
1359 }
1360 );
1361 let cancel: LinkCommand =
1362 serde_json::from_str(r#"{"type":"sleepTimer","timer":null}"#).unwrap();
1363 assert_eq!(cancel, LinkCommand::SleepTimer { timer: None });
1364 for cmd in [cmd, cancel] {
1365 assert!(cmd.allowed_playback() && cmd.allowed_nearby(), "{cmd:?}");
1366 }
1367
1368 let state = LinkState {
1369 sleep: Some(Sleep::EndOfRecord),
1370 ..Default::default()
1371 };
1372 let json = serde_json::to_string(&state).unwrap();
1373 assert!(json.contains(r#""sleep":{"kind":"endOfRecord"}"#), "{json}");
1374 assert_eq!(serde_json::from_str::<LinkState>(&json).unwrap(), state);
1375 assert!(
1376 !serde_json::to_string(&LinkState::default())
1377 .unwrap()
1378 .contains("sleep"),
1379 "nothing said with none set"
1380 );
1381 }
1382
1383 #[test]
1387 fn the_playback_set_is_playback_and_nothing_of_the_library() {
1388 let ids = vec!["t".to_string()];
1389 for cmd in [
1390 LinkCommand::Pause,
1391 LinkCommand::JumpTo {
1392 track_id: "t".into(),
1393 },
1394 LinkCommand::Enqueue {
1395 track_ids: ids.clone(),
1396 play: false,
1397 },
1398 LinkCommand::HandOff { to: "x".into() },
1399 LinkCommand::SetRendererVolume { volume: 1 },
1400 ] {
1401 assert!(cmd.allowed_playback(), "{cmd:?}");
1402 assert_eq!(cmd.from_the_network(true), Some(CommandSource::Nearby));
1403 }
1404 for cmd in [
1405 LinkCommand::Sync { full: false },
1406 LinkCommand::Evict {
1407 track_ids: ids.clone(),
1408 },
1409 LinkCommand::Devices { devices: vec![] },
1410 LinkCommand::Shares {
1411 grantees: vec![],
1412 error: None,
1413 accounts: vec![],
1414 },
1415 LinkCommand::Shared {
1416 command: Box::new(LinkCommand::Pause),
1417 },
1418 ] {
1419 assert!(!cmd.allowed_playback(), "{cmd:?}");
1420 assert_eq!(cmd.from_the_network(true), None, "{cmd:?}");
1421 assert_eq!(cmd.from_the_network(false), None, "{cmd:?}");
1422 }
1423 assert_eq!(
1425 LinkCommand::Pause.from_the_network(false),
1426 Some(CommandSource::Stranger)
1427 );
1428 assert_eq!(
1429 LinkCommand::SetRendererVolume { volume: 1 }.from_the_network(false),
1430 None
1431 );
1432 }
1433
1434 #[test]
1437 fn unknown_ids_sync_only_when_allowed_and_not_recently_missed() {
1438 let dir = tempfile::tempdir().unwrap();
1439 let db = crate::db::connection::Database::open(&dir.path().join("koan.db")).unwrap();
1440
1441 let (found, synced) = resolve_tracks(&db, &["from-a-stranger".into()], false);
1442 assert!(found.is_empty());
1443 assert!(!synced, "a nearby peer's unknown id must not start a sync");
1444
1445 MISSED
1446 .lock()
1447 .insert("deleted-on-server".into(), Instant::now());
1448 let (_, synced) = resolve_tracks(&db, &["deleted-on-server".into()], true);
1449 assert!(
1450 !synced,
1451 "an id a sync just failed to find does not start another"
1452 );
1453 }
1454
1455 #[test]
1456 fn commands_name_their_tracks() {
1457 let play: LinkCommand =
1458 serde_json::from_str(r#"{"type":"play","trackIds":["a","b"]}"#).unwrap();
1459 assert_eq!(play.track_ids(), ["a", "b"]);
1460 let jump: LinkCommand = serde_json::from_str(r#"{"type":"jumpTo","trackId":"c"}"#).unwrap();
1461 assert_eq!(jump.track_ids(), ["c"]);
1462 assert!(LinkCommand::Pause.track_ids().is_empty());
1463 }
1464
1465 #[test]
1466 fn a_push_token_is_a_tagged_report() {
1467 let report = LinkReport::Push {
1468 token: "ab12".into(),
1469 sandbox: true,
1470 };
1471 let text = serde_json::to_string(&report).unwrap();
1472 assert_eq!(text, r#"{"type":"push","token":"ab12","sandbox":true}"#);
1473 assert_eq!(serde_json::from_str::<LinkReport>(&text).unwrap(), report);
1474 }
1475
1476 #[test]
1477 fn commands_are_tagged_json() {
1478 let play = LinkCommand::Play {
1479 track_ids: vec!["12".into(), "34".into()],
1480 start_at: 1,
1481 position_ms: 0,
1482 paused: false,
1483 handoff: false,
1484 mode: QueueMode::Keep,
1485 };
1486 let json = serde_json::to_string(&play).unwrap();
1487 assert_eq!(
1488 json,
1489 r#"{"type":"play","trackIds":["12","34"],"startAt":1}"#
1490 );
1491 assert_eq!(serde_json::from_str::<LinkCommand>(&json).unwrap(), play);
1492 let held = LinkCommand::Play {
1493 track_ids: vec!["12".into()],
1494 start_at: 0,
1495 position_ms: 61_250,
1496 paused: true,
1497 handoff: false,
1498 mode: QueueMode::Keep,
1499 };
1500 let json = serde_json::to_string(&held).unwrap();
1501 assert_eq!(
1502 json,
1503 r#"{"type":"play","trackIds":["12"],"startAt":0,"positionMs":61250,"paused":true}"#
1504 );
1505 assert_eq!(serde_json::from_str::<LinkCommand>(&json).unwrap(), held);
1506 assert_eq!(
1507 serde_json::from_str::<LinkCommand>(r#"{"type":"pause"}"#).unwrap(),
1508 LinkCommand::Pause
1509 );
1510 let report = LinkReport::State(LinkState {
1511 playing: true,
1512 title: Some("Portions for Foxes".into()),
1513 ..Default::default()
1514 });
1515 let json = serde_json::to_string(&report).unwrap();
1516 assert!(json.starts_with(r#"{"type":"state","playing":true,"title":"Portions for Foxes""#));
1517 assert_eq!(serde_json::from_str::<LinkReport>(&json).unwrap(), report);
1518 }
1519
1520 #[test]
1521 fn a_playhead_moving_on_time_is_not_news() {
1522 let sent = LinkState {
1523 playing: true,
1524 position_ms: 10_000,
1525 ..Default::default()
1526 };
1527 let later = |pos| LinkState {
1528 position_ms: pos,
1529 ..sent.clone()
1530 };
1531 let five = Duration::from_secs(5);
1532 assert!(!later(15_000).differs(&sent, five));
1533 assert!(later(60_000).differs(&sent, five), "a seek");
1534 let paused = LinkState {
1535 playing: false,
1536 ..later(15_000)
1537 };
1538 assert!(paused.differs(&sent, five));
1539 }
1540
1541 #[test]
1542 fn the_url_follows_the_scheme_and_names_the_device() {
1543 let identity = LinkIdentity {
1544 name: "J's iPhone".into(),
1545 platform: "ios".into(),
1546 device_id: "abc".into(),
1547 files: "",
1548 };
1549 let url = link_url(
1550 &SubsonicAuth::new("https://music.example.com", "j", "pw"),
1551 &identity,
1552 )
1553 .unwrap();
1554 assert!(url.starts_with("wss://music.example.com/rest/koanLink?"));
1555 assert!(url.contains("client=J%27s%20iPhone"));
1556 assert!(url.contains("device=abc"));
1557 assert!(url.contains("devices=1"));
1558 assert!(
1559 link_url(&SubsonicAuth::new("http://h:4000", "j", "pw"), &identity)
1560 .unwrap()
1561 .starts_with("ws://h:4000/")
1562 );
1563 }
1564
1565 #[test]
1566 fn a_relayed_command_nests_the_command() {
1567 let report = LinkReport::Command {
1568 to: "phone".into(),
1569 command: LinkCommand::HandOff { to: "mac".into() },
1570 ack: None,
1571 };
1572 let json = serde_json::to_string(&report).unwrap();
1573 assert_eq!(
1574 json,
1575 r#"{"type":"command","to":"phone","command":{"type":"handOff","to":"mac"}}"#
1576 );
1577 assert_eq!(serde_json::from_str::<LinkReport>(&json).unwrap(), report);
1578 }
1579
1580 #[test]
1581 fn an_older_queue_entry_still_reads() {
1582 let e: LinkQueueEntry =
1583 serde_json::from_str(r#"{"trackId":"7","title":"t","artist":"a","current":true}"#)
1584 .unwrap();
1585 assert_eq!(e.id, None);
1586 assert_eq!(e.duration_ms, 0);
1587 }
1588
1589 #[test]
1590 fn strangers_cannot_touch_the_library() {
1591 assert!(LinkCommand::Pause.allowed_nearby());
1592 assert!(LinkCommand::HandOff { to: "x".into() }.allowed_nearby());
1593 assert!(!LinkCommand::Sync { full: true }.allowed_nearby());
1594 assert!(!LinkCommand::Evict { track_ids: vec![] }.allowed_nearby());
1595 }
1596
1597 #[test]
1598 fn the_device_id_is_kept() {
1599 let dir = tempfile::tempdir().unwrap();
1600 let first = device_id(dir.path());
1601 assert_eq!(device_id(dir.path()), first);
1602 }
1603}
1604
1605#[cfg(test)]
1606mod device_key_tests {
1607 use super::*;
1608
1609 #[test]
1613 fn device_keys_come_from_the_server_alone() {
1614 let cmd = LinkCommand::DeviceKeys { keys: vec![] };
1615 assert!(!cmd.allowed_nearby());
1616 assert!(!cmd.allowed_playback());
1617 assert_eq!(cmd.from_the_network(true), None);
1618 assert_eq!(cmd.from_the_network(false), None);
1619 }
1620
1621 #[test]
1622 fn the_server_never_relays_its_own_news() {
1623 for news in [
1624 LinkCommand::DeviceKeys { keys: vec![] },
1625 LinkCommand::Devices { devices: vec![] },
1626 LinkCommand::HistoryChanged,
1627 LinkCommand::DspProfilesChanged,
1628 LinkCommand::Shared {
1629 command: Box::new(LinkCommand::Pause),
1630 },
1631 ] {
1632 assert!(!news.relayable(), "{news:?}");
1633 }
1634 assert!(LinkCommand::Pause.relayable());
1635 }
1636
1637 #[test]
1638 fn a_device_key_is_32_bytes_of_base64() {
1639 use base64::Engine as _;
1640 let b64 = base64::engine::general_purpose::STANDARD;
1641 assert!(valid_device_key(&b64.encode([7u8; 32])));
1642 assert!(!valid_device_key(&b64.encode([7u8; 31])));
1643 assert!(!valid_device_key(&b64.encode([7u8; 33])));
1644 assert!(!valid_device_key("not base64!"));
1645 assert!(!valid_device_key(""));
1646 }
1647}