mcpls-core 0.8.0

//! Shared in-memory mock LSP transport for `#[cfg(test)]` code across the
//! crate.
//!
//! Wires an [`LspClient`]'s [`LspTransport`] to a pair of
//! [`tokio::io::duplex`] pipes instead of a real subprocess loopback
//! (previously two `cat` child processes echoing bytes back to themselves,
//! duplicated near-verbatim across four modules -- see #378). This removes
//! the OS process-spawn/schedule dependency from the framing round-trip
//! itself. `LspServer`'s test-only constructors (`new_for_test`,
//! `lsp::fake_lsp_server`) no longer need a real child process either --
//! `LspServer`'s internal `child` field is `Option<tokio::process::Child>`,
//! `None` for every fixture built through this module.
//!
//! `pub`, not `pub(crate)`, on the items below: this module is itself
//! private (unexported), so `pub(crate)` here would be redundant -- see
//! `clippy::redundant_pub_crate`. Still only reachable crate-internally via
//! `crate::test_lsp::*`, since the module isn't `pub`.

use serde_json::Value;
use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader, DuplexStream};
use tokio::time::Duration;
use tokio_util::sync::CancellationToken;

#[cfg(unix)]
use crate::ServerId;
use crate::bridge::WorkspaceRoots;
#[cfg(unix)]
use crate::config::{LanguageId, ServerCommand, TimeoutSecs};
use crate::config::{LspServerConfig, PositionEncodings};
use crate::lsp::{LspClient, LspTransport, LspTransportReader, ServerInitConfig};

/// Duplex buffer capacity for the mock pipes below. Framed JSON-RPC
/// messages exchanged in these tests run from a few dozen bytes to a
/// handful of KB, so this is generous headroom, not a tuned value.
const MOCK_PIPE_CAPACITY: usize = 256 * 1024;

/// Upper bound [`read_framed_message`] waits for one complete frame before
/// failing loudly, instead of hanging until an external test-runner kill
/// (see that function's docs for why this matters). The longest legitimate
/// wait anywhere in the current suite is ~2s (the backoff delays in
/// `lsp::client::tests::retry_behavior`), so this still leaves a wide
/// margin on a slow/contended CI runner.
///
/// Not safe under `#[tokio::test(start_paused = true)]`: a paused clock
/// auto-advances to the next scheduled timer once the runtime goes idle,
/// so a call that is merely waiting on real (unadvanced) time would have
/// this timeout fire immediately instead of after a genuine 30s wall-clock
/// wait. Safe today only because no current caller of
/// [`read_framed_message`] runs under a paused-clock test.
const READ_FRAME_TIMEOUT: Duration = Duration::from_secs(30);

/// Test-side counterpart of a mocked LSP server, connected to an
/// [`LspClient`]'s transport through two independent [`tokio::io::duplex`]
/// pipes rather than a real subprocess.
pub struct FakeServer {
    /// Written by the test to inject a fake server response or
    /// notification; the client reads it as incoming bytes.
    pub read_half_stdin: DuplexStream,
    /// Written by the client; read by the test to observe the framed bytes
    /// the client actually sent.
    pub write_stdout: DuplexStream,
}

/// Builds an [`LspClient`] wired to an in-memory mock server, for a caller
/// that doesn't care which [`LspServerConfig`] is attached.
pub fn fake_lsp_client() -> (LspClient, FakeServer) {
    fake_lsp_client_with_config(LspServerConfig::rust_analyzer())
}

/// As [`fake_lsp_client`], with a caller-chosen [`LspServerConfig`].
pub fn fake_lsp_client_with_config(config: LspServerConfig) -> (LspClient, FakeServer) {
    let (transport, fake_server) = fake_transport();
    (LspClient::from_transport(config, transport), fake_server)
}

/// Both notification lanes of a client built by
/// [`fake_lsp_client_with_lanes`].
pub struct FakeLanes {
    /// Log/showMessage lane.
    pub notification_rx: tokio::sync::mpsc::Receiver<crate::lsp::ServerMessage>,
    /// The diagnostics mailbox.
    pub publishes: crate::lsp::PublishReader,
    /// Lifecycle lane (`$/progress` `begin`/`end`, unrecognized notifications).
    pub lifecycle_rx: tokio::sync::mpsc::Receiver<crate::lsp::LspNotification>,
}

