koan_core/
graphql_client.rs

1//! GraphQL client for connecting to a `koan serve` instance.
2//!
3//! Uses blocking reqwest — call from a background thread when used from the TUI.
4
5use std::sync::Arc;
6
7use parking_lot::Mutex;
8use reqwest::StatusCode;
9use serde_json::Value;
10
11use crate::config::Config;
12
13/// A GraphQL client that talks to a koan server.
14///
15/// Clones share one session, so a token refreshed by one thread is the token
16/// every other thread uses next.
17#[derive(Clone)]
18pub struct GraphQLClient {
19    url: String,
20    http: reqwest::blocking::Client,
21    session: Option<Arc<Session>>,
22    /// Cleared once the server turns out to predate play modes, after which
23    /// `nowPlaying` is asked for without them. Shared by clones, like the
24    /// session.
25    play_modes: Arc<std::sync::atomic::AtomicBool>,
26    /// The same for the sleep timer.
27    sleep_timer: Arc<std::sync::atomic::AtomicBool>,
28}
29
30/// A sign-in to a server with auth enabled: the refresh token `koan auth login`
31/// stored, and the access token it last bought.
32struct Session {
33    tokens: Mutex<Tokens>,
34    /// Called with each new refresh token. The server spends the old one on
35    /// every refresh, so a copy not written back is dead.
36    on_rotate: Box<dyn Fn(&str) + Send + Sync>,
37}
38
39struct Tokens {
40    access: Option<String>,
41    refresh: String,
42    /// Why the server refused the refresh token. Set once and kept: the token
43    /// will not become valid again, and the bridge polls several times a second.
44    refused: Option<String>,
45}
46
47impl GraphQLClient {
48    pub fn new(server_url: &str) -> Self {
49        let url = format!("{}/graphql", server_url.trim_end_matches('/'));
50        Self {
51            url,
52            http: reqwest::blocking::Client::builder()
53                .timeout(std::time::Duration::from_secs(30))
54                .build()
55                .expect("failed to build HTTP client"),
56            session: None,
57            play_modes: Arc::new(std::sync::atomic::AtomicBool::new(true)),
58            sleep_timer: Arc::new(std::sync::atomic::AtomicBool::new(true)),
59        }
60    }
61
62    /// A client carrying the sign-in stored in `[auth]`, when it is for this
63    /// server. Rotated refresh tokens are written back through `Config::persist`.
64    pub fn from_config(server_url: &str) -> Self {
65        let client = Self::new(server_url);
66        let cfg = Config::load().unwrap_or_default();
67        if cfg.auth.refresh_token.is_empty()
68            || cfg.auth.server.trim_end_matches('/') != client.server_url()
69        {
70            return client;
71        }
72        client.with_session(cfg.auth.refresh_token, |token| {
73            if let Err(e) = Config::persist(|cfg| cfg.auth.refresh_token = token.to_owned()) {
74                log::warn!("could not store the rotated refresh token: {e}");
75            }
76        })
77    }
78
79    /// Authenticate with `refresh_token`, exchanging it for an access token
80    /// when the server first refuses a request.
81    pub fn with_session(
82        mut self,
83        refresh_token: impl Into<String>,
84        on_rotate: impl Fn(&str) + Send + Sync + 'static,
85    ) -> Self {
86        self.session = Some(Arc::new(Session {
87            tokens: Mutex::new(Tokens {
88                access: None,
89                refresh: refresh_token.into(),
90                refused: None,
91            }),
92            on_rotate: Box::new(on_rotate),
93        }));
94        self
95    }
96
97    /// Whether requests carry a stored sign-in.
98    pub fn has_session(&self) -> bool {
99        self.session.is_some()
100    }
101
102    /// Execute a raw GraphQL query/mutation.
103    ///
104    /// Access tokens are short-lived, so a 401 is answered by one refresh and
105    /// one retry; a second refusal is `GraphQLError::Unauthorized`.
106    pub fn execute(&self, query: &str, variables: Option<Value>) -> Result<Value, GraphQLError> {
107        let mut body = serde_json::json!({ "query": query });
108        if let Some(vars) = variables {
109            body["variables"] = vars;
110        }
111
112        let access = self
113            .session
114            .as_ref()
115            .and_then(|s| s.tokens.lock().access.clone());
116        let mut resp = self.post(&body, access.as_deref())?;
117        if resp.status() == StatusCode::UNAUTHORIZED
118            && let Some(session) = &self.session
119        {
120            let fresh = self.refresh(session, access.as_deref())?;
121            resp = self.post(&body, Some(&fresh))?;
122        }
123
124        let status = resp.status();
125        if status == StatusCode::UNAUTHORIZED {
126            return Err(GraphQLError::Unauthorized(reason(resp)));
127        }
128        if !status.is_success() {
129            return Err(GraphQLError::Http(format!("{status}: {}", reason(resp))));
130        }
131
132        let resp: Value = resp.json().map_err(|e| GraphQLError::Http(e.to_string()))?;
133
134        if let Some(errors) = resp.get("errors")
135            && let Some(arr) = errors.as_array()
136            && !arr.is_empty()
137        {
138            let msg = arr[0]
139                .get("message")
140                .and_then(|m| m.as_str())
141                .unwrap_or("unknown error");
142            return Err(GraphQLError::Query(msg.to_string()));
143        }
144
145        Ok(resp.get("data").cloned().unwrap_or(Value::Null))
146    }
147
148    fn post(
149        &self,
150        body: &Value,
151        access: Option<&str>,
152    ) -> Result<reqwest::blocking::Response, GraphQLError> {
153        let mut req = self.http.post(&self.url).json(body);
154        if let Some(token) = access {
155            req = req.bearer_auth(token);
156        }
157        req.send().map_err(|e| GraphQLError::Http(e.to_string()))
158    }
159
160    /// A new access token, replacing `stale`.
161    ///
162    /// The lock is held across the request: the server spends a refresh token
163    /// on use, so two threads refreshing at once would leave one holding a
164    /// revoked token. A thread that waited finds `stale` already replaced and
165    /// takes the new token instead.
166    fn refresh(&self, session: &Session, stale: Option<&str>) -> Result<String, GraphQLError> {
167        let mut tokens = session.tokens.lock();
168        if let Some(access) = &tokens.access
169            && Some(access.as_str()) != stale
170        {
171            return Ok(access.clone());
172        }
173        if let Some(reason) = &tokens.refused {
174            return Err(GraphQLError::Unauthorized(reason.clone()));
175        }
176
177        let resp = self
178            .http
179            .post(format!("{}/auth/refresh", self.server_url()))
180            .json(&serde_json::json!({ "refresh_token": tokens.refresh }))
181            .send()
182            .map_err(|e| GraphQLError::Http(e.to_string()))?;
183        if resp.status() == StatusCode::UNAUTHORIZED {
184            let refused = format!("the stored sign-in was refused: {}", reason(resp));
185            tokens.access = None;
186            tokens.refused = Some(refused.clone());
187            return Err(GraphQLError::Unauthorized(refused));
188        }
189        if !resp.status().is_success() {
190            let status = resp.status();
191            return Err(GraphQLError::Http(format!(
192                "/auth/refresh {status}: {}",
193                reason(resp)
194            )));
195        }
196
197        let body: Value = resp.json().map_err(|e| GraphQLError::Http(e.to_string()))?;
198        let (Some(access), Some(refresh)) = (
199            body["access_token"].as_str(),
200            body["refresh_token"].as_str(),
201        ) else {
202            return Err(GraphQLError::Http(
203                "malformed /auth/refresh response".into(),
204            ));
205        };
206        tokens.access = Some(access.to_owned());
207        tokens.refresh = refresh.to_owned();
208        (session.on_rotate)(refresh);
209        Ok(access.to_owned())
210    }
211
212    // -----------------------------------------------------------------------
213    // Typed helpers
214    // -----------------------------------------------------------------------
215
216    /// What the server is playing. A server older than play modes or the
217    /// sleep timer rejects a query naming them, so on that refusal the query
218    /// is asked again without them, and from then on, and they read as off.
219    pub fn now_playing(&self) -> Result<NowPlaying, GraphQLError> {
220        use std::sync::atomic::Ordering;
221        const TRACK: &str = "track { trackId title artist album codec sampleRate bitDepth \
222                             bitrateKbps channels durationMs }";
223        let ask = |modes: bool, sleep: bool| {
224            let modes = if modes { "shuffle repeat " } else { "" };
225            let sleep = if sleep {
226                "sleep { endsAtMs endOf } "
227            } else {
228                ""
229            };
230            self.execute(
231                &format!(
232                    "{{ nowPlaying {{ state positionMs durationMs queueItemId {modes}{sleep}{TRACK} }} }}"
233                ),
234                None,
235            )
236        };
237        let data = loop {
238            let (modes, sleep) = (
239                self.play_modes.load(Ordering::Relaxed),
240                self.sleep_timer.load(Ordering::Relaxed),
241            );
242            match ask(modes, sleep) {
243                Err(GraphQLError::Query(e)) if sleep && e.contains("sleep") => {
244                    log::info!("server predates the sleep timer: {e}");
245                    self.sleep_timer.store(false, Ordering::Relaxed);
246                }
247                Err(GraphQLError::Query(e))
248                    if modes && (e.contains("shuffle") || e.contains("repeat")) =>
249                {
250                    log::info!("server predates play modes: {e}");
251                    self.play_modes.store(false, Ordering::Relaxed);
252                }
253                other => break other?,
254            }
255        };
256        let np = &data["nowPlaying"];
257        Ok(NowPlaying {
258            state: np["state"].as_str().unwrap_or("STOPPED").to_string(),
259            position_ms: np["positionMs"].as_u64().unwrap_or(0),
260            duration_ms: np["durationMs"].as_u64(),
261            queue_item_id: np["queueItemId"].as_str().map(String::from),
262            mode: crate::player::state::PlayMode {
263                shuffle: np["shuffle"].as_bool().unwrap_or(false),
264                repeat: np["repeat"]
265                    .as_str()
266                    .and_then(|r| crate::player::state::Repeat::parse(&r.to_lowercase()))
267                    .unwrap_or_default(),
268            },
269            sleep: match (
270                np["sleep"]["endsAtMs"].as_u64(),
271                np["sleep"]["endOf"].as_str(),
272            ) {
273                (Some(unix_ms), _) => Some(crate::player::state::Sleep::At { unix_ms }),
274                (None, Some("TRACK")) => Some(crate::player::state::Sleep::EndOfTrack),
275                (None, Some("RECORD")) => Some(crate::player::state::Sleep::EndOfRecord),
276                _ => None,
277            },
278            track: np.get("track").and_then(|t| {
279                if t.is_null() {
280                    return None;
281                }
282                Some(NowPlayingTrack {
283                    track_id: t["trackId"].as_str().map(String::from),
284                    title: t["title"].as_str().unwrap_or("").to_string(),
285                    artist: t["artist"].as_str().unwrap_or("").to_string(),
286                    album: t["album"].as_str().unwrap_or("").to_string(),
287                    codec: t["codec"].as_str().unwrap_or("").to_string(),
288                    sample_rate: t["sampleRate"].as_u64().unwrap_or(0) as u32,
289                    bit_depth: t["bitDepth"].as_u64().map(|v| v as u16),
290                    bitrate_kbps: t["bitrateKbps"].as_u64().map(|v| v as u32),
291                    channels: t["channels"].as_u64().unwrap_or(0) as u16,
292                    duration_ms: t["durationMs"].as_u64().unwrap_or(0),
293                })
294            }),
295        })
296    }
297
298    pub fn queue(&self) -> Result<Vec<QueueEntry>, GraphQLError> {
299        let data = self.execute(
300            "{ queue { queueItemId trackId title artist album codec trackNumber disc durationMs isCurrent } }",
301            None,
302        )?;
303        let entries = data["queue"]
304            .as_array()
305            .map(|arr| {
306                arr.iter()
307                    .map(|e| QueueEntry {
308                        queue_item_id: e["queueItemId"].as_str().unwrap_or("").to_string(),
309                        track_id: e["trackId"].as_str().map(String::from),
310                        title: e["title"].as_str().unwrap_or("").to_string(),
311                        artist: e["artist"].as_str().unwrap_or("").to_string(),
312                        album: e["album"].as_str().unwrap_or("").to_string(),
313                        codec: e["codec"].as_str().map(String::from),
314                        track_number: e["trackNumber"].as_i64(),
315                        disc: e["disc"].as_i64(),
316                        duration_ms: e["durationMs"].as_u64(),
317                        is_current: e["isCurrent"].as_bool().unwrap_or(false),
318                    })
319                    .collect()
320            })
321            .unwrap_or_default();
322        Ok(entries)
323    }
324
325    // -- Mutations --
326
327    pub fn pause(&self) -> Result<(), GraphQLError> {
328        self.execute("mutation { pause { ok } }", None)?;
329        Ok(())
330    }
331
332    pub fn resume(&self) -> Result<(), GraphQLError> {
333        self.execute("mutation { resume { ok } }", None)?;
334        Ok(())
335    }
336
337    pub fn stop(&self) -> Result<(), GraphQLError> {
338        self.execute("mutation { stop { ok } }", None)?;
339        Ok(())
340    }
341
342    pub fn next(&self) -> Result<(), GraphQLError> {
343        self.execute("mutation { next { ok } }", None)?;
344        Ok(())
345    }
346
347    pub fn previous(&self) -> Result<(), GraphQLError> {
348        self.execute("mutation { previous { ok } }", None)?;
349        Ok(())
350    }
351
352    pub fn set_shuffle(&self, on: bool) -> Result<(), GraphQLError> {
353        self.execute(
354            "mutation($on: Boolean!) { setPlayMode(shuffle: $on) { ok } }",
355            Some(serde_json::json!({ "on": on })),
356        )?;
357        Ok(())
358    }
359
360    /// Set the server's sleep timer, or with `None` cancel it.
361    pub fn set_sleep_timer(
362        &self,
363        timer: Option<crate::player::state::SleepTimer>,
364    ) -> Result<(), GraphQLError> {
365        use crate::player::state::SleepTimer;
366        let (query, vars) = match timer {
367            None => ("mutation { cancelSleepTimer { ok } }", None),
368            Some(SleepTimer::After { minutes }) => (
369                "mutation($m: Int!) { setSleepTimer(minutes: $m) { ok } }",
370                Some(serde_json::json!({ "m": minutes })),
371            ),
372            Some(SleepTimer::EndOfTrack) => {
373                ("mutation { setSleepTimer(endOf: TRACK) { ok } }", None)
374            }
375            Some(SleepTimer::EndOfRecord) => {
376                ("mutation { setSleepTimer(endOf: RECORD) { ok } }", None)
377            }
378        };
379        self.execute(query, vars)?;
380        Ok(())
381    }
382
383    pub fn set_repeat(&self, repeat: crate::player::state::Repeat) -> Result<(), GraphQLError> {
384        self.execute(
385            "mutation($repeat: Repeat!) { setPlayMode(repeat: $repeat) { ok } }",
386            Some(serde_json::json!({ "repeat": repeat.as_str().to_uppercase() })),
387        )?;
388        Ok(())
389    }
390
391    pub fn seek(&self, position_ms: u64) -> Result<(), GraphQLError> {
392        self.execute(
393            "mutation($positionMs: Int!) { seek(positionMs: $positionMs) { ok } }",
394            Some(serde_json::json!({ "positionMs": position_ms })),
395        )?;
396        Ok(())
397    }
398
399    pub fn play(&self, queue_item_id: &str) -> Result<(), GraphQLError> {
400        self.execute(
401            "mutation($queueItemId: String!) { play(queueItemId: $queueItemId) { ok } }",
402            Some(serde_json::json!({ "queueItemId": queue_item_id })),
403        )?;
404        Ok(())
405    }
406
407    pub fn clear_queue(&self) -> Result<(), GraphQLError> {
408        self.execute("mutation { clearQueue { ok } }", None)?;
409        Ok(())
410    }
411
412    pub fn library_stats(&self) -> Result<Value, GraphQLError> {
413        self.execute(
414            "{ libraryStats { totalTracks totalArtists totalAlbums localTracks remoteTracks cachedTracks } }",
415            None,
416        )
417    }
418
419    /// Server URL (without /graphql path).
420    pub fn server_url(&self) -> &str {
421        self.url.trim_end_matches("/graphql")
422    }
423}
424
425// ---------------------------------------------------------------------------
426// Result types
427// ---------------------------------------------------------------------------
428
429#[derive(Debug, thiserror::Error)]
430pub enum GraphQLError {
431    #[error("http error: {0}")]
432    Http(String),
433    #[error("query error: {0}")]
434    Query(String),
435    #[error("unauthorised: {0}")]
436    Unauthorized(String),
437}
438
439/// What the server said when it refused a request: the `message` of a JSON
440/// body, or the body as text.
441fn reason(resp: reqwest::blocking::Response) -> String {
442    let status = resp.status();
443    let text = resp.text().unwrap_or_default();
444    let message = serde_json::from_str::<Value>(&text)
445        .ok()
446        .and_then(|v| v["message"].as_str().map(str::to_owned))
447        .unwrap_or(text);
448    if message.trim().is_empty() {
449        status.to_string()
450    } else {
451        message.trim().to_owned()
452    }
453}
454
455#[derive(Debug, Clone)]
456pub struct NowPlaying {
457    pub state: String,
458    pub position_ms: u64,
459    pub duration_ms: Option<u64>,
460    pub queue_item_id: Option<String>,
461    pub mode: crate::player::state::PlayMode,
462    pub sleep: Option<crate::player::state::Sleep>,
463    pub track: Option<NowPlayingTrack>,
464}
465
466#[derive(Debug, Clone)]
467pub struct NowPlayingTrack {
468    /// The track's id on the server. `None` for a queue entry the server built
469    /// from a file with no database row, which cannot be streamed.
470    pub track_id: Option<String>,
471    pub title: String,
472    pub artist: String,
473    pub album: String,
474    pub codec: String,
475    pub sample_rate: u32,
476    pub bit_depth: Option<u16>,
477    pub bitrate_kbps: Option<u32>,
478    pub channels: u16,
479    pub duration_ms: u64,
480}
481
482#[derive(Debug, Clone)]
483pub struct QueueEntry {
484    pub queue_item_id: String,
485    pub track_id: Option<String>,
486    pub title: String,
487    pub artist: String,
488    pub album: String,
489    pub codec: Option<String>,
490    pub track_number: Option<i64>,
491    pub disc: Option<i64>,
492    pub duration_ms: Option<u64>,
493    pub is_current: bool,
494}
495
496#[cfg(test)]
497mod tests {
498    use super::*;
499
500    #[test]
501    fn client_constructs_url() {
502        let c = GraphQLClient::new("http://localhost:4000");
503        assert_eq!(c.url, "http://localhost:4000/graphql");
504    }
505
506    #[test]
507    fn client_trailing_slash() {
508        let c = GraphQLClient::new("http://localhost:4000/");
509        assert_eq!(c.url, "http://localhost:4000/graphql");
510    }
511
512    /// A request as the test server saw it.
513    struct Seen {
514        path: String,
515        bearer: Option<String>,
516        body: String,
517    }
518
519    /// An HTTP server on a loopback port answering each request with
520    /// `respond`. Returns its base URL and every request it received.
521    fn serve(
522        respond: impl Fn(&Seen) -> (u16, &'static str) + Send + 'static,
523    ) -> (String, Arc<Mutex<Vec<String>>>) {
524        use std::io::{BufRead, BufReader, Read, Write};
525
526        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
527        let base = format!("http://{}", listener.local_addr().unwrap());
528        let log = Arc::new(Mutex::new(Vec::new()));
529        let seen_log = log.clone();
530        std::thread::spawn(move || {
531            for stream in listener.incoming() {
532                let Ok(mut stream) = stream else { return };
533                let mut reader = BufReader::new(stream.try_clone().unwrap());
534                let mut line = String::new();
535                reader.read_line(&mut line).unwrap();
536                let path = line.split_whitespace().nth(1).unwrap_or("").to_owned();
537                let (mut bearer, mut length) = (None, 0);
538                loop {
539                    line.clear();
540                    reader.read_line(&mut line).unwrap();
541                    let header = line.trim_end();
542                    if header.is_empty() {
543                        break;
544                    }
545                    let (name, value) = header.split_once(": ").unwrap_or((header, ""));
546                    match name.to_ascii_lowercase().as_str() {
547                        "authorization" => {
548                            bearer = value.strip_prefix("Bearer ").map(str::to_owned)
549                        }
550                        "content-length" => length = value.parse().unwrap_or(0),
551                        _ => {}
552                    }
553                }
554                let mut body = vec![0; length];
555                reader.read_exact(&mut body).unwrap();
556                let seen = Seen {
557                    path,
558                    bearer,
559                    body: String::from_utf8_lossy(&body).into_owned(),
560                };
561                seen_log.lock().push(seen.path.clone());
562                let (code, reply) = respond(&seen);
563                write!(
564                    stream,
565                    "HTTP/1.1 {code} X\r\nContent-Type: application/json\r\n\
566                     Content-Length: {}\r\nConnection: close\r\n\r\n{reply}",
567                    reply.len()
568                )
569                .unwrap();
570            }
571        });
572        (base, log)
573    }
574
575    /// A server that takes `access-2` and trades `refresh-1` for it once.
576    fn signed_in_server(seen: &Seen) -> (u16, &'static str) {
577        match seen.path.as_str() {
578            "/graphql" if seen.bearer.as_deref() == Some("access-2") => {
579                (200, r#"{"data":{"ok":true}}"#)
580            }
581            "/auth/refresh" if seen.body.contains("refresh-1") => (
582                200,
583                r#"{"access_token":"access-2","refresh_token":"refresh-2"}"#,
584            ),
585            "/auth/refresh" => (401, r#"{"message":"invalid or expired refresh token"}"#),
586            _ => (401, "missing or invalid Authorization header"),
587        }
588    }
589
590    #[test]
591    fn refused_request_refreshes_once_and_retries() {
592        let (base, log) = serve(signed_in_server);
593        let rotated = Arc::new(Mutex::new(Vec::new()));
594        let stored = rotated.clone();
595        let client = GraphQLClient::new(&base)
596            .with_session("refresh-1", move |t| stored.lock().push(t.to_owned()));
597
598        let data = client.execute("{ ok }", None).unwrap();
599        assert_eq!(data["ok"], true);
600        assert_eq!(*rotated.lock(), ["refresh-2"]);
601
602        // The new access token is kept: no second refresh.
603        client.clone().execute("{ ok }", None).unwrap();
604        assert_eq!(
605            *log.lock(),
606            ["/graphql", "/auth/refresh", "/graphql", "/graphql"]
607        );
608    }
609
610    #[test]
611    fn refused_refresh_is_unauthorised() {
612        let (base, log) = serve(signed_in_server);
613        let client = GraphQLClient::new(&base).with_session("revoked", |_| {});
614
615        for _ in 0..2 {
616            match client.execute("{ ok }", None) {
617                Err(GraphQLError::Unauthorized(msg)) => {
618                    assert!(msg.contains("invalid or expired refresh token"), "{msg}")
619                }
620                other => panic!("expected Unauthorized, got {other:?}"),
621            }
622        }
623        // A refused refresh token is not offered again.
624        assert_eq!(*log.lock(), ["/graphql", "/auth/refresh", "/graphql"]);
625    }
626
627    #[test]
628    fn unauthorised_without_a_session_says_why() {
629        let (base, log) = serve(signed_in_server);
630
631        match GraphQLClient::new(&base).execute("{ ok }", None) {
632            Err(GraphQLError::Unauthorized(msg)) => {
633                assert_eq!(msg, "missing or invalid Authorization header")
634            }
635            other => panic!("expected Unauthorized, got {other:?}"),
636        }
637        assert_eq!(*log.lock(), ["/graphql"]);
638    }
639}
koan_core/graphql_client.rs

koan_core/
graphql_client.rs
