mod input_handler;

pub use lsp_types::request::*;
pub use lsp_types::*;

use anyhow::{Context as _, Result, anyhow};
use collections::{BTreeMap, HashMap};
use futures::{
    AsyncRead, AsyncWrite, Future, FutureExt,
    channel::oneshot::{self, Canceled},
    future::{self, Either},
    io::BufWriter,
    select,
};
use gpui::{App, AppContext as _, AsyncApp, BackgroundExecutor, SharedString, Task};
use notification::DidChangeWorkspaceFolders;
use parking_lot::{Mutex, RwLock};
use postage::{barrier, prelude::Stream};
use schemars::JsonSchema;
use serde::{Deserialize, Serialize, de::DeserializeOwned};
use serde_json::{Value, json, value::RawValue};
use smol::{
    channel,
    io::{AsyncBufReadExt, AsyncWriteExt, BufReader},
};
use util::command::{Child, Stdio};

use std::path::Path;
use std::{
    any::TypeId,
    collections::BTreeSet,
    ffi::{OsStr, OsString},
    fmt,
    io::Write,
    ops::DerefMut,
    path::PathBuf,
    pin::Pin,
    sync::{
        Arc, Weak,
        atomic::{AtomicI32, Ordering::SeqCst},
    },
    task::Poll,
    time::{Duration, Instant},
};
use util::{ConnectionResult, ResultExt, TryFutureExt, redact};

const JSON_RPC_VERSION: &str = "2.0";
const CONTENT_LEN_HEADER: &str = "Content-Length: ";

/// The default amount of time to wait while initializing or fetching LSP servers, in seconds.
///
/// Should not be used (in favor of DEFAULT_LSP_REQUEST_TIMEOUT) and is exported solely for use inside ProjectSettings defaults.
pub const DEFAULT_LSP_REQUEST_TIMEOUT_SECS: u64 = 120;
/// A timeout representing the value of [DEFAULT_LSP_REQUEST_TIMEOUT_SECS].
///
/// Should **only be used** in tests and as a fallback when a corresponding config value cannot be obtained!
pub const DEFAULT_LSP_REQUEST_TIMEOUT: Duration =
    Duration::from_secs(DEFAULT_LSP_REQUEST_TIMEOUT_SECS);

/// The shutdown timeout for LSP servers (including Prettier/Copilot).
const SERVER_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(5);

type NotificationHandler = Box<dyn Send + FnMut(Option<RequestId>, Value, &mut AsyncApp)>;
type PendingRespondTasks = Arc<Mutex<HashMap<RequestId, Task<()>>>>;
type ResponseHandler = Box<dyn Send + FnOnce(Result<String, Error>) -> Task<()>>;
type IoHandler = Box<dyn Send + FnMut(IoKind, &str)>;

/// Kind of language server stdio given to an IO handler.
#[derive(Debug, Clone, Copy)]
pub enum IoKind {
    StdOut,
    StdIn,
    StdErr,
}

/// Represents a launchable language server. This can either be a standalone binary or the path
/// to a runtime with arguments to instruct it to launch the actual language server file.
#[derive(Clone, Serialize)]
pub struct LanguageServerBinary {
    pub path: PathBuf,
    pub arguments: Vec<OsString>,
    pub env: Option<HashMap<String, String>>,
}

/// Configures the search (and installation) of language servers.
#[derive(Debug, Clone)]
pub struct LanguageServerBinaryOptions {
    /// Whether the adapter should look at the users system
    pub allow_path_lookup: bool,
    /// Whether the adapter should download its own version
    pub allow_binary_download: bool,
    /// Whether the adapter should download a pre-release version
    pub pre_release: bool,
}

struct NotificationSerializer(Box<dyn FnOnce() -> String + Send + Sync>);

/// A running language server process.
pub struct LanguageServer {
    server_id: LanguageServerId,
    next_id: AtomicI32,
    outbound_tx: channel::Sender<String>,
    notification_tx: channel::Sender<NotificationSerializer>,
    name: LanguageServerName,
    version: Option<SharedString>,
    process_name: Arc<str>,
    binary: LanguageServerBinary,
    capabilities: RwLock<ServerCapabilities>,
    /// Configuration sent to the server, stored for display in the language server logs
    /// buffer. This is represented as the message sent to the LSP in order to avoid cloning it (can
    /// be large in cases like sending schemas to the json server).
    configuration: Arc<DidChangeConfigurationParams>,
    code_action_kinds: Option<Vec<CodeActionKind>>,
    notification_handlers: Arc<Mutex<HashMap<&'static str, NotificationHandler>>>,
    response_handlers: Arc<Mutex<Option<HashMap<RequestId, ResponseHandler>>>>,
    /// Tasks spawned by `on_custom_request` to compute responses. Tracked so that
    /// incoming `$/cancelRequest` notifications can cancel them by dropping the task.
    pending_respond_tasks: PendingRespondTasks,
    io_handlers: Arc<Mutex<HashMap<i32, IoHandler>>>,
    executor: BackgroundExecutor,
    #[allow(clippy::type_complexity)]
    io_tasks: Mutex<Option<(Task<Option<()>>, Task<Option<()>>)>>,
    output_done_rx: Mutex<Option<barrier::Receiver>>,
    server: Arc<Mutex<Option<Child>>>,
    workspace_folders: Option<Arc<Mutex<BTreeSet<Uri>>>>,
    root_uri: Uri,
}

#[derive(Clone, Debug, PartialEq, Eq, Hash)]
pub enum LanguageServerSelector {
    Id(LanguageServerId),
    Name(LanguageServerName),
}

/// Identifies a running language server.
#[derive(Copy, Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
#[repr(transparent)]
pub struct LanguageServerId(pub usize);

impl LanguageServerId {
    pub fn from_proto(id: u64) -> Self {
        Self(id as usize)
    }

    pub fn to_proto(self) -> u64 {
        self.0 as u64
    }
}

/// A name of a language server.
#[derive(
    Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Deserialize, Serialize, JsonSchema,
)]
#[serde(transparent)]
pub struct LanguageServerName(pub SharedString);

impl std::fmt::Display for LanguageServerName {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        std::fmt::Display::fmt(&self.0, f)
    }
}

impl AsRef<str> for LanguageServerName {
    fn as_ref(&self) -> &str {
        self.0.as_ref()
    }
}

impl AsRef<OsStr> for LanguageServerName {
    fn as_ref(&self) -> &OsStr {
        self.0.as_ref().as_ref()
    }
}

impl LanguageServerName {
    pub const fn new_static(s: &'static str) -> Self {
        Self(SharedString::new_static(s))
    }

    pub fn from_proto(s: String) -> Self {
        Self(s.into())
    }
}

impl<'a> From<&'a str> for LanguageServerName {
    fn from(str: &'a str) -> LanguageServerName {
        LanguageServerName(str.to_string().into())
    }
}

impl PartialEq<str> for LanguageServerName {
    fn eq(&self, other: &str) -> bool {
        self.0 == other
    }
}

/// Handle to a language server RPC activity subscription.
pub enum Subscription {
    Notification {
        method: &'static str,
        notification_handlers: Option<Arc<Mutex<HashMap<&'static str, NotificationHandler>>>>,
    },
    Io {
        id: i32,
        io_handlers: Option<Weak<Mutex<HashMap<i32, IoHandler>>>>,
    },
}

/// Language server protocol RPC request message ID.
///
/// [LSP Specification](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#requestMessage)
#[derive(Debug, Clone, Eq, PartialEq, Hash, Serialize, Deserialize)]
#[serde(untagged)]
pub enum RequestId {
    Int(i32),
    Str(String),
}

fn is_unit<T: 'static>(_: &T) -> bool {
    TypeId::of::<T>() == TypeId::of::<()>()
}

/// Language server protocol RPC request message.
///
/// [LSP Specification](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#requestMessage)
#[derive(Serialize, Deserialize)]
pub struct Request<'a, T>
where
    T: 'static,
{
    jsonrpc: &'static str,
    id: RequestId,
    method: &'a str,
    #[serde(default, skip_serializing_if = "is_unit")]
    params: T,
}

/// Language server protocol RPC request response message before it is deserialized into a concrete type.
#[derive(Serialize, Deserialize)]
struct AnyResponse<'a> {
    jsonrpc: &'a str,
    id: RequestId,
    #[serde(default)]
    error: Option<Error>,
    #[serde(borrow)]
    result: Option<&'a RawValue>,
}

/// Language server protocol RPC request response message.
///
/// [LSP Specification](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#responseMessage)
#[derive(Serialize)]
struct Response<T> {
    jsonrpc: &'static str,
    id: RequestId,
    #[serde(flatten)]
    value: LspResult<T>,
}

#[derive(Serialize)]
#[serde(rename_all = "snake_case")]
enum LspResult<T> {
    #[serde(rename = "result")]
    Ok(Option<T>),
    Error(Option<Error>),
}

/// Language server protocol RPC notification message.
///
/// [LSP Specification](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#notificationMessage)
#[derive(Serialize, Deserialize)]
struct Notification<'a, T>
where
    T: 'static,
{
    jsonrpc: &'static str,
    #[serde(borrow)]
    method: &'a str,
    #[serde(default, skip_serializing_if = "is_unit")]
    params: T,
}