/// As [`fake_lsp_client`], but the client forwards notifications onto real
/// lanes, returned for the test to drain or hand to a pump.
pub fn fake_lsp_client_with_lanes() -> (LspClient, FakeServer, FakeLanes) {
    fake_lsp_client_with_redactions(crate::redaction::Redactions::default())
}

/// As [`fake_lsp_client_with_lanes`], with the secrets the client hides.
pub fn fake_lsp_client_with_redactions(
    redactions: crate::redaction::Redactions,
) -> (LspClient, FakeServer, FakeLanes) {
    let (transport, fake_server) = fake_transport();
    let (notification_tx, notification_rx) = tokio::sync::mpsc::channel(32);
    let (notification_sink, publishes) = crate::lsp::NotificationSink::new(notification_tx);
    let (lifecycle_tx, lifecycle_rx) = tokio::sync::mpsc::channel(8);
    let client = LspClient::from_transport_with_notifications(
        LspServerConfig::rust_analyzer(),
        transport,
        notification_sink,
        lifecycle_tx,
        std::sync::Arc::new(redactions),
    );
    (
        client,
        fake_server,
        FakeLanes {
            notification_rx,
            publishes,
            lifecycle_rx,
        },
    )
}

fn fake_transport() -> ((LspTransport, LspTransportReader), FakeServer) {
    let (client_stdin, write_stdout) = tokio::io::duplex(MOCK_PIPE_CAPACITY);
    let (read_half_stdin, client_stdout) = tokio::io::duplex(MOCK_PIPE_CAPACITY);
    (
        LspTransport::new(client_stdin, client_stdout),
        FakeServer {
            read_half_stdin,
            write_stdout,
        },
    )
}

/// An [`LspTransport`] backed by duplex pipes whose peer half is dropped
/// immediately -- for tests that only need a real transport to satisfy a
/// constructor and never exercise it over the wire. A subsequent write
/// fails with a broken-pipe error and a read observes EOF, both
/// deterministically and immediately (unlike the old `cat`-backed
/// placeholder, which accepted writes and raced on EOF) -- the message
/// loop this feeds dies right away either way, so callers that construct
/// an `LspServer`/`LspClient` around this and never send real traffic
/// through it are unaffected.
#[must_use]
pub fn inert_transport() -> (LspTransport, LspTransportReader) {
    let (stdin, _unused_read) = tokio::io::duplex(1);
    let (_unused_write, stdout) = tokio::io::duplex(1);
    LspTransport::new(stdin, stdout)
}

/// Reads one `Content-Length`-framed JSON-RPC message off `reader`.
///
/// `reader` must be reused across calls, not recreated per message: a fresh
/// `BufReader` would silently drop any bytes of a later message it
/// over-read into its internal buffer while parsing an earlier one.
///
/// Bounded by [`READ_FRAME_TIMEOUT`]: a test bug that never writes the
/// expected message (or an EOF on the underlying duplex pipe, e.g. its
/// writer half was dropped) must fail fast with a clear panic message,
/// rather than hanging until nextest's external 120s kill turns it into an
/// opaque, unattributed timeout.
///
/// # Panics
///
/// Panics if no complete frame arrives within [`READ_FRAME_TIMEOUT`], or if
/// the stream reaches EOF before a complete frame is read.
pub async fn read_framed_message(reader: &mut BufReader<&mut DuplexStream>) -> Value {
    tokio::time::timeout(READ_FRAME_TIMEOUT, read_framed_message_inner(reader))
        .await
        .expect("timed out waiting for a complete framed JSON-RPC message")
}

/// As [`read_framed_message`], but waits as long as the next frame takes, for a
/// scripted server that sits idle between requests.
///
/// # Panics
///
/// Panics if the stream reaches EOF before a complete frame is read.
pub async fn next_framed_message(reader: &mut BufReader<&mut DuplexStream>) -> Value {
    read_framed_message_inner(reader).await
}

