koan_server/graphql/

mod.rs

mod helpers;
pub(crate) mod jobs;
mod loaders;
mod mutations;
mod queries;
mod server;
mod subscriptions;
mod types;

use std::ops::Deref;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;

use async_graphql::dataloader::DataLoader;
use async_graphql::{Context, Schema};
use crossbeam_channel::Sender;
use koan_core::audio::viz::VizSnapshot;
use koan_core::db::connection::Database;
use koan_core::db::pool::Pool;
use koan_core::db::queries::{UidKind, is_uid};
use koan_core::player::commands::PlayerCommand;
use koan_core::player::state::{QueueItemId, SharedPlayerState};
use uuid::Uuid;

use koan_core::auth::Role;
use loaders::DbLoader;
use mutations::MutationRoot;
use queries::QueryRoot;
pub use server::{
    ApiServerOpts, cmd_serve, cmd_serve_daemon, execute_in_process, start_api_background,
};
use subscriptions::SubscriptionRoot;

use crate::auth::AuthUser;

// ---------------------------------------------------------------------------
// Connection pool
// ---------------------------------------------------------------------------

/// How long the dataloader gathers keys before running a batch.
///
/// The default 1ms closes the window while a wide selection set is still
/// registering keys, splitting one query into several.
const BATCH_WINDOW: Duration = Duration::from_millis(10);

/// How long a resolver waits for a free connection before giving up. Long
/// enough to ride out a scan chunk, short enough that a wedged writer surfaces
/// as an error rather than a hung request.
const ACQUIRE_TIMEOUT: Duration = Duration::from_secs(10);

/// The process's connections, handed out per query rather than per field, and
/// at most `max` at once.
///
/// The connections are the server's own pool, shared with Subsonic, the web UI
/// and MCP: one set of page caches for the one database, rather than one per
/// front end. What this adds is the bound. A resolver past it waits up to
/// `ACQUIRE_TIMEOUT` for another to finish, so a query fanning out wide cannot
/// take a connection per field, and a wedged writer surfaces as an error rather
/// than a hung request.
struct DbPool {
    pool: Arc<Pool>,
    /// One token per connection that may be out at once. Taken to acquire,
    /// returned on drop.
    permits_tx: Sender<()>,
    permits_rx: crossbeam_channel::Receiver<()>,
    /// Dataloader batches run. Instrumentation only — the N+1 tests assert one
    /// batch serves a whole selection set.
    batches: AtomicUsize,
}

impl DbPool {
    fn new(pool: Arc<Pool>) -> Arc<Self> {
        // SQLite readers scale with cores; past that they only queue on the OS.
        let max = std::thread::available_parallelism()
            .map(|n| n.get())
            .unwrap_or(4)
            .clamp(4, 16);
        let (permits_tx, permits_rx) = crossbeam_channel::bounded(max);
        for _ in 0..max {
            let _ = permits_tx.try_send(());
        }
        Arc::new(Self {
            pool,
            permits_tx,
            permits_rx,
            batches: AtomicUsize::new(0),
        })
    }

    fn acquire(self: &Arc<Self>) -> async_graphql::Result<PooledDb> {
        self.permits_rx
            .recv_timeout(ACQUIRE_TIMEOUT)
            .map_err(|_| async_graphql::Error::new("database busy"))?;
        match self.pool.take() {
            Ok(db) => Ok(PooledDb {
                db: Some(db),
                pool: self.clone(),
            }),
            Err(e) => {
                let _ = self.permits_tx.try_send(());
                Err(internal_error("db open", e))
            }
        }
    }
}

/// A connection checked out of the pool, returned when the guard drops.
struct PooledDb {
    db: Option<Database>,
    pool: Arc<DbPool>,
}

impl Deref for PooledDb {
    type Target = Database;

    fn deref(&self) -> &Database {
        self.db.as_ref().expect("connection taken before drop")
    }
}

impl Drop for PooledDb {
    fn drop(&mut self) {
        if let Some(db) = self.db.take() {
            self.pool.pool.put_back(db);
        }
        let _ = self.pool.permits_tx.try_send(());
    }
}

// ---------------------------------------------------------------------------
// DB handle wrapper (so we can put it in Context)
// ---------------------------------------------------------------------------

#[derive(Clone)]
struct DbHandle {
    pool: Arc<DbPool>,
}

impl DbHandle {
    fn new(pool: Arc<Pool>) -> Self {
        Self {
            pool: DbPool::new(pool),
        }
    }

    fn acquire(&self) -> async_graphql::Result<PooledDb> {
        self.pool.acquire()
    }

    /// The database this schema serves, which need not be the default one.
    fn path(&self) -> std::path::PathBuf {
        self.pool.pool.path().to_owned()
    }

    /// A connection outside the bound, for work that runs for minutes and must
    /// not deny a connection to request-path resolvers.
    fn open_detached(&self) -> Result<Database, koan_core::db::connection::DbError> {
        self.pool.pool.take()
    }

    fn note_batch(&self) {
        self.pool.batches.fetch_add(1, Ordering::Relaxed);
    }

    /// Connections the underlying pool has opened.
    #[cfg_attr(not(test), allow(dead_code))]
    fn open_count(&self) -> usize {
        self.pool.pool.opened()
    }

    /// Dataloader batches run since the schema was built.
    #[cfg_attr(not(test), allow(dead_code))]
    fn batch_count(&self) -> usize {
        self.pool.batches.load(Ordering::Relaxed)
    }
}