/// Language server RPC notification message before it is deserialized into a concrete type.
#[derive(Debug, Clone, Deserialize)]
struct NotificationOrRequest {
    #[serde(default)]
    id: Option<RequestId>,
    method: String,
    #[serde(default)]
    params: Option<Value>,
}

#[derive(Debug, Serialize, Deserialize)]
struct Error {
    code: i64,
    message: String,
    #[serde(default)]
    data: Option<serde_json::Value>,
}

pub trait LspRequestFuture<O>: Future<Output = ConnectionResult<O>> {
    fn id(&self) -> i32;
}

struct LspRequest<F> {
    id: i32,
    request: F,
}

impl<F> LspRequest<F> {
    pub fn new(id: i32, request: F) -> Self {
        Self { id, request }
    }
}

impl<F: Future> Future for LspRequest<F> {
    type Output = F::Output;

    fn poll(self: Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll<Self::Output> {
        // SAFETY: This is standard pin projection, we're pinned so our fields must be pinned.
        let inner = unsafe { Pin::new_unchecked(&mut self.get_unchecked_mut().request) };
        inner.poll(cx)
    }
}

impl<F, O> LspRequestFuture<O> for LspRequest<F>
where
    F: Future<Output = ConnectionResult<O>>,
{
    fn id(&self) -> i32 {
        self.id
    }
}

/// Combined capabilities of the server and the adapter.
#[derive(Debug, Clone)]
pub struct AdapterServerCapabilities {
    // Reported capabilities by the server
    pub server_capabilities: ServerCapabilities,
    // List of code actions supported by the LspAdapter matching the server
    pub code_action_kinds: Option<Vec<CodeActionKind>>,
}

// See the VSCode docs [1] and the LSP Spec [2]
//
// [1]: https://code.visualstudio.com/api/language-extensions/semantic-highlight-guide#standard-token-types-and-modifiers
// [2]: https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#semanticTokenTypes
pub const SEMANTIC_TOKEN_TYPES: &[SemanticTokenType] = &[
    SemanticTokenType::NAMESPACE,
    SemanticTokenType::CLASS,
    SemanticTokenType::ENUM,
    SemanticTokenType::INTERFACE,
    SemanticTokenType::STRUCT,
    SemanticTokenType::TYPE_PARAMETER,
    SemanticTokenType::TYPE,
    SemanticTokenType::PARAMETER,
    SemanticTokenType::VARIABLE,
    SemanticTokenType::PROPERTY,
    SemanticTokenType::ENUM_MEMBER,
    SemanticTokenType::DECORATOR,
    SemanticTokenType::FUNCTION,
    SemanticTokenType::METHOD,
    SemanticTokenType::MACRO,
    SemanticTokenType::new("label"), // Not in the spec, but in the docs.
    SemanticTokenType::COMMENT,
    SemanticTokenType::STRING,
    SemanticTokenType::KEYWORD,
    SemanticTokenType::NUMBER,
    SemanticTokenType::REGEXP,
    SemanticTokenType::OPERATOR,
    SemanticTokenType::MODIFIER, // Only in the spec, not in the docs.
    // Language specific things below.
    // C#
    SemanticTokenType::EVENT,
    // Rust
    SemanticTokenType::new("lifetime"),
];
pub const SEMANTIC_TOKEN_MODIFIERS: &[SemanticTokenModifier] = &[
    SemanticTokenModifier::DECLARATION,
    SemanticTokenModifier::DEFINITION,
    SemanticTokenModifier::READONLY,
    SemanticTokenModifier::STATIC,
    SemanticTokenModifier::DEPRECATED,
    SemanticTokenModifier::ABSTRACT,
    SemanticTokenModifier::ASYNC,
    SemanticTokenModifier::MODIFICATION,
    SemanticTokenModifier::DOCUMENTATION,
    SemanticTokenModifier::DEFAULT_LIBRARY,
    // Language specific things below.
    // Rust
    SemanticTokenModifier::new("constant"),
];

impl LanguageServer {
    /// Starts a language server process.
    /// A request_timeout of zero or Duration::MAX indicates an indefinite timeout.
    pub fn new(
        stderr_capture: Arc<Mutex<Option<String>>>,
        server_id: LanguageServerId,
        server_name: LanguageServerName,
        binary: LanguageServerBinary,
        root_path: &Path,
        code_action_kinds: Option<Vec<CodeActionKind>>,
        workspace_folders: Option<Arc<Mutex<BTreeSet<Uri>>>>,
        cx: &mut AsyncApp,
    ) -> Result<Self> {
        let working_dir = if root_path.is_dir() {
            root_path
        } else {
            root_path.parent().unwrap_or_else(|| Path::new("/"))
        };
        let root_uri = Uri::from_file_path(&working_dir)
            .map_err(|()| anyhow!("{working_dir:?} is not a valid URI"))?;
        log::info!(
            "starting language server process. binary path: \
            {:?}, working directory: {:?}, args: {:?}",
            binary.path,
            working_dir,
            &binary.arguments
        );
        let mut command = util::command::new_command(&binary.path);
        command
            .current_dir(working_dir)
            .args(&binary.arguments)
            .envs(binary.env.clone().unwrap_or_default())
            .stdin(Stdio::piped())
            .stdout(Stdio::piped())
            .stderr(Stdio::piped())
            .kill_on_drop(true);

        let mut server = command
            .spawn()
            .with_context(|| format!("failed to spawn command {command:?}",))?;

        let stdin = server.stdin.take().unwrap();
        let stdout = server.stdout.take().unwrap();
        let stderr = server.stderr.take().unwrap();
        let server = Self::new_internal(
            server_id,
            server_name,
            stdin,
            stdout,
            Some(stderr),
            stderr_capture,
            Some(server),
            code_action_kinds,
            binary,
            root_uri,
            workspace_folders,
            cx,
            move |notification| {
                log::info!(
                    "Language server with id {} sent unhandled notification {}:\n{}",
                    server_id,
                    notification.method,
                    serde_json::to_string_pretty(&notification.params).unwrap(),
                );
                false
            },
        );

        Ok(server)
    }

    fn new_internal<Stdin, Stdout, Stderr, F>(
        server_id: LanguageServerId,
        server_name: LanguageServerName,
        stdin: Stdin,
        stdout: Stdout,
        stderr: Option<Stderr>,
        stderr_capture: Arc<Mutex<Option<String>>>,
        server: Option<Child>,
        code_action_kinds: Option<Vec<CodeActionKind>>,
        binary: LanguageServerBinary,
        root_uri: Uri,
        workspace_folders: Option<Arc<Mutex<BTreeSet<Uri>>>>,
        cx: &mut AsyncApp,
        on_unhandled_notification: F,
    ) -> Self
    where
        Stdin: AsyncWrite + Unpin + Send + 'static,
        Stdout: AsyncRead + Unpin + Send + 'static,
        Stderr: AsyncRead + Unpin + Send + 'static,
        F: Fn(&NotificationOrRequest) -> bool + 'static + Send + Sync + Clone,
    {
        let (outbound_tx, outbound_rx) = channel::unbounded::<String>();
        let (output_done_tx, output_done_rx) = barrier::channel();
        let notification_handlers =
            Arc::new(Mutex::new(HashMap::<_, NotificationHandler>::default()));
        let response_handlers =
            Arc::new(Mutex::new(Some(HashMap::<_, ResponseHandler>::default())));
        let pending_respond_tasks = PendingRespondTasks::default();
        let io_handlers = Arc::new(Mutex::new(HashMap::default()));

        let stdout_input_task = cx.spawn({
            let unhandled_notification_wrapper = {
                let response_channel = outbound_tx.clone();
                async move |msg: NotificationOrRequest| {
                    let did_handle = on_unhandled_notification(&msg);
                    if !did_handle && let Some(message_id) = msg.id {
                        let response = AnyResponse {
                            jsonrpc: JSON_RPC_VERSION,
                            id: message_id,
                            error: Some(Error {
                                code: -32601,
                                message: format!("Unrecognized method `{}`", msg.method),
                                data: None,
                            }),
                            result: None,
                        };
                        if let Ok(response) = serde_json::to_string(&response) {
                            response_channel.send(response).await.ok();
                        }
                    }
                }
            };
            let notification_handlers = notification_handlers.clone();
            let response_handlers = response_handlers.clone();
            let io_handlers = io_handlers.clone();
            let pending_respond_tasks = pending_respond_tasks.clone();
            async move |cx| {
                Self::handle_incoming_messages(
                    stdout,
                    unhandled_notification_wrapper,
                    notification_handlers,
                    response_handlers,
                    pending_respond_tasks,
                    io_handlers,
                    cx,
                )
                .log_err()
                .await
            }
        });
        let stderr_input_task = stderr
            .map(|stderr| {
                let io_handlers = io_handlers.clone();
                let stderr_captures = stderr_capture.clone();
                cx.background_spawn(async move {
                    Self::handle_stderr(stderr, io_handlers, stderr_captures)
                        .log_err()
                        .await
                })
            })
            .unwrap_or_else(|| Task::ready(None));
        let input_task = cx.background_spawn(async move {
            let (stdout, stderr) = futures::join!(stdout_input_task, stderr_input_task);
            stdout.or(stderr)
        });
        let output_task = cx.background_spawn({
            Self::handle_outgoing_messages(
                stdin,
                outbound_rx,
                output_done_tx,
                response_handlers.clone(),
                io_handlers.clone(),
            )
            .log_err()
        });

        let configuration = DidChangeConfigurationParams {
            settings: Value::Null,
        }
        .into();

        let (notification_tx, notification_rx) = channel::unbounded::<NotificationSerializer>();
        cx.background_spawn({
            let outbound_tx = outbound_tx.clone();
            async move {
                while let Ok(serializer) = notification_rx.recv().await {
                    let serialized = (serializer.0)();
                    let Ok(_) = outbound_tx.send(serialized).await else {
                        return;
                    };
                }
                outbound_tx.close();
            }
        })
        .detach();
        Self {
            server_id,
            notification_handlers,
            notification_tx,
            response_handlers,
            pending_respond_tasks,
            io_handlers,
            name: server_name,
            version: None,
            process_name: binary
                .path
                .file_name()
                .map(|name| Arc::from(name.to_string_lossy()))
                .unwrap_or_default(),
            binary,
            capabilities: Default::default(),
            configuration,
            code_action_kinds,
            next_id: Default::default(),
            outbound_tx,
            executor: cx.background_executor().clone(),
            io_tasks: Mutex::new(Some((input_task, output_task))),
            output_done_rx: Mutex::new(Some(output_done_rx)),
            server: Arc::new(Mutex::new(server)),
            workspace_folders,
            root_uri,
        }
    }