async fn read_framed_message_inner(reader: &mut BufReader<&mut DuplexStream>) -> Value {
    let mut content_length = None;
    let mut line = String::new();
    loop {
        line.clear();
        // `read_line` returns `Ok(0)` at EOF without erroring; left
        // unchecked, `line` stays `""`, which matches neither line-ending
        // check below, so the loop would spin forever re-reading EOF
        // instead of failing.
        let bytes_read = reader.read_line(&mut line).await.unwrap();
        assert!(bytes_read != 0, "EOF before a complete frame was read");
        if line == "\r\n" || line == "\n" {
            break;
        }
        if let Some((key, value)) = line.trim_end().split_once(':')
            && key.trim().eq_ignore_ascii_case("content-length")
        {
            content_length = Some(value.trim().parse::<usize>().unwrap());
        }
    }
    let mut buf = vec![0u8; content_length.unwrap()];
    reader.read_exact(&mut buf).await.unwrap();
    serde_json::from_slice(&buf).unwrap()
}

/// Writes a framed JSON-RPC success response, as a real LSP server would.
pub async fn write_response(writer: &mut DuplexStream, id: &Value, result: Value) {
    write_framed(
        writer,
        &serde_json::json!({
            "jsonrpc": "2.0",
            "id": id,
            "result": result,
        }),
    )
    .await;
}

/// Writes a framed JSON-RPC error response, e.g. to simulate a push-only
/// server answering a request with method-not-found.
pub async fn write_error_response(writer: &mut DuplexStream, id: &Value, code: i64, message: &str) {
    write_framed(
        writer,
        &serde_json::json!({
            "jsonrpc": "2.0",
            "id": id,
            "error": { "code": code, "message": message },
        }),
    )
    .await;
}

/// Writes a framed server-to-client JSON-RPC request.
pub async fn write_request(writer: &mut DuplexStream, id: &Value, method: &str, params: Value) {
    write_framed(
        writer,
        &serde_json::json!({
            "jsonrpc": "2.0",
            "id": id,
            "method": method,
            "params": params,
        }),
    )
    .await;
}

/// Writes a framed JSON-RPC notification, as a real LSP server would.
pub async fn write_notification(writer: &mut DuplexStream, method: &str, params: Value) {
    write_framed(
        writer,
        &serde_json::json!({
            "jsonrpc": "2.0",
            "method": method,
            "params": params,
        }),
    )
    .await;
}

async fn write_framed(writer: &mut DuplexStream, message: &Value) {
    write_raw_frame(writer, &serde_json::to_string(message).unwrap()).await;
}

/// Writes `content` as one frame without parsing it, so a test can send a
/// body no `serde_json::Value` could hold, such as one nested past the
/// parser's recursion limit.
pub async fn write_raw_frame(writer: &mut DuplexStream, content: &str) {
    let header = format!("Content-Length: {}\r\n\r\n", content.len());
    writer.write_all(header.as_bytes()).await.unwrap();
    writer.write_all(content.as_bytes()).await.unwrap();
    writer.flush().await.unwrap();
}

/// Captures `(level, message)` pairs for `tracing` events emitted while a
/// closure runs, as a `tracing_subscriber::Layer`.
///
/// Shared here rather than duplicated per-module: `transport.rs`, `lib.rs`
/// and `bridge/notifications.rs` each used to carry their own
/// message-only `CapturedMessages` copy. This is a strict superset (it
/// also records the event's [`tracing::Level`], needed to assert that a
/// log line is or is not at `error!` severity, not just that it contains
/// certain text) so it replaces all of them.
#[derive(Clone, Default)]
pub struct CapturedLogs(std::sync::Arc<std::sync::Mutex<Vec<(tracing::Level, String)>>>);

impl CapturedLogs {
    /// Snapshot of everything captured so far, as `(level, message)` pairs.
    pub fn entries(&self) -> Vec<(tracing::Level, String)> {
        self.0.lock().unwrap().clone()
    }

    /// Snapshot of captured message text only, for callers that don't care
    /// about severity.
    pub fn messages(&self) -> Vec<String> {
        self.0
            .lock()
            .unwrap()
            .iter()
            .map(|(_, message)| message.clone())
            .collect()
    }
}

impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> for CapturedLogs {
    fn on_event(
        &self,
        event: &tracing::Event<'_>,
        _ctx: tracing_subscriber::layer::Context<'_, S>,
    ) {
        struct MessageVisitor(String);
        impl tracing::field::Visit for MessageVisitor {
            fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
                if field.name() == "message" {
                    self.0 = format!("{value:?}");
                }
            }
        }
        let mut visitor = MessageVisitor(String::new());
        event.record(&mut visitor);
        self.0
            .lock()
            .unwrap()
            .push((*event.metadata().level(), visitor.0));
    }
}

/// Minimal [`ServerInitConfig`] around `server_config` for test fixtures.
pub fn init_config_for(server_config: LspServerConfig) -> ServerInitConfig {
    ServerInitConfig::new(
        server_config,
        WorkspaceRoots::default(),
        PositionEncodings::DEFAULT,
        std::sync::Arc::default(),
    )
}

/// Creates a FIFO at `path`, panicking when `mkfifo` fails.
#[cfg(unix)]
pub fn make_fifo(path: &std::path::Path) {
    let status = std::process::Command::new("mkfifo")
        .arg(path)
        .status()
        .unwrap();
    assert!(status.success(), "mkfifo must succeed to set up this test");
}

/// Extension map shared by routing tests: `.rs` and `.tsx`.
pub fn test_extensions()
-> std::collections::HashMap<crate::config::FileExtension, crate::config::LanguageId> {
    std::collections::HashMap::from([
        (
            crate::config::FileExtension::from_static("rs"),
            crate::config::LanguageId::from_static("rust"),
        ),
        (
            crate::config::FileExtension::from_static("tsx"),
            crate::config::LanguageId::from_static("typescriptreact"),
        ),
    ])
}

/// Writes `script_body` to `dir/server.sh` and returns a config that runs it
/// under `sh` as an LSP server.
#[cfg(unix)]
pub fn sh_script_init_config(dir: &std::path::Path, script_body: &str) -> ServerInitConfig {
    let script = dir.join("server.sh");
    std::fs::write(&script, script_body).unwrap();
    let mut server_config = LspServerConfig::rust_analyzer();
    server_config.command = ServerCommand::from_static("sh").into();
    server_config.args =
        crate::config::ServerArgs::new([script.to_string_lossy().to_string()]).unwrap();
    init_config_for(server_config)
}

/// Script of an `sh` fake server that answers `initialize` with empty
/// capabilities: it first creates `announce` and waits until `wait_for`
/// exists (when given), so tests can order servers without wall-clock sleeps.
#[cfg(unix)]
pub fn answer_initialize_script(
    announce: Option<&std::path::Path>,
    wait_for: Option<&std::path::Path>,
) -> String {
    let announce =
        announce.map_or_else(String::new, |path| format!("touch '{}'\n", path.display()));
    let wait = wait_for.map_or_else(String::new, |path| {
        format!("while [ ! -f '{}' ]; do sleep 0.05; done\n", path.display())
    });
    let answer = r#"body='{"jsonrpc":"2.0","id":1,"result":{"capabilities":{}}}'
printf 'Content-Length: %d\r\n\r\n%s' ${#body} "$body"
sleep 60
"#;
    format!("{announce}{wait}{}", with_read_preamble(answer))
}

/// An `sh` fake server named `name` for `language`, living in its own
/// subdirectory of `dir` and bounded by a 10 s initialize timeout.
#[cfg(unix)]
pub fn named_sh_init_config(
    dir: &std::path::Path,
    name: &str,
    language: &str,
    script_body: &str,
) -> ServerInitConfig {
    let sub = dir.join(name);
    std::fs::create_dir_all(&sub).unwrap();
    let mut config = sh_script_init_config(&sub, script_body);
    config.server_config_mut().name = Some(ServerId::new(name).unwrap());
    config.server_config_mut().language_id = LanguageId::new(language).unwrap();
    config.server_config_mut().timeout_seconds = TimeoutSecs::new(10).unwrap();
    config
}

/// Reads and discards a full LSP-framed request from stdin, so a fixture
/// replying after it can't answer before the request is even sent (#447).
#[cfg(unix)]
pub const READ_REQUEST_SH: &str = r#"content_length=0
while IFS= read -r header; do
  header=$(printf '%s' "$header" | tr -d '\r')
  [ -z "$header" ] && break
  case "$header" in
    Content-Length:*) content_length=$(printf '%s' "$header" | sed 's/^Content-Length: *//') ;;
  esac