/// Run a rusqlite closure on the blocking pool with a pooled connection.
///
/// rusqlite is blocking start to finish. Calling it inline in an `async fn`
/// parks a tokio worker for the duration, and enough of those at once starve
/// the runtime — including the `ReaderStream`s feeding in-flight audio.
async fn with_db<T, F>(ctx: &Context<'_>, f: F) -> async_graphql::Result<T>
where
    F: FnOnce(&Database) -> async_graphql::Result<T> + Send + 'static,
    T: Send + 'static,
{
    let handle = ctx.data::<DbHandle>()?.clone();
    blocking(move || {
        let db = handle.acquire()?;
        f(&db)
    })
    .await
}

/// A row as its published id: its uid, the same on the server and every device
/// synced from it. See `koan_core::db::queries::uids`.
async fn uid(
    ctx: &Context<'_>,
    kind: UidKind,
    id: i64,
) -> async_graphql::Result<async_graphql::ID> {
    Ok(async_graphql::ID(
        id_loader(ctx)?
            .load_one(loaders::UidOf(kind, id))
            .await?
            .unwrap_or_else(|| id.to_string()),
    ))
}

async fn uids(
    ctx: &Context<'_>,
    kind: UidKind,
    ids: &[i64],
) -> async_graphql::Result<Vec<async_graphql::ID>> {
    let found = id_loader(ctx)?
        .load_many(ids.iter().map(|&id| loaders::UidOf(kind, id)))
        .await?;
    Ok(ids
        .iter()
        .map(|&id| {
            async_graphql::ID(
                found
                    .get(&loaders::UidOf(kind, id))
                    .cloned()
                    .unwrap_or_else(|| id.to_string()),
            )
        })
        .collect())
}

async fn opt_uid(
    ctx: &Context<'_>,
    kind: UidKind,
    id: Option<i64>,
) -> async_graphql::Result<Option<async_graphql::ID>> {
    match id {
        Some(id) => uid(ctx, kind, id).await.map(Some),
        None => Ok(None),
    }
}

/// The row an id argument names: its uid, or the bare row id that callers
/// from before uids still send.
async fn row_id(ctx: &Context<'_>, kind: UidKind, id: &str) -> async_graphql::Result<i64> {
    Ok(row_ids(ctx, kind, &[id]).await?[0])
}

/// `row_id` for each, in one lookup.
async fn row_ids<S: AsRef<str> + Sync>(
    ctx: &Context<'_>,
    kind: UidKind,
    ids: &[S],
) -> async_graphql::Result<Vec<i64>> {
    let loader = id_loader(ctx)?;
    let uids = ids
        .iter()
        .map(AsRef::as_ref)
        .filter(|id| id.parse::<i64>().is_err())
        .map(|id| {
            if is_uid(id) {
                Ok(loaders::RowOf(kind, id.to_string()))
            } else {
                Err(async_graphql::Error::new(format!("not an id: {id}")))
            }
        })
        .collect::<async_graphql::Result<Vec<_>>>()?;
    let found = loader.load_many(uids).await?;
    ids.iter()
        .map(AsRef::as_ref)
        .map(|id| match id.parse::<i64>() {
            Ok(row) => Ok(row),
            Err(_) => found
                .get(&loaders::RowOf(kind, id.to_string()))
                .copied()
                .ok_or_else(|| {
                    async_graphql::Error::new(format!("no {kind:?} with id {id}").to_lowercase())
                }),
        })
        .collect()
}

async fn opt_row_id(
    ctx: &Context<'_>,
    kind: UidKind,
    id: Option<&async_graphql::ID>,
) -> async_graphql::Result<Option<i64>> {
    match id {
        Some(id) => row_id(ctx, kind, id).await.map(Some),
        None => Ok(None),
    }
}

async fn opt_row_ids(
    ctx: &Context<'_>,
    kind: UidKind,
    ids: Option<&[async_graphql::ID]>,
) -> async_graphql::Result<Option<Vec<i64>>> {
    match ids {
        Some(ids) => row_ids(ctx, kind, ids).await.map(Some),
        None => Ok(None),
    }
}

fn id_loader<'a>(ctx: &'a Context<'_>) -> async_graphql::Result<&'a DataLoader<DbLoader>> {
    ctx.data::<DataLoader<DbLoader>>()
}

/// Run any other blocking work (HTTP fetches, file decoding, tag reads) off the
/// async workers.
async fn blocking<T, F>(f: F) -> async_graphql::Result<T>
where
    F: FnOnce() -> async_graphql::Result<T> + Send + 'static,
    T: Send + 'static,
{
    tokio::task::spawn_blocking(f)
        .await
        .map_err(|e| internal_error("blocking task", e))?
}

/// Log the detail, return a generic message.
///
/// SQLite errors quote the offending statement and filesystem errors quote
/// absolute paths — a map of the host handed to whoever asked.
pub(super) fn internal_error(context: &str, e: impl std::fmt::Display) -> async_graphql::Error {
    log::error!("graphql {}: {}", context, e);
    async_graphql::Error::new("internal error")
}

// ---------------------------------------------------------------------------
// Schema builder
// ---------------------------------------------------------------------------

pub type KoanSchema = Schema<QueryRoot, MutationRoot, SubscriptionRoot>;

pub fn build_schema(
    state: Arc<SharedPlayerState>,
    cmd_tx: Sender<PlayerCommand>,
    pool: Arc<Pool>,
    viz: Option<Arc<VizSnapshot>>,
) -> KoanSchema {
    build_schema_with(DbHandle::new(pool), state, cmd_tx, viz)
}

/// `build_schema`, with its jobs kept in `jobs`, which the web UI shares.
pub(crate) fn build_schema_sharing(
    state: Arc<SharedPlayerState>,
    cmd_tx: Sender<PlayerCommand>,
    pool: Arc<Pool>,
    viz: Option<Arc<VizSnapshot>>,
    jobs: jobs::JobRegistry,
) -> KoanSchema {
    builder_with(DbHandle::new(pool), state, cmd_tx, viz, jobs).finish()
}

/// `build_schema` with `extension` running on every request.
pub(crate) fn build_schema_extended(
    state: Arc<SharedPlayerState>,
    cmd_tx: Sender<PlayerCommand>,
    pool: Arc<Pool>,
    extension: impl async_graphql::extensions::ExtensionFactory,
) -> KoanSchema {
    builder_with(DbHandle::new(pool), state, cmd_tx, None, Default::default())
        .extension(extension)
        .finish()
}

fn build_schema_with(
    handle: DbHandle,
    state: Arc<SharedPlayerState>,
    cmd_tx: Sender<PlayerCommand>,
    viz: Option<Arc<VizSnapshot>>,
) -> KoanSchema {
    builder_with(handle, state, cmd_tx, viz, Default::default()).finish()
}

fn builder_with(
    handle: DbHandle,
    state: Arc<SharedPlayerState>,
    cmd_tx: Sender<PlayerCommand>,
    viz: Option<Arc<VizSnapshot>>,
    jobs: jobs::JobRegistry,
) -> async_graphql::SchemaBuilder<QueryRoot, MutationRoot, SubscriptionRoot> {
    // Batching only, no caching: a schema-wide loader outlives the request, and
    // favourites and library contents change under it.
    let loader = DataLoader::new(DbLoader::new(handle.clone()), tokio::spawn).delay(BATCH_WINDOW);
    loader.enable_all_cache(false);

    let mut builder = Schema::build(QueryRoot, MutationRoot, SubscriptionRoot)
        .data(handle)
        .data(loader)
        .data(jobs)
        .data(Arc::new(queries::Fuzzy::default()))
        .data(state)
        .data(cmd_tx);
    if let Some(viz) = viz {
        builder = builder.data(viz);
    }
    // A single nested query can otherwise fan out across the whole library and
    // pin the process for minutes.
    builder.limit_depth(12).limit_complexity(2000)
}

// ---------------------------------------------------------------------------
// Shared helpers used by queries + mutations
// ---------------------------------------------------------------------------

fn parse_queue_item_id(s: &str) -> async_graphql::Result<QueueItemId> {
    Uuid::parse_str(s)
        .map(QueueItemId)
        .map_err(|e| async_graphql::Error::new(format!("invalid queue item ID '{}': {}", s, e)))
}

/// How long to wait for room on the player command channel.
///
/// The channel is bounded(16) and the player can sit inside `start_playback`
/// for about a second during a device sample-rate change. A blocking `send`
/// there parks a tokio worker; this gives the player a moment to drain and
/// then reports back rather than holding the thread.
const CMD_SEND_TIMEOUT: Duration = Duration::from_millis(250);

fn send_cmd(ctx: &Context<'_>, cmd: PlayerCommand) -> async_graphql::Result<()> {
    let tx = ctx.data::<Sender<PlayerCommand>>()?;
    send_cmd_via(tx, cmd)
}

fn send_cmd_via(tx: &Sender<PlayerCommand>, cmd: PlayerCommand) -> async_graphql::Result<()> {
    tx.send_timeout(cmd, CMD_SEND_TIMEOUT)
        .map_err(|_| async_graphql::Error::new("player busy — command not accepted"))
}

/// The authenticated user, or an anonymous admin when there is none (auth
/// disabled, or an in-process caller).
fn get_auth_user(ctx: &Context<'_>) -> AuthUser {
    ctx.data::<AuthUser>()
        .cloned()
        .unwrap_or_else(|_| AuthUser::anonymous_admin())
}

/// Whose favourites, playlists and history the current request reads and writes.
fn user_id(ctx: &Context<'_>) -> i64 {
    get_auth_user(ctx).user_id
}

/// Whose shares the current user may list and change: their own, or
/// everyone's for an admin.
fn share_owner(ctx: &Context<'_>) -> Option<i64> {
    let user = get_auth_user(ctx);
    (user.role != Role::Admin).then_some(user.user_id)
}

/// A playlist `user` may see: their own, or anyone's public one. Anyone
/// else's is not there, as far as they can tell.
fn readable_playlist(
    db: &Database,
    user: i64,
    id: i64,
) -> async_graphql::Result<koan_core::db::queries::PlaylistRow> {
    use koan_core::db::queries;
    let me = queries::auth::resolve_user(&db.conn, user).map_err(|e| internal_error("db", e))?;
    queries::get_playlist(&db.conn, id)
        .map_err(|e| internal_error("db", e))?
        .filter(|p| p.readable_by(me))
        .ok_or_else(|| async_graphql::Error::new(format!("playlist {id} not found")))
}

/// A playlist `user` may change: their own only.
fn editable_playlist(
    db: &Database,
    user: i64,
    id: i64,
) -> async_graphql::Result<koan_core::db::queries::PlaylistRow> {
    let list = readable_playlist(db, user, id)?;
    let me = koan_core::db::queries::auth::resolve_user(&db.conn, user)
        .map_err(|e| internal_error("db", e))?;
    if list.editable_by(me) {
        Ok(list)
    } else {
        Err(async_graphql::Error::new(format!(
            "playlist {id} belongs to someone else"
        )))
    }
}

/// A playlist whose contents `user` may change: their own, and not a smart
/// one.
fn fillable_playlist(
    db: &Database,
    user: i64,
    id: i64,
) -> async_graphql::Result<koan_core::db::queries::PlaylistRow> {
    let list = editable_playlist(db, user, id)?;
    if list.readonly {
        return Err(async_graphql::Error::new(format!(
            "playlist {id} is read-only: its rules (setPlaylistRules changes \
             them), its file in the library or its server decide what it holds"
        )));
    }
    Ok(list)
}

/// A playlist `user` may rename or give rules: their own, and not one read
/// from a file in the library, which decides both.
fn renamable_playlist(
    db: &Database,
    user: i64,
    id: i64,
) -> async_graphql::Result<koan_core::db::queries::PlaylistRow> {
    let list = editable_playlist(db, user, id)?;
    if let Some(path) = &list.source_path {
        return Err(async_graphql::Error::new(format!(
            "playlist {id} is read from {path}: edit the file instead"
        )));
    }
    Ok(list)
}

/// Evaluate the smart playlists `user` can see that are due, and have every
/// device pull any whose contents moved.
fn refresh_smart(db: &Database, user: i64) {
    match koan_core::db::queries::smart::refresh_due(&db.conn, user) {
        Ok(changed) if !changed.is_empty() => crate::clients::changed(),
        Ok(_) => {}
        Err(e) => log::warn!("smart playlists not refreshed: {e}"),
    }
}

/// Whose linked clients the current user may see and command: their own,
/// whatever their role.
fn client_scope(ctx: &Context<'_>) -> Option<String> {
    Some(get_auth_user(ctx).username)
}

/// Check that the current user has at least the required role.
/// Returns an error suitable for GraphQL if the check fails.
fn require_role(ctx: &Context<'_>, required: Role) -> async_graphql::Result<()> {
    let user = get_auth_user(ctx);
    if user.role.has_permission(required) {
        Ok(())
    } else {
        Err(async_graphql::Error::new(format!(
            "forbidden: requires {} role, you have {}",
            required, user.role
        )))
    }
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::*;
    use koan_core::db::connection::Database;
    use koan_core::db::pool::Pool;
    use koan_core::db::queries;
    use koan_core::player::commands::CommandChannel;
    use tempfile::TempDir;

    /// One disposable configuration directory for the whole test binary.
    ///
    /// Deliberately not the per-test `TempDir`: enqueueing spawns downloads on
    /// a thread that outlives the test which started it, and that thread reads
    /// the configuration when it runs. Without this it reads the developer's
    /// own — their library, their server, and their credentials.
    fn isolate_config() {
        static DIR: std::sync::OnceLock<TempDir> = std::sync::OnceLock::new();
        koan_core::config::set_config_dir(DIR.get_or_init(|| TempDir::new().unwrap()).path());
    }

    fn test_schema() -> (
        KoanSchema,
        crossbeam_channel::Receiver<PlayerCommand>,
        TempDir,
    ) {
        isolate_config();
        let tmp = TempDir::new().unwrap();
        let db_path = tmp.path().join("test.db");
        let db = Database::open(&db_path).unwrap();
        koan_core::db::schema::create_tables(&db.conn).unwrap();

        let state = SharedPlayerState::new();
        let ch = CommandChannel::new();
        let tx = ch.tx.clone();
        let rx = ch.rx.clone();

        let schema = build_schema(state, tx, Arc::new(Pool::new(db_path.clone())), None);
        (schema, rx, tmp)
    }

    /// Same schema, but keeping the `DbHandle` so tests can read its counters.
    fn instrumented_schema() -> (
        KoanSchema,
        DbHandle,
        crossbeam_channel::Receiver<PlayerCommand>,
        TempDir,
    ) {
        let tmp = TempDir::new().unwrap();
        let db_path = tmp.path().join("test.db");
        let db = Database::open(&db_path).unwrap();
        koan_core::db::schema::create_tables(&db.conn).unwrap();

        let state = SharedPlayerState::new();
        let ch = CommandChannel::new();
        let handle = DbHandle::new(Arc::new(Pool::new(db_path)));
        let schema = build_schema_with(handle.clone(), state, ch.tx.clone(), None);
        (schema, handle, ch.rx.clone(), tmp)
    }

    /// Seed a library on one connection — the per-track helper opens its own.
    fn seed_library(db_path: &std::path::Path, artists: usize, albums: usize, tracks: usize) {
        let db = Database::open(db_path).unwrap();
        for a in 0..artists {
            for al in 0..albums {
                for t in 0..tracks {
                    let meta = queries::TrackMeta {
                        title: format!("Track {:03}-{}-{:03}", a, al, t),
                        artist: format!("Artist {:03}", a),
                        album_artist: Some(format!("Artist {:03}", a)),
                        album: format!("Album {:03}-{}", a, al),
                        track_number: Some(t as i32),
                        disc: Some(1),
                        disc_total: None,
                        date: Some("2024".into()),
                        genre: Some("Electronic".into()),
                        duration_ms: Some(240_000),
                        path: Some(format!("/tmp/koan-test/{}/{}/{}.flac", a, al, t)),
                        server_path: None,
                        codec: Some("FLAC".into()),
                        sample_rate: Some(44100),
                        bit_depth: Some(16),
                        channels: Some(2),
                        bitrate: Some(1411),
                        size_bytes: Some(42_000_000),
                        mtime: Some(1700000000),
                        source: "local".into(),
                        remote_id: None,
                        remote_url: None,
                        album_remote_id: None,
                        artist_remote_id: None,
                        mbid: None,
                        album_mbid: None,
                        album_added_at: None,
                        label: None,
                        credits: Default::default(),
                    };
                    queries::upsert_track(&db.conn, &meta).unwrap();
                }
            }
        }
    }

    fn insert_test_track(db_path: &std::path::Path, title: &str, artist: &str, album: &str) -> i64 {
        let db = Database::open(db_path).unwrap();
        let meta = queries::TrackMeta {
            title: title.to_string(),
            artist: artist.to_string(),
            album_artist: Some(artist.to_string()),
            album: album.to_string(),
            track_number: Some(1),
            disc: Some(1),
            disc_total: None,
            date: Some("2024".into()),
            genre: Some("Electronic".into()),
            duration_ms: Some(240000),
            path: Some(format!(
                "/tmp/test/{}.flac",
                title.to_lowercase().replace(' ', "_")
            )),
            server_path: None,
            codec: Some("FLAC".into()),
            sample_rate: Some(44100),
            bit_depth: Some(16),
            channels: Some(2),
            bitrate: Some(1411),
            size_bytes: Some(42_000_000),
            mtime: Some(1700000000),
            source: "local".into(),
            remote_id: None,
            remote_url: None,
            album_remote_id: None,
            artist_remote_id: None,
            mbid: None,
            album_mbid: None,
            album_added_at: None,
            label: None,
            credits: Default::default(),
        };
        queries::upsert_track(&db.conn, &meta).unwrap()
    }

    #[test]
    fn schema_builds() {
        let (_schema, _rx, _tmp) = test_schema();
    }

    #[tokio::test]
    async fn library_stats_query() {
        let (schema, _rx, tmp) = test_schema();
        let db_path = tmp.path().join("test.db");
        insert_test_track(&db_path, "Track1", "Artist1", "Album1");

        let resp = schema
            .execute("{ libraryStats { totalTracks totalAlbums totalArtists } }")
            .await;
        assert!(resp.errors.is_empty(), "errors: {:?}", resp.errors);
        let data = resp.data.into_json().unwrap();
        assert_eq!(data["libraryStats"]["totalTracks"], 1);
        assert_eq!(data["libraryStats"]["totalAlbums"], 1);
        assert_eq!(data["libraryStats"]["totalArtists"], 1);
    }

    #[tokio::test]
    async fn artists_query() {
        let (schema, _rx, tmp) = test_schema();
        let db_path = tmp.path().join("test.db");
        insert_test_track(&db_path, "T1", "Aphex Twin", "Drukqs");
        insert_test_track(&db_path, "T2", "Boards of Canada", "MHTRTC");

        let resp = schema
            .execute("{ artists { edges { node { id name } } } }")
            .await;
        assert!(resp.errors.is_empty(), "errors: {:?}", resp.errors);
        let data = resp.data.into_json().unwrap();
        let edges = data["artists"]["edges"].as_array().unwrap();
        assert_eq!(edges.len(), 2);
    }

    #[tokio::test]
    async fn tracks_search() {
        let (schema, _rx, tmp) = test_schema();
        let db_path = tmp.path().join("test.db");
        insert_test_track(&db_path, "Windowlicker", "Aphex Twin", "Windowlicker EP");
        insert_test_track(&db_path, "Roygbiv", "Boards of Canada", "MHTRTC");

        let resp = schema
            .execute(r#"{ tracks(search: "Aphex") { edges { node { id title artist } } } }"#)
            .await;
        assert!(resp.errors.is_empty(), "errors: {:?}", resp.errors);
        let data = resp.data.into_json().unwrap();
        let edges = data["tracks"]["edges"].as_array().unwrap();
        assert_eq!(edges.len(), 1);
        assert_eq!(edges[0]["node"]["title"], "Windowlicker");
    }

    #[tokio::test]
    async fn now_playing_stopped() {
        let (schema, _rx, _tmp) = test_schema();
        let resp = schema
            .execute("{ nowPlaying { state positionMs track { title } } }")
            .await;
        assert!(resp.errors.is_empty(), "errors: {:?}", resp.errors);
        let data = resp.data.into_json().unwrap();
        assert_eq!(data["nowPlaying"]["state"], "STOPPED");
    }

    #[tokio::test]
    async fn pause_mutation() {
        let (schema, rx, _tmp) = test_schema();
        let resp = schema.execute("mutation { pause { ok message } }").await;
        assert!(resp.errors.is_empty(), "errors: {:?}", resp.errors);
        let data = resp.data.into_json().unwrap();
        assert_eq!(data["pause"]["ok"], true);
        let cmd = rx.try_recv().unwrap();
        assert!(matches!(cmd, PlayerCommand::Pause));
    }

    #[tokio::test]
    async fn nested_artist_albums_tracks() {
        let (schema, _rx, tmp) = test_schema();
        let db_path = tmp.path().join("test.db");
        insert_test_track(&db_path, "Vordhosbn", "Aphex Twin", "Drukqs");
        insert_test_track(&db_path, "Avril 14th", "Aphex Twin", "Drukqs");

        let resp = schema
            .execute(
                r#"{ artists(search: "Aphex") {
                    edges { node {
                        name
                        albums { edges { node {
                            title
                            tracks { edges { node { title } } }
                        } } }
                    } }
                } }"#,
            )
            .await;
        assert!(resp.errors.is_empty(), "errors: {:?}", resp.errors);
        let data = resp.data.into_json().unwrap();
        let artist = &data["artists"]["edges"][0]["node"];
        assert_eq!(artist["name"], "Aphex Twin");
        let album = &artist["albums"]["edges"][0]["node"];
        assert_eq!(album["title"], "Drukqs");
        let tracks = album["tracks"]["edges"].as_array().unwrap();
        assert_eq!(tracks.len(), 2);
    }

    #[tokio::test]
    async fn pagination_has_next() {
        let (schema, _rx, tmp) = test_schema();
        let db_path = tmp.path().join("test.db");
        for i in 0..5 {
            insert_test_track(
                &db_path,
                &format!("Track{}", i),
                "Artist",
                &format!("Album{}", i),
            );
        }

        let resp = schema
            .execute(
                r#"{ artists(first: 1) {
                    edges { node { name } cursor }
                    pageInfo { hasNextPage endCursor }
                } }"#,
            )
            .await;
        assert!(resp.errors.is_empty(), "errors: {:?}", resp.errors);
        let data = resp.data.into_json().unwrap();
        // Only 1 artist ("Artist"), so hasNextPage should be false
        // since all 5 tracks are by the same artist.
        assert_eq!(data["artists"]["edges"].as_array().unwrap().len(), 1);
    }

    #[tokio::test]
    async fn clear_queue_mutation() {
        let (schema, rx, _tmp) = test_schema();
        let resp = schema
            .execute("mutation { clearQueue { ok message } }")
            .await;
        assert!(resp.errors.is_empty(), "errors: {:?}", resp.errors);
        let cmd = rx.try_recv().unwrap();
        assert!(matches!(cmd, PlayerCommand::ClearPlaylist));
    }

    #[tokio::test]
    async fn enqueue_mutation_adds_to_queue() {
        let (schema, rx, tmp) = test_schema();
        let db_path = tmp.path().join("test.db");

        // Insert a track into the DB.
        let track_id = insert_test_track(&db_path, "Windowlicker", "Aphex Twin", "Windowlicker EP");

        // Execute the addToQueue mutation.
        let query = format!(
            "mutation {{ addToQueue(trackIds: [{}]) {{ ok message addedCount queueItemIds }} }}",
            track_id
        );
        let resp = schema.execute(&query).await;
        assert!(resp.errors.is_empty(), "errors: {:?}", resp.errors);

        let data = resp.data.into_json().unwrap();
        assert_eq!(data["addToQueue"]["ok"], true);
        assert_eq!(data["addToQueue"]["addedCount"], 1);

        let queue_ids = data["addToQueue"]["queueItemIds"].as_array().unwrap();
        assert_eq!(queue_ids.len(), 1, "should return one queue item ID");

        // Verify the PlayerCommand was sent through the channel.
        // The mutation sends AddToPlaylist and then Play (auto-play when stopped).
        let cmd = rx.try_recv().unwrap();
        match cmd {
            PlayerCommand::AddToPlaylist(items) => {
                assert_eq!(items.len(), 1);
                assert_eq!(items[0].title, "Windowlicker");
                assert_eq!(items[0].artist, "Aphex Twin");
                assert_eq!(items[0].album, "Windowlicker EP");
            }
            other => panic!("expected AddToPlaylist, got {:?}", other),
        }

        // Auto-play command should follow.
        let play_cmd = rx.try_recv().unwrap();
        assert!(
            matches!(play_cmd, PlayerCommand::Play(_)),
            "expected Play command for auto-play, got {:?}",
            play_cmd
        );
    }

    #[tokio::test]
    async fn replace_queue_mutation_clears_and_enqueues() {
        let (schema, rx, tmp) = test_schema();
        let db_path = tmp.path().join("test.db");

        let id1 = insert_test_track(&db_path, "Track A", "Artist", "Album");
        let id2 = insert_test_track(&db_path, "Track B", "Artist", "Album");

        let query = format!(
            "mutation {{ replaceQueue(trackIds: [{}, {}]) {{ ok addedCount queueItemIds }} }}",
            id1, id2
        );
        let resp = schema.execute(&query).await;
        assert!(resp.errors.is_empty(), "errors: {:?}", resp.errors);

        let data = resp.data.into_json().unwrap();
        assert_eq!(data["replaceQueue"]["addedCount"], 2);

        // One command, not clear-then-add-then-play: three commands down a
        // bounded channel means the first track starts before the cursor lands
        // on the one that was asked for.
        match rx.try_recv().unwrap() {
            PlayerCommand::ReplacePlaylist { items, start, .. } => {
                assert_eq!(items.len(), 2);
                assert_eq!(start, 0, "defaults to the first track");
            }
            other => panic!("expected ReplacePlaylist, got {:?}", other),
        }
        assert!(rx.try_recv().is_err(), "no follow-up commands");
    }

    #[tokio::test]
    async fn replace_queue_starts_where_it_was_asked_to() {
        let (schema, rx, tmp) = test_schema();
        let db_path = tmp.path().join("test.db");

        let id1 = insert_test_track(&db_path, "Track A", "Artist", "Album");
        let id2 = insert_test_track(&db_path, "Track B", "Artist", "Album");

        let resp = schema
            .execute(&format!(
                "mutation {{ replaceQueue(trackIds: [{}, {}], startAt: 1) {{ ok }} }}",
                id1, id2
            ))
            .await;
        assert!(resp.errors.is_empty(), "errors: {:?}", resp.errors);

        match rx.try_recv().unwrap() {
            PlayerCommand::ReplacePlaylist { start, .. } => assert_eq!(start, 1),
            other => panic!("expected ReplacePlaylist, got {:?}", other),
        }
    }

    // ---- Playlists ----

    /// Ids are published as uids; an argument takes a uid, or the bare row id
    /// older callers send, and a uid only names a row of the kind asked for.
    #[tokio::test]
    async fn ids_are_uids_and_arguments_take_either() {
        let (schema, _rx, tmp) = test_schema();
        let row = insert_test_track(&tmp.path().join("test.db"), "Archangel", "Burial", "Untrue");

        let resp = schema
            .execute("{ tracks { edges { node { id albumId artistId } } } }")
            .await;
        assert!(resp.errors.is_empty(), "errors: {:?}", resp.errors);
        let data = resp.data.into_json().unwrap();
        let node = &data["tracks"]["edges"][0]["node"];
        let [id, album, artist] = ["id", "albumId", "artistId"].map(|k| {
            let v = node[k].as_str().unwrap().to_string();
            assert!(is_uid(&v), "{k}: {v}");
            v
        });
        assert!(id != album && album != artist);

        for arg in [format!("\"{id}\""), row.to_string()] {
            let resp = schema
                .execute(format!("{{ track(id: {arg}) {{ id }} }}"))
                .await;
            assert!(resp.errors.is_empty(), "errors: {:?}", resp.errors);
            assert_eq!(resp.data.into_json().unwrap()["track"]["id"], id);
        }

        let resp = schema
            .execute(format!("{{ track(id: \"{album}\") {{ id }} }}"))
            .await;
        assert!(!resp.errors.is_empty(), "an album's uid named a track");
    }

    /// A smart playlist made over the API selects by its rules, reports them
    /// back, refuses edits to its contents, and becomes ordinary on request.
    #[tokio::test]
    async fn smart_playlists_over_the_api() {
        let (schema, _rx, tmp) = test_schema();
        let db_path = tmp.path().join("test.db");
        let a = insert_test_track(&db_path, "Archangel", "Burial", "Untrue");
        insert_test_track(&db_path, "Teardrop", "Massive Attack", "Mezzanine");

        let resp = schema
            .execute(
                r#"mutation { createSmartPlaylist(name: "Burial", rules: {
                    rules: [{ field: "artist", op: "is", value: "burial" }]
                }) { id trackCount readonly rules } }"#,
            )
            .await;
        assert!(resp.errors.is_empty(), "errors: {:?}", resp.errors);
        let data = resp.data.into_json().unwrap();
        let made = &data["createSmartPlaylist"];
        assert_eq!(made["trackCount"], 1);
        assert_eq!(made["readonly"], true);
        assert_eq!(made["rules"]["rules"][0]["field"], "artist");
        let id = made["id"].to_string();

        let resp = schema
            .execute(format!(
                "mutation {{ addToPlaylist(id: {id}, trackIds: [{a}]) {{ ok }} }}"
            ))
            .await;
        assert!(
            !resp.errors.is_empty(),
            "an edit to a smart playlist's contents"
        );

        let resp = schema
            .execute(
                r#"mutation { createSmartPlaylist(name: "Bad", rules: {
                    rules: [{ field: "plays", op: "gt", value: 1 }]
                }) { id } }"#,
            )
            .await;
        let message = &resp.errors[0].message;
        assert!(message.contains("unknown field 'plays'"), "{message}");

        let resp = schema
            .execute(format!(
                "mutation {{ setPlaylistRules(id: {id}, rules: null) {{ readonly rules trackCount }} }}"
            ))
            .await;
        assert!(resp.errors.is_empty(), "errors: {:?}", resp.errors);
        let data = resp.data.into_json().unwrap();
        assert_eq!(data["setPlaylistRules"]["readonly"], false);
        assert_eq!(
            data["setPlaylistRules"]["trackCount"], 1,
            "keeps what it held"
        );
    }

    /// The whole life of a playlist over the API: made, added to, reordered,
    /// renamed, played, deleted.
    #[tokio::test]
    async fn playlists_round_trip_over_the_api() {
        let (schema, rx, tmp) = test_schema();
        let db_path = tmp.path().join("test.db");

        let a = insert_test_track(&db_path, "Track A", "Artist", "Album");
        let b = insert_test_track(&db_path, "Track B", "Artist", "Album");
        let c = insert_test_track(&db_path, "Track C", "Artist", "Album");

        let resp = schema
            .execute(&format!(
                "mutation {{ createPlaylist(name: \"Evening\", trackIds: [{a}, {b}]) \
                 {{ id name trackCount }} }}"
            ))
            .await;
        assert!(resp.errors.is_empty(), "errors: {:?}", resp.errors);
        let data = resp.data.into_json().unwrap();
        assert_eq!(data["createPlaylist"]["name"], "Evening");
        assert_eq!(data["createPlaylist"]["trackCount"], 2);
        assert!(is_uid(data["createPlaylist"]["id"].as_str().unwrap()));
        // As a quoted GraphQL literal; the track ids stay bare row ids, which
        // arguments still accept.
        let id = data["createPlaylist"]["id"].to_string();

        let resp = schema
            .execute(&format!(
                "mutation {{ addToPlaylist(id: {id}, trackIds: [{c}]) {{ ok }} }}"
            ))
            .await;
        assert!(resp.errors.is_empty(), "errors: {:?}", resp.errors);

        let resp = schema
            .execute(&format!("{{ playlistTracks(id: {id}) {{ title }} }}"))
            .await;
        let data = resp.data.into_json().unwrap();
        let titles: Vec<&str> = data["playlistTracks"]
            .as_array()
            .unwrap()
            .iter()
            .map(|t| t["title"].as_str().unwrap())
            .collect();
        assert_eq!(
            titles,
            ["Track A", "Track B", "Track C"],
            "playlist order is kept"
        );

        // A reorder is a wholesale replacement — the only shape Subsonic can
        // express, so it is the only shape koan stores.
        let resp = schema
            .execute(&format!(
                "mutation {{ setPlaylistTracks(id: {id}, trackIds: [{c}, {a}]) {{ ok }} }}"
            ))
            .await;
        assert!(resp.errors.is_empty(), "errors: {:?}", resp.errors);

        let resp = schema
            .execute(&format!(
                "mutation {{ renamePlaylist(id: {id}, name: \"Late\") {{ ok }} }}"
            ))
            .await;
        assert!(resp.errors.is_empty(), "errors: {:?}", resp.errors);

        let resp = schema.execute("{ playlists { id name trackCount } }").await;
        let data = resp.data.into_json().unwrap();
        assert_eq!(data["playlists"][0]["name"], "Late");
        assert_eq!(data["playlists"][0]["trackCount"], 2);

        // Playing it replaces the queue rather than adding to it.
        let resp = schema
            .execute(&format!("mutation {{ playPlaylist(id: {id}) {{ ok }} }}"))
            .await;
        assert!(resp.errors.is_empty(), "errors: {:?}", resp.errors);
        match rx.try_recv().unwrap() {
            PlayerCommand::ReplacePlaylist { items, start, .. } => {
                assert_eq!(items.len(), 2);
                assert_eq!(start, 0);
                assert!(items.iter().all(|i| i.playlist_entry_id.is_some()));
            }
            other => panic!("expected ReplacePlaylist, got {:?}", other),
        }
        assert!(rx.try_recv().is_err());

        let resp = schema
            .execute(&format!("mutation {{ deletePlaylist(id: {id}) {{ ok }} }}"))
            .await;
        assert!(resp.errors.is_empty(), "errors: {:?}", resp.errors);
        let resp = schema.execute("{ playlists { id } }").await;
        let data = resp.data.into_json().unwrap();
        assert!(data["playlists"].as_array().unwrap().is_empty());
    }

    /// A queue item with no library row behind it cannot go in a playlist: a
    /// playlist points at rows, not at paths.
    #[tokio::test]
    async fn saving_the_queue_keeps_only_what_the_library_knows() {
        use koan_core::player::state::{ItemState, PlaylistItem};

        let tmp = TempDir::new().unwrap();
        let db_path = tmp.path().join("test.db");
        let db = Database::open(&db_path).unwrap();
        koan_core::db::schema::create_tables(&db.conn).unwrap();
        drop(db);
        let known = insert_test_track(&db_path, "Known", "Artist", "Album");

        let state = SharedPlayerState::new();
        let ch = CommandChannel::new();
        let schema = build_schema(
            state.clone(),
            ch.tx.clone(),
            Arc::new(Pool::new(db_path.clone())),
            None,
        );

        let item = |title: &str, path: &str, db_id: Option<i64>| PlaylistItem {
            uid: None,
            playlist_entry_id: None,
            id: QueueItemId::new(),
            db_id,
            path: std::path::PathBuf::from(path),
            title: title.to_string(),
            artist: "Artist".to_string(),
            album_artist: "Artist".to_string(),
            album: "Album".to_string(),
            year: None,
            codec: None,
            track_number: None,
            disc: None,
            duration_ms: None,
            state: ItemState::Ready,
            played: false,
        };
        state.add_items(vec![
            item("Known", "/music/known.flac", Some(known)),
            // Played from a file that was never indexed — a queue can hold one,
            // a playlist cannot.
            item("Stranger", "/elsewhere/stranger.flac", None),
        ]);

        let resp = schema
            .execute("mutation { saveQueueAsPlaylist(name: \"Tonight\") { id trackCount } }")
            .await;
        assert!(resp.errors.is_empty(), "errors: {:?}", resp.errors);
        let data = resp.data.into_json().unwrap();
        assert_eq!(data["saveQueueAsPlaylist"]["trackCount"], 1);
    }

    // ---- Queue snapshot, viz, config, playlist version, subscriptions ----

    #[tokio::test]
    async fn queue_snapshot_has_version_and_status() {
        let (schema, _rx, _tmp) = test_schema();

        let resp = schema
            .execute("{ queue { version entries { queueItemId status } hasPlaying queueCount } }")
            .await;
        assert!(resp.errors.is_empty(), "errors: {:?}", resp.errors);
        let data = resp.data.into_json().unwrap();
        // Empty queue.
        assert_eq!(data["queue"]["version"], 0);
        assert_eq!(data["queue"]["entries"].as_array().unwrap().len(), 0);
        assert_eq!(data["queue"]["hasPlaying"], false);
        assert_eq!(data["queue"]["queueCount"], 0);
    }

    #[tokio::test]
    async fn queue_entries_have_status_and_download_progress() {
        use koan_core::player::state::{ItemState, PlaylistItem};

        // Build schema with a shared state we can manipulate directly.
        let tmp = TempDir::new().unwrap();
        let db_path = tmp.path().join("test.db");
        let db = Database::open(&db_path).unwrap();
        koan_core::db::schema::create_tables(&db.conn).unwrap();

        let state = SharedPlayerState::new();
        let ch = CommandChannel::new();
        let schema = build_schema(
            state.clone(),
            ch.tx.clone(),
            Arc::new(Pool::new(db_path.clone())),
            None,
        );

        // Directly add items to the playlist (simulating what the player thread does).
        let item = PlaylistItem {
            uid: None,
            playlist_entry_id: None,
            id: QueueItemId::new(),
            db_id: None,
            path: std::path::PathBuf::from("/tmp/test/windowlicker.flac"),
            title: "Windowlicker".to_string(),
            artist: "Aphex Twin".to_string(),
            album_artist: "Aphex Twin".to_string(),
            album: "Windowlicker EP".to_string(),
            year: None,
            codec: Some("FLAC".to_string()),
            track_number: Some(1),
            disc: Some(1),
            duration_ms: Some(240000),
            state: ItemState::Ready,
            played: false,
        };
        state.add_items(vec![item]);

        // Query the queue — should have one entry with QUEUED status.
        let resp = schema
            .execute(
                "{ queue { version entries { queueItemId title status downloadProgress { downloaded total } isCurrent } finishedCount } }",
            )
            .await;
        assert!(resp.errors.is_empty(), "errors: {:?}", resp.errors);
        let data = resp.data.into_json().unwrap();
        let entries = data["queue"]["entries"].as_array().unwrap();
        assert_eq!(entries.len(), 1);
        assert_eq!(entries[0]["title"], "Windowlicker");
        // Without a cursor set, all entries are QUEUED.
        assert_eq!(entries[0]["status"], "QUEUED");
        assert_eq!(entries[0]["isCurrent"], false);
        // Local track — no download progress.
        assert!(entries[0]["downloadProgress"].is_null());
    }

    #[tokio::test]
    async fn viz_frame_returns_none_without_viz() {
        let (schema, _rx, _tmp) = test_schema();

        let resp = schema
            .execute("{ vizFrame { spectrum peaks vuLevels beatEnergy } }")
            .await;
        assert!(resp.errors.is_empty(), "errors: {:?}", resp.errors);
        let data = resp.data.into_json().unwrap();
        assert!(data["vizFrame"].is_null());
    }

    #[tokio::test]
    async fn viz_frame_returns_data_with_viz() {
        // Build schema with a VizSnapshot.
        let tmp = TempDir::new().unwrap();
        let db_path = tmp.path().join("test.db");
        let db = Database::open(&db_path).unwrap();
        koan_core::db::schema::create_tables(&db.conn).unwrap();

        let state = SharedPlayerState::new();
        let ch = CommandChannel::new();
        let viz = koan_core::audio::viz::VizSnapshot::new();

        // Write some test data.
        let mut spectrum = [0.0f32; 48];
        spectrum[0] = 0.75;
        viz.write(koan_core::audio::viz::VizFrame {
            spectrum,
            peaks: [0.0; 48],
            vu_levels: [0.42, 0.38],
            beat_energy: 0.6,
            timestamp: std::time::Instant::now(),
            waveform: Vec::new(),
            loudness: Default::default(),
        });

        let schema = build_schema(
            state,
            ch.tx.clone(),
            Arc::new(Pool::new(db_path.clone())),
            Some(viz),
        );

        let resp = schema
            .execute("{ vizFrame { spectrum peaks vuLevels beatEnergy waveform } }")
            .await;
        assert!(resp.errors.is_empty(), "errors: {:?}", resp.errors);
        let data = resp.data.into_json().unwrap();
        let frame = &data["vizFrame"];
        assert!(!frame.is_null());
        let spectrum = frame["spectrum"].as_array().unwrap();
        assert_eq!(spectrum.len(), 48);
        assert!((spectrum[0].as_f64().unwrap() - 0.75).abs() < 0.01);
        let vu = frame["vuLevels"].as_array().unwrap();
        assert_eq!(vu.len(), 2);
        assert!((vu[0].as_f64().unwrap() - 0.42).abs() < 0.01);
        assert!((frame["beatEnergy"].as_f64().unwrap() - 0.6).abs() < 0.01);
        // Waveform empty — we didn't request includeWaveform.
        let waveform = frame["waveform"].as_array().unwrap();
        assert!(waveform.is_empty());
    }

    #[tokio::test]
    async fn config_query() {
        let (schema, _rx, _tmp) = test_schema();

        let resp = schema
            .execute(
                "{ config { libraryFolders replaygainMode targetFps artSize remoteEnabled graphqlPort } }",
            )
            .await;
        assert!(resp.errors.is_empty(), "errors: {:?}", resp.errors);
        let data = resp.data.into_json().unwrap();
        let cfg = &data["config"];
        // Defaults from Config::default().
        assert!(cfg["libraryFolders"].is_array());
        assert!(cfg["targetFps"].as_i64().unwrap() > 0);
        assert!(cfg["artSize"].as_i64().unwrap() > 0);
    }

    #[tokio::test]
    async fn playlist_version_query() {
        let (schema, _rx, _tmp) = test_schema();

        let resp = schema.execute("{ playlistVersion }").await;
        assert!(resp.errors.is_empty(), "errors: {:?}", resp.errors);
        let data = resp.data.into_json().unwrap();
        assert_eq!(data["playlistVersion"], 0);
    }

    #[tokio::test]
    async fn subscription_types_in_schema() {
        // Verify that subscriptions are registered by introspecting the schema.
        let (schema, _rx, _tmp) = test_schema();

        let resp = schema
            .execute("{ __schema { subscriptionType { fields { name } } } }")
            .await;
        assert!(resp.errors.is_empty(), "errors: {:?}", resp.errors);
        let data = resp.data.into_json().unwrap();
        let fields = data["__schema"]["subscriptionType"]["fields"]
            .as_array()
            .unwrap();
        let names: Vec<&str> = fields.iter().filter_map(|f| f["name"].as_str()).collect();
        assert!(
            names.contains(&"nowPlaying"),
            "missing nowPlaying subscription"
        );
        assert!(
            names.contains(&"queueUpdated"),
            "missing queueUpdated subscription"
        );
        assert!(names.contains(&"vizFrame"), "missing vizFrame subscription");
    }
    // -- Fan-out and pagination --

    #[tokio::test]
    async fn nested_fan_out_opens_a_bounded_number_of_connections() {
        let (schema, handle, _rx, tmp) = instrumented_schema();
        seed_library(&tmp.path().join("test.db"), 10, 3, 4);

        let before = handle.open_count();
        let resp = schema
            .execute(
                "{ artists(first: 10) { edges { node { name albumCount trackCount \
                   albums(first: 10) { edges { node { title trackCount totalDurationMs \
                   tracks(first: 10) { edges { node { title isFavourite } } } } } } } } } }",
            )
            .await;
        assert!(resp.errors.is_empty(), "errors: {:?}", resp.errors);

        let data = resp.data.into_json().unwrap();
        let artists = data["artists"]["edges"].as_array().unwrap();
        assert_eq!(artists.len(), 10, "query did not actually fan out");
        assert_eq!(artists[0]["node"]["albumCount"], 3);
        assert_eq!(artists[0]["node"]["trackCount"], 12);

        // Before pooling this was one full `Database::open` — DDL batch,
        // migrations and a WAL checkpoint — per resolver field.
        let opened = handle.open_count() - before;
        assert!(opened <= 16, "opened {} connections", opened);
    }

    #[tokio::test]
    async fn is_favourite_batches_into_one_query() {
        let (schema, handle, _rx, tmp) = instrumented_schema();
        seed_library(&tmp.path().join("test.db"), 1, 1, 100);

        let before = handle.batch_count();
        let resp = schema
            .execute("{ tracks(first: 100) { edges { node { title isFavourite } } } }")
            .await;
        assert!(resp.errors.is_empty(), "errors: {:?}", resp.errors);
        let data = resp.data.into_json().unwrap();
        assert_eq!(data["tracks"]["edges"].as_array().unwrap().len(), 100);

        // The invariant is that the query count does not scale with the row
        // count: without the dataloader this was one full `favourites` scan per
        // track. The dataloader's gather window can close more than once under
        // load, so assert the property rather than an exact batch count.
        let batches = handle.batch_count() - before;
        assert!(
            batches < 10,
            "{batches} batches for 100 tracks — expected a handful, not one per row"
        );
    }

    #[tokio::test]
    async fn tracks_without_first_returns_the_default_page() {
        let (schema, _handle, _rx, tmp) = instrumented_schema();
        seed_library(
            &tmp.path().join("test.db"),
            1,
            1,
            helpers::DEFAULT_PAGE + 20,
        );

        let resp = schema
            .execute("{ tracks { edges { node { title } } pageInfo { hasNextPage } } }")
            .await;
        assert!(resp.errors.is_empty(), "errors: {:?}", resp.errors);
        let data = resp.data.into_json().unwrap();
        assert_eq!(
            data["tracks"]["edges"].as_array().unwrap().len(),
            helpers::DEFAULT_PAGE
        );
        assert_eq!(data["tracks"]["pageInfo"]["hasNextPage"], true);
    }

    #[tokio::test]
    async fn first_is_clamped_to_the_maximum_page() {
        let (schema, _handle, _rx, tmp) = instrumented_schema();
        seed_library(&tmp.path().join("test.db"), 1, 1, 10);

        let resp = schema
            .execute("{ tracks(first: 100000) { edges { node { title } } } }")
            .await;
        assert!(resp.errors.is_empty(), "errors: {:?}", resp.errors);
        let data = resp.data.into_json().unwrap();
        assert_eq!(data["tracks"]["edges"].as_array().unwrap().len(), 10);
    }

    #[tokio::test]
    async fn negative_first_is_empty_not_a_panic() {
        let (schema, _handle, _rx, tmp) = instrumented_schema();
        seed_library(&tmp.path().join("test.db"), 2, 1, 3);

        for query in [
            "{ tracks(first: -1) { edges { node { title } } } }",
            "{ artists(first: -1) { edges { node { name } } } }",
            "{ albums(first: -1) { edges { node { title } } } }",
        ] {
            let resp = schema.execute(query).await;
            assert!(resp.errors.is_empty(), "{}: {:?}", query, resp.errors);
        }
    }

    #[tokio::test]
    async fn sort_arguments_are_honoured() {
        let (schema, _handle, _rx, tmp) = instrumented_schema();
        seed_library(&tmp.path().join("test.db"), 3, 1, 2);

        let resp = schema
            .execute(
                "{ tracks(first: 100, sortBy: TITLE, sortDir: DESC) { edges { node { title } } } }",
            )
            .await;
        assert!(resp.errors.is_empty(), "errors: {:?}", resp.errors);
        let data = resp.data.into_json().unwrap();
        let titles: Vec<String> = data["tracks"]["edges"]
            .as_array()
            .unwrap()
            .iter()
            .map(|e| e["node"]["title"].as_str().unwrap().to_string())
            .collect();
        let mut sorted = titles.clone();
        sorted.sort();
        sorted.reverse();
        assert_eq!(titles, sorted, "sortBy/sortDir were ignored");
    }

    /// A blocking resolver must not hold the runtime. On a single-threaded
    /// runtime, anything running inline would freeze every other task for the
    /// whole query — which is what stalled in-flight audio streams.
    #[test]
    fn a_slow_query_does_not_stall_other_tasks() {
        use std::sync::atomic::{AtomicUsize, Ordering};

        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_time()
            .build()
            .unwrap();
        let (schema, _handle, _rx, tmp) = instrumented_schema();
        seed_library(&tmp.path().join("test.db"), 20, 5, 8);

        rt.block_on(async move {
            let ticks = Arc::new(AtomicUsize::new(0));
            let counter = ticks.clone();
            let ticker = tokio::spawn(async move {
                loop {
                    tokio::time::sleep(Duration::from_millis(1)).await;
                    counter.fetch_add(1, Ordering::Relaxed);
                }
            });

            // fuzzySearch loads the library and builds a whole Nucleo matcher.
            let mut running = Vec::new();
            for _ in 0..4 {
                let schema = schema.clone();
                running.push(tokio::spawn(async move {
                    schema
                        .execute("{ fuzzySearch(query: \"track\") { id } }")
                        .await
                }));
            }
            for handle in running {
                let resp = handle.await.unwrap();
                assert!(resp.errors.is_empty(), "errors: {:?}", resp.errors);
            }

            ticker.abort();
            assert!(
                ticks.load(Ordering::Relaxed) >= 2,
                "no other task ran while the queries were in flight"
            );
        });
    }

    #[tokio::test]
    async fn accounts_are_made_and_invited_through_graphql() {
        let (schema, _rx, _tmp) = test_schema();
        let run = |q: &str| schema.execute(async_graphql::Request::new(q));
        let r = run(r#"mutation { createUser(username: "sarita", role: READONLY) { username link password emailText } }"#).await;
        assert!(
            r.errors[0].message.contains("server.url"),
            "no address to invite to: {:?}",
            r.errors
        );
        let r = run(r#"mutation { createUser(username: "sarita", role: READONLY, server: "https://music.example.com") { username link password emailText } }"#).await;
        assert!(r.errors.is_empty(), "{:?}", r.errors);
        let v = r.data.into_json().unwrap();
        let link = v["createUser"]["link"].as_str().unwrap();
        let invite = koan_core::invite::Invite::parse(link).unwrap();
        assert_eq!(invite.server, "https://music.example.com");
        assert!(invite.token.is_some() && invite.password.is_none());
        let password = v["createUser"]["password"].as_str().unwrap();
        assert!(
            v["createUser"]["emailText"]
                .as_str()
                .unwrap()
                .contains(password)
        );

        let r = run(r#"mutation { inviteUser(username: "sarita", server: "https://other.example.com") { server password } }"#).await;
        let v = r.data.into_json().unwrap();
        assert_eq!(v["inviteUser"]["server"], "https://other.example.com");
        assert!(v["inviteUser"]["password"].is_null());

        let r = run(
            r#"mutation { setUserPassword(username: "sarita", password: "correct horse") { ok } }"#,
        )
        .await;
        assert!(r.errors.is_empty(), "{:?}", r.errors);

        let r = run(r#"mutation { setUserRole(username: "sarita", role: USER) { ok } }"#).await;
        assert!(r.errors.is_empty(), "{:?}", r.errors);
        let r = run("{ users { username role } }").await;
        let v = r.data.into_json().unwrap();
        assert_eq!(v["users"][0]["role"], "USER");

        let r = run(r#"mutation { deleteUser(username: "sarita") { ok } }"#).await;
        assert!(r.errors.is_empty(), "{:?}", r.errors);
        let r = run("{ users { username } }").await;
        assert_eq!(
            r.data.into_json().unwrap()["users"]
                .as_array()
                .unwrap()
                .len(),
            0
        );
    }
}