    /// List of code action kinds this language server reports being able to emit.
    pub fn code_action_kinds(&self) -> Option<Vec<CodeActionKind>> {
        self.code_action_kinds.clone()
    }

    async fn handle_incoming_messages<Stdout>(
        stdout: Stdout,
        on_unhandled_notification: impl AsyncFn(NotificationOrRequest) + 'static + Send,
        notification_handlers: Arc<Mutex<HashMap<&'static str, NotificationHandler>>>,
        response_handlers: Arc<Mutex<Option<HashMap<RequestId, ResponseHandler>>>>,
        pending_respond_tasks: PendingRespondTasks,
        io_handlers: Arc<Mutex<HashMap<i32, IoHandler>>>,
        cx: &mut AsyncApp,
    ) -> anyhow::Result<()>
    where
        Stdout: AsyncRead + Unpin + Send + 'static,
    {
        use smol::stream::StreamExt;
        let stdout = BufReader::new(stdout);
        let _clear_response_handlers = util::defer({
            let response_handlers = response_handlers.clone();
            move || {
                response_handlers.lock().take();
            }
        });
        let mut input_handler = input_handler::LspStdoutHandler::new(
            stdout,
            response_handlers,
            io_handlers,
            cx.background_executor().clone(),
        );

        while let Some(msg) = input_handler.incoming_messages.next().await {
            if msg.method == <notification::Cancel as notification::Notification>::METHOD {
                if let Some(params) = msg.params {
                    if let Ok(cancel_params) = serde_json::from_value::<CancelParams>(params) {
                        let id = match cancel_params.id {
                            NumberOrString::Number(id) => RequestId::Int(id),
                            NumberOrString::String(id) => RequestId::Str(id),
                        };
                        pending_respond_tasks.lock().remove(&id);
                    }
                }
                continue;
            }

            let unhandled_message = {
                let mut notification_handlers = notification_handlers.lock();
                if let Some(handler) = notification_handlers.get_mut(msg.method.as_str()) {
                    handler(msg.id, msg.params.unwrap_or(Value::Null), cx);
                    None
                } else {
                    Some(msg)
                }
            };

            if let Some(msg) = unhandled_message {
                on_unhandled_notification(msg).await;
            }

            // Don't starve the main thread when receiving lots of notifications at once.
            smol::future::yield_now().await;
        }
        input_handler.loop_handle.await
    }

    async fn handle_stderr<Stderr>(
        stderr: Stderr,
        io_handlers: Arc<Mutex<HashMap<i32, IoHandler>>>,
        stderr_capture: Arc<Mutex<Option<String>>>,
    ) -> anyhow::Result<()>
    where
        Stderr: AsyncRead + Unpin + Send + 'static,
    {
        let mut stderr = BufReader::new(stderr);
        let mut buffer = Vec::new();

        loop {
            buffer.clear();

            let bytes_read = stderr.read_until(b'\n', &mut buffer).await?;
            if bytes_read == 0 {
                return Ok(());
            }

            if let Ok(message) = std::str::from_utf8(&buffer) {
                log::trace!("incoming stderr message:{message}");
                for handler in io_handlers.lock().values_mut() {
                    handler(IoKind::StdErr, message);
                }

                if let Some(stderr) = stderr_capture.lock().as_mut() {
                    stderr.push_str(message);
                }
            }

            // Don't starve the main thread when receiving lots of messages at once.
            smol::future::yield_now().await;
        }
    }

    async fn handle_outgoing_messages<Stdin>(
        stdin: Stdin,
        outbound_rx: channel::Receiver<String>,
        output_done_tx: barrier::Sender,
        response_handlers: Arc<Mutex<Option<HashMap<RequestId, ResponseHandler>>>>,
        io_handlers: Arc<Mutex<HashMap<i32, IoHandler>>>,
    ) -> anyhow::Result<()>
    where
        Stdin: AsyncWrite + Unpin + Send + 'static,
    {
        let mut stdin = BufWriter::new(stdin);
        let _clear_response_handlers = util::defer({
            let response_handlers = response_handlers.clone();
            move || {
                response_handlers.lock().take();
            }
        });
        let mut content_len_buffer = Vec::new();
        while let Ok(message) = outbound_rx.recv().await {
            log::trace!("outgoing message:{}", message);
            for handler in io_handlers.lock().values_mut() {
                handler(IoKind::StdIn, &message);
            }

            content_len_buffer.clear();
            write!(content_len_buffer, "{}", message.len()).unwrap();
            stdin.write_all(CONTENT_LEN_HEADER.as_bytes()).await?;
            stdin.write_all(&content_len_buffer).await?;
            stdin.write_all("\r\n\r\n".as_bytes()).await?;
            stdin.write_all(message.as_bytes()).await?;
            stdin.flush().await?;
        }
        drop(output_done_tx);
        Ok(())
    }