done
[ "$content_length" -gt 0 ] 2>/dev/null && dd bs=1 count="$content_length" 2>/dev/null >/dev/null
"#;

/// Prepends [`READ_REQUEST_SH`] to a fake `sh`-server fixture body.
#[cfg(unix)]
pub fn with_read_preamble(body: &str) -> String {
    format!("{READ_REQUEST_SH}{body}")
}

/// Feeds a pump the way a client's message loop does: publishes go through the
/// diagnostics mailbox, log and `showMessage` frames through the bounded
/// channel.
#[derive(Debug)]
pub struct PumpFeed {
    messages: tokio::sync::mpsc::Sender<crate::lsp::ServerMessage>,
    publishes: crate::lsp::PublishWriter,
}

impl PumpFeed {
    /// Delivers `notification` on the lane the client routes it to.
    ///
    /// # Panics
    ///
    /// On a notification of the lifecycle lane, which a [`PumpFeed`] does not
    /// carry.
    pub async fn send(
        &self,
        notification: crate::lsp::LspNotification,
    ) -> Result<(), tokio::sync::mpsc::error::SendError<crate::lsp::ServerMessage>> {
        use crate::lsp::{LspNotification, ServerMessage};
        match notification {
            LspNotification::PublishDiagnostics(params) => {
                self.publishes.publish(params);
                Ok(())
            }
            LspNotification::LogMessage(params) => {
                self.messages.send(ServerMessage::Log(params)).await
            }
            LspNotification::ShowMessage(params) => {
                self.messages.send(ServerMessage::Show(params)).await
            }
            lifecycle @ (LspNotification::Progress(_) | LspNotification::Other { .. }) => {
                panic!("not a notification-lane frame: {lifecycle:?}")
            }
        }
    }
}

/// A [`PumpFeed`] and the inbox of the pump it feeds.
pub fn pump_feed() -> (PumpFeed, crate::lsp::NotificationInbox) {
    let (messages, messages_rx) = tokio::sync::mpsc::channel(32);
    let (publishes, publishes_rx) = crate::lsp::mailbox(crate::lsp::MailboxLimits::default());
    (
        PumpFeed {
            messages,
            publishes,
        },
        crate::lsp::NotificationInbox::new(messages_rx, publishes_rx),
    )
}

/// Spawns a real [`crate::runtime::pump::diagnostics_pump`] over `subs` and returns the
/// feed of its notification lane plus the cancel token.
pub fn spawn_test_pump(
    subs: crate::mcp::SubscriptionRegistry,
    workspace_roots: WorkspaceRoots,
) -> (PumpFeed, CancellationToken) {
    let cache = std::sync::Arc::new(tokio::sync::Mutex::new(
        crate::bridge::NotificationCache::new(),
    ));
    spawn_test_pump_with_cache(subs, workspace_roots, cache)
}

/// As [`spawn_test_pump`], over a caller-supplied notification cache so the
/// test can inspect what the pump stored.
pub fn spawn_test_pump_with_cache(
    subs: crate::mcp::SubscriptionRegistry,
    workspace_roots: WorkspaceRoots,
    notification_cache: std::sync::Arc<tokio::sync::Mutex<crate::bridge::NotificationCache>>,
) -> (PumpFeed, CancellationToken) {
    let (tx, rx) = pump_feed();
    let (lifecycle_tx, lifecycle_rx) = tokio::sync::mpsc::channel(8);
    // Held so the lifecycle lane stays open for the pump's lifetime.
    let (_cache, cancel) = spawn_pump(
        rx,
        lifecycle_rx,
        lifecycle_tx,
        subs,
        workspace_roots,
        notification_cache,
        None,
    );
    (tx, cancel)
}

/// A pump for a server configured with `tsserver.path = pinned`, plus the
/// sender of its lifecycle lane and the cancel token.
pub fn spawn_test_pump_with_tsserver_pin(
    pinned: std::path::PathBuf,
) -> (
    tokio::sync::mpsc::Sender<crate::lsp::LspNotification>,
    CancellationToken,
) {
    let (feed, rx) = pump_feed();
    let (lifecycle_tx, lifecycle_rx) = tokio::sync::mpsc::channel(8);
    let (_cache, cancel) = spawn_pump(
        rx,
        lifecycle_rx,
        feed,
        crate::mcp::SubscriptionRegistry::default(),
        WorkspaceRoots::default(),
        std::sync::Arc::new(tokio::sync::Mutex::new(
            crate::bridge::NotificationCache::new(),
        )),
        Some(pinned),
    );
    (lifecycle_tx, cancel)
}

/// As [`spawn_test_pump`], but over a client's own [`FakeLanes`]; returns the
/// pump's notification cache (for asserting on indexing state) and the
/// cancel token.
pub fn spawn_test_pump_over_lanes(
    lanes: FakeLanes,
) -> (
    std::sync::Arc<tokio::sync::Mutex<crate::bridge::NotificationCache>>,
    CancellationToken,
) {
    spawn_pump(
        crate::lsp::NotificationInbox::new(lanes.notification_rx, lanes.publishes),
        lanes.lifecycle_rx,
        (),
        crate::mcp::SubscriptionRegistry::default(),
        WorkspaceRoots::default(),
        std::sync::Arc::new(tokio::sync::Mutex::new(
            crate::bridge::NotificationCache::new(),
        )),
        None,
    )
}

fn spawn_pump<K: Send + 'static>(
    rx: impl Into<crate::lsp::NotificationInbox> + Send + 'static,
    lifecycle_rx: tokio::sync::mpsc::Receiver<crate::lsp::LspNotification>,
    keep_alive: K,
    subs: crate::mcp::SubscriptionRegistry,
    workspace_roots: WorkspaceRoots,
    notification_cache: std::sync::Arc<tokio::sync::Mutex<crate::bridge::NotificationCache>>,
    pinned_tsserver: Option<std::path::PathBuf>,
) -> (
    std::sync::Arc<tokio::sync::Mutex<crate::bridge::NotificationCache>>,
    CancellationToken,
) {
    let cancel = CancellationToken::new();
    let shared = crate::runtime::pump::PumpShared {
        roles: crate::runtime::pump::DiagnosticsRoles::default(),
        notification_cache: std::sync::Arc::clone(&notification_cache),
        subs,
        workspace_roots,
    };
    let pump_cancel = cancel.clone();
    tokio::spawn(async move {
        let _keep_alive = keep_alive;
        crate::runtime::pump::diagnostics_pump(
            crate::config::ServerId::from_static("rust"),
            rx,
            lifecycle_rx,
            pump_cancel,
            tokio::sync::watch::channel(crate::bridge::DiagnosticsRole::Authoritative).1,
            pinned_tsserver,
            shared,
        )
        .await;
    });
    (notification_cache, cancel)
}

/// A temp workspace holding `main.rs`: the guard, the canonical root and the file.
pub fn workspace_with_main_rs() -> (tempfile::TempDir, std::path::PathBuf, std::path::PathBuf) {
    let dir = tempfile::TempDir::new().unwrap();
    let root = dunce::canonicalize(dir.path()).unwrap();
    let file = root.join("main.rs");
    std::fs::write(&file, "fn main() {}").unwrap();
    (dir, root, file)
}