    pub fn default_initialize_params(
        &self,
        pull_diagnostics: bool,
        augments_syntax_tokens: bool,
        cx: &App,
    ) -> InitializeParams {
        let workspace_folders = self.workspace_folders.as_ref().map_or_else(
            || {
                vec![WorkspaceFolder {
                    name: Default::default(),
                    uri: self.root_uri.clone(),
                }]
            },
            |folders| {
                folders
                    .lock()
                    .iter()
                    .cloned()
                    .map(|uri| WorkspaceFolder {
                        name: Default::default(),
                        uri,
                    })
                    .collect()
            },
        );

        #[allow(deprecated)]
        InitializeParams {
            process_id: Some(std::process::id()),
            root_path: Some(
                self.root_uri
                    .to_file_path()
                    .map(|path| path.to_string_lossy().into_owned())
                    .unwrap_or_else(|_| self.root_uri.path().to_string()),
            ),
            root_uri: Some(self.root_uri.clone()),
            initialization_options: None,
            capabilities: ClientCapabilities {
                general: Some(GeneralClientCapabilities {
                    position_encodings: Some(vec![PositionEncodingKind::UTF16]),
                    ..GeneralClientCapabilities::default()
                }),
                workspace: Some(WorkspaceClientCapabilities {
                    configuration: Some(true),
                    did_change_watched_files: Some(DidChangeWatchedFilesClientCapabilities {
                        dynamic_registration: Some(true),
                        relative_pattern_support: Some(true),
                    }),
                    did_change_configuration: Some(DynamicRegistrationClientCapabilities {
                        dynamic_registration: Some(true),
                    }),
                    workspace_folders: Some(true),
                    symbol: Some(WorkspaceSymbolClientCapabilities {
                        resolve_support: None,
                        dynamic_registration: Some(true),
                        ..WorkspaceSymbolClientCapabilities::default()
                    }),
                    inlay_hint: Some(InlayHintWorkspaceClientCapabilities {
                        refresh_support: Some(true),
                    }),
                    diagnostics: Some(DiagnosticWorkspaceClientCapabilities {
                        refresh_support: Some(true),
                    })
                    .filter(|_| pull_diagnostics),
                    code_lens: Some(CodeLensWorkspaceClientCapabilities {
                        refresh_support: Some(true),
                    }),
                    workspace_edit: Some(WorkspaceEditClientCapabilities {
                        resource_operations: Some(vec![
                            ResourceOperationKind::Create,
                            ResourceOperationKind::Rename,
                            ResourceOperationKind::Delete,
                        ]),
                        document_changes: Some(true),
                        snippet_edit_support: Some(true),
                        ..WorkspaceEditClientCapabilities::default()
                    }),
                    file_operations: Some(WorkspaceFileOperationsClientCapabilities {
                        dynamic_registration: Some(true),
                        did_rename: Some(true),
                        will_rename: Some(true),
                        ..WorkspaceFileOperationsClientCapabilities::default()
                    }),
                    apply_edit: Some(true),
                    execute_command: Some(ExecuteCommandClientCapabilities {
                        dynamic_registration: Some(true),
                    }),
                    semantic_tokens: Some(SemanticTokensWorkspaceClientCapabilities {
                        refresh_support: Some(true),
                    }),
                    ..WorkspaceClientCapabilities::default()
                }),
                text_document: Some(TextDocumentClientCapabilities {
                    definition: Some(GotoCapability {
                        link_support: Some(true),
                        dynamic_registration: Some(true),
                    }),
                    code_action: Some(CodeActionClientCapabilities {
                        code_action_literal_support: Some(CodeActionLiteralSupport {
                            code_action_kind: CodeActionKindLiteralSupport {
                                value_set: vec![
                                    CodeActionKind::REFACTOR.as_str().into(),
                                    CodeActionKind::QUICKFIX.as_str().into(),
                                    CodeActionKind::SOURCE.as_str().into(),
                                ],
                            },
                        }),
                        data_support: Some(true),
                        resolve_support: Some(CodeActionCapabilityResolveSupport {
                            properties: vec![
                                "kind".to_string(),
                                "diagnostics".to_string(),
                                "isPreferred".to_string(),
                                "disabled".to_string(),
                                "edit".to_string(),
                                "command".to_string(),
                            ],
                        }),
                        dynamic_registration: Some(true),
                        ..CodeActionClientCapabilities::default()
                    }),
                    completion: Some(CompletionClientCapabilities {
                        completion_item: Some(CompletionItemCapability {
                            snippet_support: Some(true),
                            resolve_support: Some(CompletionItemCapabilityResolveSupport {
                                properties: vec![
                                    "additionalTextEdits".to_string(),
                                    "command".to_string(),
                                    "detail".to_string(),
                                    "documentation".to_string(),
                                    // NB: Do not have this resolved, otherwise Zed becomes slow to complete things
                                    // "textEdit".to_string(),
                                ],
                            }),
                            deprecated_support: Some(true),
                            tag_support: Some(TagSupport {
                                value_set: vec![CompletionItemTag::DEPRECATED],
                            }),
                            insert_replace_support: Some(true),
                            label_details_support: Some(true),
                            insert_text_mode_support: Some(InsertTextModeSupport {
                                value_set: vec![
                                    InsertTextMode::AS_IS,
                                    InsertTextMode::ADJUST_INDENTATION,
                                ],
                            }),
                            documentation_format: Some(vec![
                                MarkupKind::Markdown,
                                MarkupKind::PlainText,
                            ]),
                            ..CompletionItemCapability::default()
                        }),
                        insert_text_mode: Some(InsertTextMode::ADJUST_INDENTATION),
                        completion_list: Some(CompletionListCapability {
                            item_defaults: Some(vec![
                                "commitCharacters".to_owned(),
                                "editRange".to_owned(),
                                "insertTextMode".to_owned(),
                                "insertTextFormat".to_owned(),
                                "data".to_owned(),
                            ]),
                        }),
                        context_support: Some(true),
                        dynamic_registration: Some(true),
                        ..CompletionClientCapabilities::default()
                    }),
                    rename: Some(RenameClientCapabilities {
                        prepare_support: Some(true),
                        prepare_support_default_behavior: Some(
                            PrepareSupportDefaultBehavior::IDENTIFIER,
                        ),
                        dynamic_registration: Some(true),
                        ..RenameClientCapabilities::default()
                    }),
                    hover: Some(HoverClientCapabilities {
                        content_format: Some(vec![MarkupKind::Markdown]),
                        dynamic_registration: Some(true),
                    }),
                    inlay_hint: Some(InlayHintClientCapabilities {
                        resolve_support: Some(InlayHintResolveClientCapabilities {
                            properties: vec![
                                "textEdits".to_string(),
                                "tooltip".to_string(),
                                "label.tooltip".to_string(),
                                "label.location".to_string(),
                                "label.command".to_string(),
                            ],
                        }),
                        dynamic_registration: Some(true),
                    }),
                    semantic_tokens: Some(SemanticTokensClientCapabilities {
                        dynamic_registration: Some(false),
                        requests: SemanticTokensClientCapabilitiesRequests {
                            range: None,
                            full: Some(SemanticTokensFullOptions::Delta { delta: Some(true) }),
                        },
                        token_types: SEMANTIC_TOKEN_TYPES.to_vec(),
                        token_modifiers: SEMANTIC_TOKEN_MODIFIERS.to_vec(),
                        formats: vec![TokenFormat::RELATIVE],
                        overlapping_token_support: Some(true),
                        multiline_token_support: Some(true),
                        server_cancel_support: Some(true),
                        augments_syntax_tokens: Some(augments_syntax_tokens),
                    }),
                    publish_diagnostics: Some(PublishDiagnosticsClientCapabilities {
                        related_information: Some(true),
                        version_support: Some(true),
                        data_support: Some(true),
                        tag_support: Some(TagSupport {
                            value_set: vec![DiagnosticTag::UNNECESSARY, DiagnosticTag::DEPRECATED],
                        }),
                        code_description_support: Some(true),
                    }),
                    formatting: Some(DynamicRegistrationClientCapabilities {
                        dynamic_registration: Some(true),
                    }),
                    range_formatting: Some(DynamicRegistrationClientCapabilities {
                        dynamic_registration: Some(true),
                    }),
                    on_type_formatting: Some(DynamicRegistrationClientCapabilities {
                        dynamic_registration: Some(true),
                    }),
                    signature_help: Some(SignatureHelpClientCapabilities {
                        signature_information: Some(SignatureInformationSettings {
                            documentation_format: Some(vec![
                                MarkupKind::Markdown,
                                MarkupKind::PlainText,
                            ]),
                            parameter_information: Some(ParameterInformationSettings {
                                label_offset_support: Some(true),
                            }),
                            active_parameter_support: Some(true),
                        }),
                        dynamic_registration: Some(true),
                        ..SignatureHelpClientCapabilities::default()
                    }),
                    synchronization: Some(TextDocumentSyncClientCapabilities {
                        did_save: Some(true),
                        dynamic_registration: Some(true),
                        ..TextDocumentSyncClientCapabilities::default()
                    }),
                    code_lens: Some(CodeLensClientCapabilities {
                        dynamic_registration: Some(true),
                    }),
                    document_symbol: Some(DocumentSymbolClientCapabilities {
                        hierarchical_document_symbol_support: Some(true),
                        dynamic_registration: Some(true),
                        ..DocumentSymbolClientCapabilities::default()
                    }),
                    diagnostic: Some(DiagnosticClientCapabilities {
                        dynamic_registration: Some(true),
                        related_document_support: Some(true),
                    })
                    .filter(|_| pull_diagnostics),
                    color_provider: Some(DocumentColorClientCapabilities {
                        dynamic_registration: Some(true),
                    }),
                    folding_range: Some(FoldingRangeClientCapabilities {
                        dynamic_registration: Some(true),
                        line_folding_only: Some(false),
                        range_limit: None,
                        folding_range: Some(FoldingRangeCapability {
                            collapsed_text: Some(true),
                        }),
                        folding_range_kind: Some(FoldingRangeKindCapability {
                            value_set: Some(vec![
                                FoldingRangeKind::Comment,
                                FoldingRangeKind::Region,
                                FoldingRangeKind::Imports,
                            ]),
                        }),
                    }),
                    ..TextDocumentClientCapabilities::default()
                }),
                experimental: Some(json!({
                    "serverStatusNotification": true,
                    "localDocs": true,
                })),
                window: Some(WindowClientCapabilities {
                    work_done_progress: Some(true),
                    show_message: Some(ShowMessageRequestClientCapabilities {
                        message_action_item: Some(MessageActionItemCapabilities {
                            additional_properties_support: Some(true),
                        }),
                    }),
                    ..WindowClientCapabilities::default()
                }),
            },
            trace: None,
            workspace_folders: Some(workspace_folders),
            client_info: release_channel::ReleaseChannel::try_global(cx).map(|release_channel| {
                ClientInfo {
                    name: release_channel.display_name().to_string(),
                    version: Some(release_channel::AppVersion::global(cx).to_string()),
                }
            }),
            locale: None,
            ..InitializeParams::default()
        }
    }

    /// Initializes a language server by sending the `Initialize` request.
    /// Note that `options` is used directly to construct [`InitializeParams`], which is why it is owned.
    ///
    /// [LSP Specification](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#initialize)
    pub fn initialize(
        mut self,
        params: InitializeParams,
        configuration: Arc<DidChangeConfigurationParams>,
        timeout: Duration,
        cx: &App,
    ) -> Task<Result<Arc<Self>>> {
        cx.background_spawn(async move {
            let response = self
                .request::<request::Initialize>(params, timeout)
                .await
                .into_response()
                .with_context(|| {
                    format!(
                        "initializing server {}, id {}",
                        self.name(),
                        self.server_id()
                    )
                })?;
            if let Some(info) = response.server_info {
                self.version = info.version.map(SharedString::from);
                self.process_name = info.name.into();
            }
            self.capabilities = RwLock::new(response.capabilities);
            self.configuration = configuration;

            self.notify::<notification::Initialized>(InitializedParams {})?;
            Ok(Arc::new(self))
        })
    }

    /// Sends a shutdown request to the language server process and prepares the [`LanguageServer`] to be dropped.
    pub fn shutdown(&self) -> Option<impl 'static + Send + Future<Output = Option<()>> + use<>> {
        let tasks = self.io_tasks.lock().take()?;

        let response_handlers = self.response_handlers.clone();
        let next_id = AtomicI32::new(self.next_id.load(SeqCst));
        let outbound_tx = self.outbound_tx.clone();
        let executor = self.executor.clone();
        let notification_serializers = self.notification_tx.clone();
        let mut output_done = self.output_done_rx.lock().take().unwrap();
        let shutdown_request = Self::request_internal::<request::Shutdown>(
            &next_id,
            &response_handlers,
            &outbound_tx,
            &notification_serializers,
            &executor,
            SERVER_SHUTDOWN_TIMEOUT,
            (),
        );

        let server = self.server.clone();
        let name = self.name.clone();
        let server_id = self.server_id;
        let mut timer = self.executor.timer(SERVER_SHUTDOWN_TIMEOUT).fuse();
        Some(async move {
            log::debug!("language server shutdown started");

            select! {
                request_result = shutdown_request.fuse() => {
                    match request_result {
                        ConnectionResult::Timeout => {
                            log::warn!("timeout waiting for language server {name} (id {server_id}) to shutdown");
                        },
                        ConnectionResult::ConnectionReset => {
                            log::warn!("language server {name} (id {server_id}) closed the shutdown request connection");
                        },
                        ConnectionResult::Result(Err(e)) => {
                            log::error!("Shutdown request failure, server {name} (id {server_id}): {e:#}");
                        },
                        ConnectionResult::Result(Ok(())) => {}
                    }
                }

                _ = timer => {
                    log::info!("timeout waiting for language server {name} (id {server_id}) to shutdown");
                },
            }

            response_handlers.lock().take();
            Self::notify_internal::<notification::Exit>(&notification_serializers, ()).ok();
            notification_serializers.close();
            output_done.recv().await;
            server.lock().take().map(|mut child| child.kill());
            drop(tasks);
            log::debug!("language server shutdown finished");
            Some(())
        })
    }

    /// Register a handler to handle incoming LSP notifications.
    ///
    /// [LSP Specification](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#notificationMessage)
    #[must_use]
    pub fn on_notification<T, F>(&self, f: F) -> Subscription
    where
        T: notification::Notification,
        F: 'static + Send + FnMut(T::Params, &mut AsyncApp),
    {
        self.on_custom_notification(T::METHOD, f)
    }

    /// Register a handler to handle incoming LSP requests.
    ///
    /// [LSP Specification](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#requestMessage)
    #[must_use]
    pub fn on_request<T, F, Fut>(&self, f: F) -> Subscription
    where
        T: request::Request,
        T::Params: 'static + Send,
        F: 'static + FnMut(T::Params, &mut AsyncApp) -> Fut + Send,
        Fut: 'static + Future<Output = Result<T::Result>>,
    {
        self.on_custom_request(T::METHOD, f)
    }

    /// Registers a handler to inspect all language server process stdio.
    #[must_use]
    pub fn on_io<F>(&self, f: F) -> Subscription
    where
        F: 'static + Send + FnMut(IoKind, &str),
    {
        let id = self.next_id.fetch_add(1, SeqCst);
        self.io_handlers.lock().insert(id, Box::new(f));
        Subscription::Io {
            id,
            io_handlers: Some(Arc::downgrade(&self.io_handlers)),
        }
    }

    /// Removes a request handler registers via [`Self::on_request`].
    pub fn remove_request_handler<T: request::Request>(&self) {
        self.notification_handlers.lock().remove(T::METHOD);
    }

    /// Removes a notification handler registers via [`Self::on_notification`].
    pub fn remove_notification_handler<T: notification::Notification>(&self) {
        self.notification_handlers.lock().remove(T::METHOD);
    }

    /// Checks if a notification handler has been registered via [`Self::on_notification`].
    pub fn has_notification_handler<T: notification::Notification>(&self) -> bool {
        self.notification_handlers.lock().contains_key(T::METHOD)
    }