/// An absolute path that is valid on the host platform (Windows needs a drive
/// letter for `lsp-diagnostics:///` URIs to parse), for URIs that never touch disk.
pub fn absolute_path(relative: &str) -> std::path::PathBuf {
    let root = if cfg!(windows) { r"C:\" } else { "/" };
    std::path::PathBuf::from(root).join(relative)
}

/// A TOML string literal for `path`, escaped as the TOML grammar requires.
pub fn toml_path_literal(path: &std::path::Path) -> String {
    toml::Value::String(path.to_string_lossy().into_owned()).to_string()
}

/// A [`ClientPath`](crate::bridge::ClientPath) for a path a test knows is
/// well-formed.
pub fn client_path(path: impl AsRef<std::path::Path>) -> crate::bridge::ClientPath {
    crate::bridge::ClientPath::try_from(path.as_ref().to_path_buf()).unwrap()
}

/// The `lsp-diagnostics:///` URI of the path [`absolute_path`] returns.
pub fn absolute_uri(relative: &str) -> String {
    crate::bridge::resources::make_uri(&absolute_path(relative)).unwrap()
}

/// A diagnostics resource URI built from a literal, bypassing resolution.
pub fn diagnostics_uri(uri: &str) -> crate::bridge::resources::DiagnosticsResourceUri {
    crate::bridge::resources::DiagnosticsResourceUri::for_test(uri)
}

/// A [`NotificationWiring`](crate::bridge::NotificationWiring) that holds every
/// invalidation back until the test releases it, and counts the ones let
/// through, so a test can pin a call at its announcement step.
#[cfg(unix)]
#[derive(Debug)]
pub struct GatedWiring {
    gate: tokio::sync::Semaphore,
    arrived: std::sync::atomic::AtomicUsize,
    published: std::sync::atomic::AtomicUsize,
}

#[cfg(unix)]
impl GatedWiring {
    /// A wiring that publishes nothing until [`Self::release`].
    pub fn closed() -> Self {
        Self {
            gate: tokio::sync::Semaphore::new(0),
            arrived: std::sync::atomic::AtomicUsize::new(0),
            published: std::sync::atomic::AtomicUsize::new(0),
        }
    }

    /// Lets one more invalidation through.
    pub fn release(&self) {
        self.gate.add_permits(1);
    }

    /// How many calls reached the gate, whether or not they were let through.
    pub fn arrived(&self) -> usize {
        self.arrived.load(std::sync::atomic::Ordering::SeqCst)
    }

    /// How many invalidations went through.
    pub fn published(&self) -> usize {
        self.published.load(std::sync::atomic::Ordering::SeqCst)
    }

    /// Waits until a call has reached the gate, whether or not it is let
    /// through.
    pub async fn wait_arrived(&self) {
        while self.arrived.load(std::sync::atomic::Ordering::SeqCst) == 0 {
            tokio::task::yield_now().await;
        }
    }

    /// Waits until at least one invalidation went through.
    pub async fn wait_published(&self) {
        while self.published() == 0 {
            tokio::task::yield_now().await;
        }
    }
}

#[cfg(unix)]
impl crate::bridge::NotificationWiring for GatedWiring {
    fn spawn_pump(
        &self,
        _id: crate::config::ServerId,
        _receivers: crate::bridge::NotificationReceivers,
        _role: crate::bridge::DiagnosticsRole,
    ) -> tokio::task::AbortHandle {
        tokio::spawn(async {}).abort_handle()
    }

    fn publish_invalidated<'a>(
        &'a self,
        _cleared: &'a [crate::bridge::DiagnosticsKey],
    ) -> futures::future::BoxFuture<'a, ()> {
        Box::pin(async move {
            self.arrived
                .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            if let Ok(permit) = self.gate.acquire().await {
                permit.forget();
            }
            self.published
                .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
        })
    }

    fn publish_changed<'a>(
        &'a self,
        _file: &'a lsp_types::Uri,
    ) -> futures::future::BoxFuture<'a, ()> {
        Box::pin(async {})
    }
}

#[cfg(all(test, unix))]
mod gated_wiring_tests {
    use std::sync::Arc;

    use super::GatedWiring;
    use crate::bridge::NotificationWiring as _;

    #[tokio::test]
    async fn one_release_lets_exactly_one_invalidation_through() {
        let wiring = Arc::new(GatedWiring::closed());
        wiring.release();
        for _ in 0..2 {
            let wiring = Arc::clone(&wiring);
            tokio::spawn(async move { wiring.publish_invalidated(&[]).await });
        }
        while wiring.arrived() < 2 {
            tokio::task::yield_now().await;
        }
        for _ in 0..16 {
            tokio::task::yield_now().await;
        }
        assert_eq!(wiring.published(), 1);
    }
}