    #[must_use]
    fn on_custom_notification<Params, F>(&self, method: &'static str, mut f: F) -> Subscription
    where
        F: 'static + FnMut(Params, &mut AsyncApp) + Send,
        Params: DeserializeOwned,
    {
        let prev_handler = self.notification_handlers.lock().insert(
            method,
            Box::new(move |_, params, cx| {
                if let Some(params) = serde_json::from_value(params).log_err() {
                    f(params, cx);
                }
            }),
        );
        assert!(
            prev_handler.is_none(),
            "registered multiple handlers for the same LSP method"
        );
        Subscription::Notification {
            method,
            notification_handlers: Some(self.notification_handlers.clone()),
        }
    }

    #[must_use]
    fn on_custom_request<Params, Res, Fut, F>(&self, method: &'static str, mut f: F) -> Subscription
    where
        F: 'static + FnMut(Params, &mut AsyncApp) -> Fut + Send,
        Fut: 'static + Future<Output = Result<Res>>,
        Params: DeserializeOwned + Send + 'static,
        Res: Serialize,
    {
        let outbound_tx = self.outbound_tx.clone();
        let pending_respond_tasks = self.pending_respond_tasks.clone();
        let prev_handler = self.notification_handlers.lock().insert(
            method,
            Box::new(move |id, params, cx| {
                if let Some(id) = id {
                    match serde_json::from_value(params) {
                        Ok(params) => {
                            let response = f(params, cx);
                            let task = cx.foreground_executor().spawn({
                                let outbound_tx = outbound_tx.clone();
                                let pending_respond_tasks = pending_respond_tasks.clone();
                                let id = id.clone();
                                async move {
                                    let response = match response.await {
                                        Ok(result) => Response {
                                            jsonrpc: JSON_RPC_VERSION,
                                            id: id.clone(),
                                            value: LspResult::Ok(Some(result)),
                                        },
                                        Err(error) => Response {
                                            jsonrpc: JSON_RPC_VERSION,
                                            id: id.clone(),
                                            value: LspResult::Error(Some(Error {
                                                code: lsp_types::error_codes::REQUEST_FAILED,
                                                message: error.to_string(),
                                                data: None,
                                            })),
                                        },
                                    };
                                    if let Some(response) =
                                        serde_json::to_string(&response).log_err()
                                    {
                                        outbound_tx.try_send(response).ok();
                                    }
                                    pending_respond_tasks.lock().remove(&id);
                                }
                            });
                            pending_respond_tasks.lock().insert(id, task);
                        }

                        Err(error) => {
                            log::error!("error deserializing {} request: {:?}", method, error);
                            let response = AnyResponse {
                                jsonrpc: JSON_RPC_VERSION,
                                id,
                                result: None,
                                error: Some(Error {
                                    code: -32700, // Parse error
                                    message: error.to_string(),
                                    data: None,
                                }),
                            };
                            if let Some(response) = serde_json::to_string(&response).log_err() {
                                outbound_tx.try_send(response).ok();
                            }
                        }
                    }
                }
            }),
        );
        assert!(
            prev_handler.is_none(),
            "registered multiple handlers for the same LSP method"
        );
        Subscription::Notification {
            method,
            notification_handlers: Some(self.notification_handlers.clone()),
        }
    }

    /// Get the name of the running language server.
    pub fn name(&self) -> LanguageServerName {
        self.name.clone()
    }

    /// Get the version of the running language server.
    pub fn version(&self) -> Option<SharedString> {
        self.version.clone()
    }

    /// Get the readable version of the running language server.
    pub fn readable_version(&self) -> Option<SharedString> {
        match self.name().as_ref() {
            "gopls" => {
                // Gopls returns a detailed JSON object as its version string; we must parse it to extract the semantic version.
                // Example: `{"GoVersion":"go1.26.0","Path":"golang.org/x/tools/gopls","Main":{},"Deps":[],"Settings":[],"Version":"v0.21.1"}`
                self.version
                    .as_ref()
                    .and_then(|obj| {
                        #[derive(Deserialize)]
                        struct GoplsVersion<'a> {
                            #[serde(rename = "Version")]
                            version: &'a str,
                        }
                        let parsed: GoplsVersion = serde_json::from_str(obj.as_str()).ok()?;
                        Some(parsed.version.trim_start_matches("v").to_owned().into())
                    })
                    .or_else(|| self.version.clone())
            }
            _ => self.version.clone(),
        }
    }

    /// Get the process name of the running language server.
    pub fn process_name(&self) -> &str {
        &self.process_name
    }

    /// Get the reported capabilities of the running language server.
    pub fn capabilities(&self) -> ServerCapabilities {
        self.capabilities.read().clone()
    }

    /// Get the reported capabilities of the running language server and
    /// what we know on the client/adapter-side of its capabilities.
    pub fn adapter_server_capabilities(&self) -> AdapterServerCapabilities {
        AdapterServerCapabilities {
            server_capabilities: self.capabilities(),
            code_action_kinds: self.code_action_kinds(),
        }
    }

    /// Update the capabilities of the running language server.
    pub fn update_capabilities(&self, update: impl FnOnce(&mut ServerCapabilities)) {
        update(self.capabilities.write().deref_mut());
    }

    /// Get the individual configuration settings for the running language server.
    /// Does not include globally applied settings (which are stored in ProjectSettings::GlobalLspSettings).
    pub fn configuration(&self) -> &Value {
        &self.configuration.settings
    }

    /// Get the ID of the running language server.
    pub fn server_id(&self) -> LanguageServerId {
        self.server_id
    }

    /// Get the process ID of the running language server, if available.
    pub fn process_id(&self) -> Option<u32> {
        self.server.lock().as_ref().map(|child| child.id())
    }

    /// Get the binary information of the running language server.
    pub fn binary(&self) -> &LanguageServerBinary {
        &self.binary
    }

    /// Send a RPC request to the language server.
    ///
    /// [LSP Specification](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#requestMessage)
    pub fn request<T: request::Request>(
        &self,
        params: T::Params,
        request_timeout: Duration,
    ) -> impl LspRequestFuture<T::Result> + use<T>
    where
        T::Result: 'static + Send,
    {
        Self::request_internal::<T>(
            &self.next_id,
            &self.response_handlers,
            &self.outbound_tx,
            &self.notification_tx,
            &self.executor,
            request_timeout,
            params,
        )
    }

    /// Send a RPC request to the language server with a custom timer.
    /// Once the attached future becomes ready, the request will time out with the provided output message.
    ///
    /// [LSP Specification](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#requestMessage)
    pub fn request_with_timer<T: request::Request, U: Future<Output = String>>(
        &self,
        params: T::Params,
        timer: U,
    ) -> impl LspRequestFuture<T::Result> + use<T, U>
    where
        T::Result: 'static + Send,
    {
        Self::request_internal_with_timer::<T, U>(
            &self.next_id,
            &self.response_handlers,
            &self.outbound_tx,
            &self.notification_tx,
            &self.executor,
            timer,
            params,
        )
    }

    fn request_internal_with_timer<T, U>(
        next_id: &AtomicI32,
        response_handlers: &Arc<Mutex<Option<HashMap<RequestId, ResponseHandler>>>>,
        outbound_tx: &channel::Sender<String>,
        notification_serializers: &channel::Sender<NotificationSerializer>,
        executor: &BackgroundExecutor,
        timer: U,
        params: T::Params,
    ) -> impl LspRequestFuture<T::Result> + use<T, U>
    where
        T::Result: 'static + Send,
        T: request::Request,
        U: Future<Output = String>,
    {
        let id = next_id.fetch_add(1, SeqCst);
        let message = serde_json::to_string(&Request {
            jsonrpc: JSON_RPC_VERSION,
            id: RequestId::Int(id),
            method: T::METHOD,
            params,
        })
        .expect("LSP message should be serializable to JSON");

        let (tx, rx) = oneshot::channel();
        let handle_response = response_handlers
            .lock()
            .as_mut()
            .context("server shut down")
            .map(|handlers| {
                let executor = executor.clone();
                handlers.insert(
                    RequestId::Int(id),
                    Box::new(move |result| {
                        executor
                            .spawn(async move {
                                let response = match result {
                                    Ok(response) => match serde_json::from_str(&response) {
                                        Ok(deserialized) => Ok(deserialized),
                                        Err(error) => {
                                            log::error!("failed to deserialize response from language server: {}. response from language server: {:?}", error, response);
                                            Err(error).context("failed to deserialize response")
                                        }
                                    }
                                    Err(error) => Err(anyhow!("{}", error.message)),
                                };
                                tx.send(response).ok();
                            })
                    }),
                );
            });

        let send = outbound_tx
            .try_send(message)
            .context("failed to write to language server's stdin");

        let response_handlers = Arc::clone(response_handlers);
        let notification_serializers = notification_serializers.downgrade();
        let started = Instant::now();
        LspRequest::new(id, async move {
            if let Err(e) = handle_response {
                return ConnectionResult::Result(Err(e));
            }
            if let Err(e) = send {
                return ConnectionResult::Result(Err(e));
            }

            let cancel_on_drop = util::defer(move || {
                if let Some(notification_serializers) = notification_serializers.upgrade() {
                    Self::notify_internal::<notification::Cancel>(
                        &notification_serializers,
                        CancelParams {
                            id: NumberOrString::Number(id),
                        },
                    )
                    .ok();
                }
            });

            let method = T::METHOD;
            select! {
                response = rx.fuse() => {
                    let elapsed = started.elapsed();
                    log::trace!("Took {elapsed:?} to receive response to {method:?} id {id}");
                    cancel_on_drop.abort();
                    match response {
                        Ok(response_result) => ConnectionResult::Result(response_result),
                        Err(Canceled) => {
                            log::error!("Server reset connection for a request {method:?} id {id}");
                            ConnectionResult::ConnectionReset
                        },
                    }
                }

                message = timer.fuse() => {
                    log::error!("Cancelled LSP request task for {method:?} id {id} {message}");
                    match response_handlers
                        .lock()
                        .as_mut()
                        .context("server shut down") {
                            Ok(handlers) => {
                                handlers.remove(&RequestId::Int(id));
                                ConnectionResult::Timeout
                            }
                            Err(e) => ConnectionResult::Result(Err(e)),
                        }
                }
            }
        })
    }

    fn request_internal<T>(
        next_id: &AtomicI32,
        response_handlers: &Arc<Mutex<Option<HashMap<RequestId, ResponseHandler>>>>,
        outbound_tx: &channel::Sender<String>,
        notification_serializers: &channel::Sender<NotificationSerializer>,
        executor: &BackgroundExecutor,
        request_timeout: Duration,
        params: T::Params,
    ) -> impl LspRequestFuture<T::Result> + use<T>
    where
        T::Result: 'static + Send,
        T: request::Request,
    {
        Self::request_internal_with_timer::<T, _>(
            next_id,
            response_handlers,
            outbound_tx,
            notification_serializers,
            executor,
            Self::request_timeout_future(executor.clone(), request_timeout),
            params,
        )
    }

    /// Internal function to return a Future from a configured timeout duration.
    /// If the duration is zero or `Duration::MAX`, the returned future never completes.
    fn request_timeout_future(
        executor: BackgroundExecutor,
        request_timeout: Duration,
    ) -> impl Future<Output = String> {
        if request_timeout == Duration::MAX || request_timeout == Duration::ZERO {
            return Either::Left(future::pending::<String>());
        }

        Either::Right(
            executor
                .timer(request_timeout)
                .map(move |_| format!("which took over {request_timeout:?}")),
        )
    }

    /// Obtain a request timer for the LSP.
    pub fn request_timer(&self, timeout: Duration) -> impl Future<Output = String> {
        Self::request_timeout_future(self.executor.clone(), timeout)
    }

    /// Sends a RPC notification to the language server.
    ///
    /// [LSP Specification](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#notificationMessage)
    pub fn notify<T: notification::Notification>(&self, params: T::Params) -> Result<()> {
        let outbound = self.notification_tx.clone();
        Self::notify_internal::<T>(&outbound, params)
    }

    fn notify_internal<T: notification::Notification>(
        outbound_tx: &channel::Sender<NotificationSerializer>,
        params: T::Params,
    ) -> Result<()> {
        let serializer = NotificationSerializer(Box::new(move || {
            serde_json::to_string(&Notification {
                jsonrpc: JSON_RPC_VERSION,
                method: T::METHOD,
                params,
            })
            .unwrap()
        }));

        outbound_tx.send_blocking(serializer)?;
        Ok(())
    }

    /// Add new workspace folder to the list.
    pub fn add_workspace_folder(&self, uri: Uri) {
        if self
            .capabilities()
            .workspace
            .and_then(|ws| {
                ws.workspace_folders.and_then(|folders| {
                    folders
                        .change_notifications
                        .map(|caps| matches!(caps, OneOf::Left(false)))
                })
            })
            .unwrap_or(true)
        {
            return;
        }

        let Some(workspace_folders) = self.workspace_folders.as_ref() else {
            return;
        };
        let is_new_folder = workspace_folders.lock().insert(uri.clone());
        if is_new_folder {
            let params = DidChangeWorkspaceFoldersParams {
                event: WorkspaceFoldersChangeEvent {
                    added: vec![WorkspaceFolder {
                        uri,
                        name: String::default(),
                    }],
                    removed: vec![],
                },
            };
            self.notify::<DidChangeWorkspaceFolders>(params).ok();
        }
    }

    /// Remove existing workspace folder from the list.
    pub fn remove_workspace_folder(&self, uri: Uri) {
        if self
            .capabilities()
            .workspace
            .and_then(|ws| {
                ws.workspace_folders.and_then(|folders| {
                    folders
                        .change_notifications
                        .map(|caps| !matches!(caps, OneOf::Left(false)))
                })
            })
            .unwrap_or(true)
        {
            return;
        }
        let Some(workspace_folders) = self.workspace_folders.as_ref() else {
            return;
        };
        let was_removed = workspace_folders.lock().remove(&uri);
        if was_removed {
            let params = DidChangeWorkspaceFoldersParams {
                event: WorkspaceFoldersChangeEvent {
                    added: vec![],
                    removed: vec![WorkspaceFolder {
                        uri,
                        name: String::default(),
                    }],
                },
            };
            self.notify::<DidChangeWorkspaceFolders>(params).ok();
        }
    }
    pub fn set_workspace_folders(&self, folders: BTreeSet<Uri>) {
        let Some(workspace_folders) = self.workspace_folders.as_ref() else {
            return;
        };
        let mut workspace_folders = workspace_folders.lock();

        let old_workspace_folders = std::mem::take(&mut *workspace_folders);
        let added: Vec<_> = folders
            .difference(&old_workspace_folders)
            .map(|uri| WorkspaceFolder {
                uri: uri.clone(),
                name: String::default(),
            })
            .collect();

        let removed: Vec<_> = old_workspace_folders
            .difference(&folders)
            .map(|uri| WorkspaceFolder {
                uri: uri.clone(),
                name: String::default(),
            })
            .collect();
        *workspace_folders = folders;
        let should_notify = !added.is_empty() || !removed.is_empty();
        if should_notify {
            drop(workspace_folders);
            let params = DidChangeWorkspaceFoldersParams {
                event: WorkspaceFoldersChangeEvent { added, removed },
            };
            self.notify::<DidChangeWorkspaceFolders>(params).ok();
        }
    }

    pub fn workspace_folders(&self) -> BTreeSet<Uri> {
        self.workspace_folders.as_ref().map_or_else(
            || BTreeSet::from_iter([self.root_uri.clone()]),
            |folders| folders.lock().clone(),
        )
    }

    pub fn register_buffer(
        &self,
        uri: Uri,
        language_id: String,
        version: i32,
        initial_text: String,
    ) {
        self.notify::<notification::DidOpenTextDocument>(DidOpenTextDocumentParams {
            text_document: TextDocumentItem::new(uri, language_id, version, initial_text),
        })
        .ok();
    }

    pub fn unregister_buffer(&self, uri: Uri) {
        self.notify::<notification::DidCloseTextDocument>(DidCloseTextDocumentParams {
            text_document: TextDocumentIdentifier::new(uri),
        })
        .ok();
    }
}

impl Drop for LanguageServer {
    fn drop(&mut self) {
        if let Some(shutdown) = self.shutdown() {
            self.executor.spawn(shutdown).detach();
        }
    }
}

impl Subscription {
    /// Detaching a subscription handle prevents it from unsubscribing on drop.
    pub fn detach(&mut self) {
        match self {
            Subscription::Notification {
                notification_handlers,
                ..
            } => *notification_handlers = None,
            Subscription::Io { io_handlers, .. } => *io_handlers = None,
        }
    }
}

impl fmt::Display for LanguageServerId {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        self.0.fmt(f)
    }
}

impl fmt::Debug for LanguageServer {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("LanguageServer")
            .field("id", &self.server_id.0)
            .field("name", &self.name)
            .finish_non_exhaustive()
    }
}

impl fmt::Debug for LanguageServerBinary {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        let mut debug = f.debug_struct("LanguageServerBinary");
        debug.field("path", &self.path);
        debug.field("arguments", &self.arguments);

        if let Some(env) = &self.env {
            let redacted_env: BTreeMap<String, String> = env
                .iter()
                .map(|(key, value)| {
                    let redacted_value = if redact::should_redact(key) {
                        "REDACTED".to_string()
                    } else {
                        value.clone()
                    };
                    (key.clone(), redacted_value)
                })
                .collect();
            debug.field("env", &Some(redacted_env));
        } else {
            debug.field("env", &self.env);
        }

        debug.finish()
    }
}

impl Drop for Subscription {
    fn drop(&mut self) {
        match self {
            Subscription::Notification {
                method,
                notification_handlers,
            } => {
                if let Some(handlers) = notification_handlers {
                    handlers.lock().remove(method);
                }
            }
            Subscription::Io { id, io_handlers } => {
                if let Some(io_handlers) = io_handlers.as_ref().and_then(|h| h.upgrade()) {
                    io_handlers.lock().remove(id);
                }
            }
        }
    }
}

/// Mock language server for use in tests.
#[cfg(any(test, feature = "test-support"))]
#[derive(Clone)]
pub struct FakeLanguageServer {
    pub binary: LanguageServerBinary,
    pub server: Arc<LanguageServer>,
    notifications_rx: channel::Receiver<(String, String)>,
}

#[cfg(any(test, feature = "test-support"))]
impl FakeLanguageServer {
    /// Construct a fake language server.
    pub fn new(
        server_id: LanguageServerId,
        binary: LanguageServerBinary,
        name: String,
        capabilities: ServerCapabilities,
        cx: &mut AsyncApp,
    ) -> (LanguageServer, FakeLanguageServer) {
        let (stdin_writer, stdin_reader) = async_pipe::pipe();
        let (stdout_writer, stdout_reader) = async_pipe::pipe();
        let (notifications_tx, notifications_rx) = channel::unbounded();

        let server_name = LanguageServerName(name.clone().into());
        let process_name = Arc::from(name.as_str());
        let root = Self::root_path();
        let workspace_folders: Arc<Mutex<BTreeSet<Uri>>> = Default::default();
        let mut server = LanguageServer::new_internal(
            server_id,
            server_name.clone(),
            stdin_writer,
            stdout_reader,
            None::<async_pipe::PipeReader>,
            Arc::new(Mutex::new(None)),
            None,
            None,
            binary.clone(),
            root,
            Some(workspace_folders.clone()),
            cx,
            |_| false,
        );
        server.process_name = process_name;
        let fake = FakeLanguageServer {
            binary: binary.clone(),
            server: Arc::new({
                let mut server = LanguageServer::new_internal(
                    server_id,
                    server_name,
                    stdout_writer,
                    stdin_reader,
                    None::<async_pipe::PipeReader>,
                    Arc::new(Mutex::new(None)),
                    None,
                    None,
                    binary,
                    Self::root_path(),
                    Some(workspace_folders),
                    cx,
                    move |msg| {
                        notifications_tx
                            .try_send((
                                msg.method.to_string(),
                                msg.params.as_ref().unwrap_or(&Value::Null).to_string(),
                            ))
                            .ok();
                        true
                    },
                );
                server.process_name = name.as_str().into();
                server
            }),
            notifications_rx,
        };
        fake.set_request_handler::<request::Initialize, _, _>({
            let capabilities = capabilities;
            move |_, _| {
                let capabilities = capabilities.clone();
                let name = name.clone();
                async move {
                    Ok(InitializeResult {
                        capabilities,
                        server_info: Some(ServerInfo {
                            name,
                            ..Default::default()
                        }),
                    })
                }
            }
        });

        fake.set_request_handler::<request::Shutdown, _, _>(|_, _| async move { Ok(()) });

        (server, fake)
    }
    #[cfg(target_os = "windows")]
    fn root_path() -> Uri {
        Uri::from_file_path("C:/").unwrap()
    }

    #[cfg(not(target_os = "windows"))]
    fn root_path() -> Uri {
        Uri::from_file_path("/").unwrap()
    }
}

#[cfg(any(test, feature = "test-support"))]
impl LanguageServer {
    pub fn full_capabilities() -> ServerCapabilities {
        ServerCapabilities {
            document_highlight_provider: Some(OneOf::Left(true)),
            code_action_provider: Some(CodeActionProviderCapability::Simple(true)),
            document_formatting_provider: Some(OneOf::Left(true)),
            document_range_formatting_provider: Some(OneOf::Left(true)),
            definition_provider: Some(OneOf::Left(true)),
            workspace_symbol_provider: Some(OneOf::Left(true)),
            implementation_provider: Some(ImplementationProviderCapability::Simple(true)),
            type_definition_provider: Some(TypeDefinitionProviderCapability::Simple(true)),
            ..ServerCapabilities::default()
        }
    }
}

#[cfg(any(test, feature = "test-support"))]
impl FakeLanguageServer {
    /// See [`LanguageServer::notify`].
    pub fn notify<T: notification::Notification>(&self, params: T::Params) {
        self.server.notify::<T>(params).ok();
    }

    /// See [`LanguageServer::request`].
    pub async fn request<T>(
        &self,
        params: T::Params,
        timeout: Duration,
    ) -> ConnectionResult<T::Result>
    where
        T: request::Request,
        T::Result: 'static + Send,
    {
        self.server.request::<T>(params, timeout).await
    }

    /// Attempts [`Self::try_receive_notification`], unwrapping if it has not received the specified type yet.
    pub async fn receive_notification<T: notification::Notification>(&mut self) -> T::Params {
        self.try_receive_notification::<T>().await.unwrap()
    }

    /// Consumes the notification channel until it finds a notification for the specified type.
    pub async fn try_receive_notification<T: notification::Notification>(
        &mut self,
    ) -> Option<T::Params> {
        loop {
            let (method, params) = self.notifications_rx.recv().await.ok()?;
            if method == T::METHOD {
                return Some(serde_json::from_str::<T::Params>(&params).unwrap());
            } else {
                log::info!("skipping message in fake language server {:?}", params);
            }
        }
    }

    /// Registers a handler for a specific kind of request. Removes any existing handler for specified request type.
    pub fn set_request_handler<T, F, Fut>(
        &self,
        mut handler: F,
    ) -> futures::channel::mpsc::UnboundedReceiver<()>
    where
        T: 'static + request::Request,
        T::Params: 'static + Send,
        F: 'static + Send + FnMut(T::Params, gpui::AsyncApp) -> Fut,
        Fut: 'static + Future<Output = Result<T::Result>>,
    {
        let (responded_tx, responded_rx) = futures::channel::mpsc::unbounded();
        self.server.remove_request_handler::<T>();
        self.server
            .on_request::<T, _, _>(move |params, cx| {
                let result = handler(params, cx.clone());
                let responded_tx = responded_tx.clone();
                let executor = cx.background_executor().clone();
                async move {
                    let _guard = gpui_util::defer({
                        let responded_tx = responded_tx.clone();
                        move || {
                            responded_tx.unbounded_send(()).ok();
                        }
                    });
                    executor.simulate_random_delay().await;
                    result.await
                }
            })
            .detach();
        responded_rx
    }

    /// Registers a handler for a specific kind of notification. Removes any existing handler for specified notification type.
    pub fn handle_notification<T, F>(
        &self,
        mut handler: F,
    ) -> futures::channel::mpsc::UnboundedReceiver<()>
    where
        T: 'static + notification::Notification,
        T::Params: 'static + Send,
        F: 'static + Send + FnMut(T::Params, gpui::AsyncApp),
    {
        let (handled_tx, handled_rx) = futures::channel::mpsc::unbounded();
        self.server.remove_notification_handler::<T>();
        self.server
            .on_notification::<T, _>(move |params, cx| {
                handler(params, cx.clone());
                handled_tx.unbounded_send(()).ok();
            })
            .detach();
        handled_rx
    }

    /// Removes any existing handler for specified notification type.
    pub fn remove_request_handler<T>(&mut self)
    where
        T: 'static + request::Request,
    {
        self.server.remove_request_handler::<T>();
    }

    /// Simulate that the server has started work and notifies about its progress with the specified token.
    pub async fn start_progress(&self, token: impl Into<String>) {
        self.start_progress_with(token, Default::default(), Default::default())
            .await
    }

    pub async fn start_progress_with(
        &self,
        token: impl Into<String>,
        progress: WorkDoneProgressBegin,
        request_timeout: Duration,
    ) {
        let token = token.into();
        self.request::<request::WorkDoneProgressCreate>(
            WorkDoneProgressCreateParams {
                token: NumberOrString::String(token.clone()),
            },
            request_timeout,
        )
        .await
        .into_response()
        .unwrap();
        self.notify::<notification::Progress>(ProgressParams {
            token: NumberOrString::String(token),
            value: ProgressParamsValue::WorkDone(WorkDoneProgress::Begin(progress)),
        });
    }

    /// Simulate that the server has completed work and notifies about that with the specified token.
    pub fn end_progress(&self, token: impl Into<String>) {
        self.notify::<notification::Progress>(ProgressParams {
            token: NumberOrString::String(token.into()),
            value: ProgressParamsValue::WorkDone(WorkDoneProgress::End(Default::default())),
        });
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use gpui::TestAppContext;
    use std::str::FromStr;

    #[ctor::ctor]
    fn init_logger() {
        zlog::init_test();
    }

    #[gpui::test]
    async fn test_fake(cx: &mut TestAppContext) {
        cx.update(|cx| {
            release_channel::init(semver::Version::new(0, 0, 0), cx);
        });
        let (server, mut fake) = FakeLanguageServer::new(
            LanguageServerId(0),
            LanguageServerBinary {
                path: "path/to/language-server".into(),
                arguments: vec![],
                env: None,
            },
            "the-lsp".to_string(),
            Default::default(),
            &mut cx.to_async(),
        );

        let (message_tx, message_rx) = channel::unbounded();
        let (diagnostics_tx, diagnostics_rx) = channel::unbounded();
        server
            .on_notification::<notification::ShowMessage, _>(move |params, _| {
                message_tx.try_send(params).unwrap()
            })
            .detach();
        server
            .on_notification::<notification::PublishDiagnostics, _>(move |params, _| {
                diagnostics_tx.try_send(params).unwrap()
            })
            .detach();

        let server = cx
            .update(|cx| {
                let params = server.default_initialize_params(false, false, cx);
                let configuration = DidChangeConfigurationParams {
                    settings: Default::default(),
                };
                server.initialize(
                    params,
                    configuration.into(),
                    DEFAULT_LSP_REQUEST_TIMEOUT,
                    cx,
                )
            })
            .await
            .unwrap();
        server
            .notify::<notification::DidOpenTextDocument>(DidOpenTextDocumentParams {
                text_document: TextDocumentItem::new(
                    Uri::from_str("file://a/b").unwrap(),
                    "rust".to_string(),
                    0,
                    "".to_string(),
                ),
            })
            .unwrap();
        assert_eq!(
            fake.receive_notification::<notification::DidOpenTextDocument>()
                .await
                .text_document
                .uri
                .as_str(),
            "file://a/b"
        );

        fake.notify::<notification::ShowMessage>(ShowMessageParams {
            typ: MessageType::ERROR,
            message: "ok".to_string(),
        });
        fake.notify::<notification::PublishDiagnostics>(PublishDiagnosticsParams {
            uri: Uri::from_str("file://b/c").unwrap(),
            version: Some(5),
            diagnostics: vec![],
        });
        assert_eq!(message_rx.recv().await.unwrap().message, "ok");
        assert_eq!(
            diagnostics_rx.recv().await.unwrap().uri.as_str(),
            "file://b/c"
        );

        fake.set_request_handler::<request::Shutdown, _, _>(|_, _| async move { Ok(()) });

        drop(server);
        cx.run_until_parked();
        fake.receive_notification::<notification::Exit>().await;
    }

    #[gpui::test]
    fn test_deserialize_string_digit_id() {
        let json = r#"{"jsonrpc":"2.0","id":"2","method":"workspace/configuration","params":{"items":[{"scopeUri":"file:///Users/mph/Devel/personal/hello-scala/","section":"metals"}]}}"#;
        let notification = serde_json::from_str::<NotificationOrRequest>(json)
            .expect("message with string id should be parsed");
        let expected_id = RequestId::Str("2".to_string());
        assert_eq!(notification.id, Some(expected_id));
    }

    #[gpui::test]
    fn test_deserialize_string_id() {
        let json = r#"{"jsonrpc":"2.0","id":"anythingAtAll","method":"workspace/configuration","params":{"items":[{"scopeUri":"file:///Users/mph/Devel/personal/hello-scala/","section":"metals"}]}}"#;
        let notification = serde_json::from_str::<NotificationOrRequest>(json)
            .expect("message with string id should be parsed");
        let expected_id = RequestId::Str("anythingAtAll".to_string());
        assert_eq!(notification.id, Some(expected_id));
    }

    #[gpui::test]
    fn test_deserialize_int_id() {
        let json = r#"{"jsonrpc":"2.0","id":2,"method":"workspace/configuration","params":{"items":[{"scopeUri":"file:///Users/mph/Devel/personal/hello-scala/","section":"metals"}]}}"#;
        let notification = serde_json::from_str::<NotificationOrRequest>(json)
            .expect("message with string id should be parsed");
        let expected_id = RequestId::Int(2);
        assert_eq!(notification.id, Some(expected_id));
    }

    #[test]
    fn test_serialize_has_no_nulls() {
        // Ensure we're not setting both result and error variants. (ticket #10595)
        let no_tag = Response::<u32> {
            jsonrpc: "",
            id: RequestId::Int(0),
            value: LspResult::Ok(None),
        };
        assert_eq!(
            serde_json::to_string(&no_tag).unwrap(),
            "{\"jsonrpc\":\"\",\"id\":0,\"result\":null}"
        );
        let no_tag = Response::<u32> {
            jsonrpc: "",
            id: RequestId::Int(0),
            value: LspResult::Error(None),
        };
        assert_eq!(
            serde_json::to_string(&no_tag).unwrap(),
            "{\"jsonrpc\":\"\",\"id\":0,\"error\":null}"
        );
    }

    #[gpui::test]
    async fn test_initialize_params_has_root_path_and_root_uri(cx: &mut TestAppContext) {
        cx.update(|cx| {
            release_channel::init(semver::Version::new(0, 0, 0), cx);
        });
        let (server, _fake) = FakeLanguageServer::new(
            LanguageServerId(0),
            LanguageServerBinary {
                path: "path/to/language-server".into(),
                arguments: vec![],
                env: None,
            },
            "test-lsp".to_string(),
            Default::default(),
            &mut cx.to_async(),
        );

        let params = cx.update(|cx| server.default_initialize_params(false, false, cx));

        #[allow(deprecated)]
        let root_uri = params.root_uri.expect("root_uri should be set");
        #[allow(deprecated)]
        let root_path = params.root_path.expect("root_path should be set");

        let expected_path = root_uri
            .to_file_path()
            .expect("root_uri should be a valid file path");
        assert_eq!(
            root_path,
            expected_path.to_string_lossy(),
            "root_path should be derived from root_uri"
        );
    }
}