ZIm/crates/extension_host/src/headless_host.rs

use std::{path::PathBuf, sync::Arc};

use anyhow::{Context as _, Result};
use client::{
    TypedEnvelope,
    proto::{self, FromProto},
};
use collections::{HashMap, HashSet};
use extension::{
    Extension, ExtensionDebugAdapterProviderProxy, ExtensionHostProxy, ExtensionLanguageProxy,
    ExtensionLanguageServerProxy, ExtensionManifest,
};
use fs::{Fs, RemoveOptions, RenameOptions};
use futures::future::join_all;
use gpui::{App, AppContext as _, AsyncApp, Context, Entity, Task, WeakEntity};
use http_client::HttpClient;
use language::{LanguageConfig, LanguageName, LanguageQueries, LoadedLanguage};
use lsp::LanguageServerName;
use node_runtime::NodeRuntime;

use crate::wasm_host::{WasmExtension, WasmHost};

#[derive(Clone, Debug)]
pub struct ExtensionVersion {
    pub id: String,
    pub version: String,
    pub dev: bool,
}

pub struct HeadlessExtensionStore {
    pub fs: Arc<dyn Fs>,
    pub extension_dir: PathBuf,
    pub proxy: Arc<ExtensionHostProxy>,
    pub wasm_host: Arc<WasmHost>,
    pub loaded_extensions: HashMap<Arc<str>, Arc<str>>,
    pub loaded_languages: HashMap<Arc<str>, Vec<LanguageName>>,
    pub loaded_language_servers: HashMap<Arc<str>, Vec<(LanguageServerName, LanguageName)>>,
}

impl HeadlessExtensionStore {
    pub fn new(
        fs: Arc<dyn Fs>,
        http_client: Arc<dyn HttpClient>,
        extension_dir: PathBuf,
        extension_host_proxy: Arc<ExtensionHostProxy>,
        node_runtime: NodeRuntime,
        cx: &mut App,
    ) -> Entity<Self> {
        cx.new(|cx| Self {
            fs: fs.clone(),
            wasm_host: WasmHost::new(
                fs.clone(),
                http_client.clone(),
                node_runtime,
                extension_host_proxy.clone(),
                extension_dir.join("work"),
                cx,
            ),
            extension_dir,
            proxy: extension_host_proxy,
            loaded_extensions: Default::default(),
            loaded_languages: Default::default(),
            loaded_language_servers: Default::default(),
        })
    }

    pub fn sync_extensions(
        &mut self,
        extensions: Vec<ExtensionVersion>,
        cx: &Context<Self>,
    ) -> Task<Result<Vec<ExtensionVersion>>> {
        let on_client = HashSet::from_iter(extensions.iter().map(|e| e.id.as_str()));
        let to_remove: Vec<Arc<str>> = self
            .loaded_extensions
            .keys()
            .filter(|id| !on_client.contains(id.as_ref()))
            .cloned()
            .collect();
        let to_load: Vec<ExtensionVersion> = extensions
            .into_iter()
            .filter(|e| {
                if e.dev {
                    return true;
                }
                self.loaded_extensions
                    .get(e.id.as_str())
                    .is_none_or(|loaded| loaded.as_ref() != e.version.as_str())
            })
            .collect();

        cx.spawn(async move |this, cx| {
            let mut missing = Vec::new();

            for extension_id in to_remove {
                log::info!("removing extension: {}", extension_id);
                this.update(cx, |this, cx| this.uninstall_extension(&extension_id, cx))?
                    .await?;
            }

            for extension in to_load {
                if let Err(e) = Self::load_extension(this.clone(), extension.clone(), cx).await {
                    log::info!("failed to load extension: {}, {:?}", extension.id, e);
                    missing.push(extension)
                } else if extension.dev {
                    missing.push(extension)
                }
            }

            Ok(missing)
        })
    }

    pub async fn load_extension(
        this: WeakEntity<Self>,
        extension: ExtensionVersion,
        cx: &mut AsyncApp,
    ) -> Result<()> {
        let (fs, wasm_host, extension_dir) = this.update(cx, |this, _cx| {
            this.loaded_extensions.insert(
                extension.id.clone().into(),
                extension.version.clone().into(),
            );
            (
                this.fs.clone(),
                this.wasm_host.clone(),
                this.extension_dir.join(&extension.id),
            )
        })?;

        let manifest = Arc::new(ExtensionManifest::load(fs.clone(), &extension_dir).await?);

        debug_assert!(!manifest.languages.is_empty() || manifest.allow_remote_load());

        if manifest.version.as_ref() != extension.version.as_str() {
            anyhow::bail!(
                "mismatched versions: ({}) != ({})",
                manifest.version,
                extension.version
            )
        }

        for language_path in &manifest.languages {
            let language_path = extension_dir.join(language_path);
            let config = fs.load(&language_path.join("config.toml")).await?;
            let mut config = ::toml::from_str::<LanguageConfig>(&config)?;

            this.update(cx, |this, _cx| {
                this.loaded_languages
                    .entry(manifest.id.clone())
                    .or_default()
                    .push(config.name.clone());

                config.grammar = None;

                this.proxy.register_language(
                    config.name.clone(),
                    None,
                    config.matcher.clone(),
                    config.hidden,
                    Arc::new(move || {
                        Ok(LoadedLanguage {
                            config: config.clone(),
                            queries: LanguageQueries::default(),
                            context_provider: None,
                            toolchain_provider: None,
                            manifest_name: None,
                        })
                    }),
                );
            })?;
        }

        if !manifest.allow_remote_load() {
            return Ok(());
        }

        let wasm_extension: Arc<dyn Extension> =
            Arc::new(WasmExtension::load(&extension_dir, &manifest, wasm_host.clone(), &cx).await?);

        for (language_server_id, language_server_config) in &manifest.language_servers {
            for language in language_server_config.languages() {
                this.update(cx, |this, _cx| {
                    this.loaded_language_servers
                        .entry(manifest.id.clone())
                        .or_default()
                        .push((language_server_id.clone(), language.clone()));
                    this.proxy.register_language_server(
                        wasm_extension.clone(),
                        language_server_id.clone(),
                        language.clone(),
                    );
                })?;
            }
            log::info!("Loaded language server: {}", language_server_id);
        }

        for (debug_adapter, meta) in &manifest.debug_adapters {
            let schema_path = extension::build_debug_adapter_schema_path(debug_adapter, meta);

            this.update(cx, |this, _cx| {
                this.proxy.register_debug_adapter(
                    wasm_extension.clone(),
                    debug_adapter.clone(),
                    &extension_dir.join(schema_path),
                );
            })?;
            log::info!("Loaded debug adapter: {}", debug_adapter);
        }

        for debug_locator in manifest.debug_locators.keys() {
            this.update(cx, |this, _cx| {
                this.proxy
                    .register_debug_locator(wasm_extension.clone(), debug_locator.clone());
            })?;
            log::info!("Loaded debug locator: {}", debug_locator);
        }

        Ok(())
    }

    fn uninstall_extension(
        &mut self,
        extension_id: &Arc<str>,
        cx: &mut Context<Self>,
    ) -> Task<Result<()>> {
        self.loaded_extensions.remove(extension_id);

        let languages_to_remove = self
            .loaded_languages
            .remove(extension_id)
            .unwrap_or_default();
        self.proxy.remove_languages(&languages_to_remove, &[]);

        let servers_to_remove = self
            .loaded_language_servers
            .remove(extension_id)
            .unwrap_or_default();
        let proxy = self.proxy.clone();
        let path = self.extension_dir.join(&extension_id.to_string());
        let fs = self.fs.clone();
        cx.spawn(async move |_, cx| {
            let mut removal_tasks = Vec::with_capacity(servers_to_remove.len());
            cx.update(|cx| {
                for (language_server_name, language) in servers_to_remove {
                    removal_tasks.push(proxy.remove_language_server(
                        &language,
                        &language_server_name,
                        cx,
                    ));
                }
            })
            .ok();
            let _ = join_all(removal_tasks).await;

            fs.remove_dir(
                &path,
                RemoveOptions {
                    recursive: true,
                    ignore_if_not_exists: true,
                },
            )
            .await
            .with_context(|| format!("Removing directory {path:?}"))
        })
    }

    pub fn install_extension(
        &mut self,
        extension: ExtensionVersion,
        tmp_path: PathBuf,
        cx: &mut Context<Self>,
    ) -> Task<Result<()>> {
        let path = self.extension_dir.join(&extension.id);
        let fs = self.fs.clone();

        cx.spawn(async move |this, cx| {
            if fs.is_dir(&path).await {
                this.update(cx, |this, cx| {
                    this.uninstall_extension(&extension.id.clone().into(), cx)
                })?
                .await?;
            }

            fs.rename(&tmp_path, &path, RenameOptions::default())
                .await?;

            Self::load_extension(this, extension, cx).await
        })
    }

    pub async fn handle_sync_extensions(
        extension_store: Entity<HeadlessExtensionStore>,
        envelope: TypedEnvelope<proto::SyncExtensions>,
        mut cx: AsyncApp,
    ) -> Result<proto::SyncExtensionsResponse> {
        let requested_extensions =
            envelope
                .payload
                .extensions
                .into_iter()
                .map(|p| ExtensionVersion {
                    id: p.id,
                    version: p.version,
                    dev: p.dev,
                });
        let missing_extensions = extension_store
            .update(&mut cx, |extension_store, cx| {
                extension_store.sync_extensions(requested_extensions.collect(), cx)
            })?
            .await?;

        Ok(proto::SyncExtensionsResponse {
            missing_extensions: missing_extensions
                .into_iter()
                .map(|e| proto::Extension {
                    id: e.id,
                    version: e.version,
                    dev: e.dev,
                })
                .collect(),
            tmp_dir: paths::remote_extensions_uploads_dir()
                .to_string_lossy()
                .to_string(),
        })
    }

    pub async fn handle_install_extension(
        extensions: Entity<HeadlessExtensionStore>,
        envelope: TypedEnvelope<proto::InstallExtension>,
        mut cx: AsyncApp,
    ) -> Result<proto::Ack> {
        let extension = envelope
            .payload
            .extension
            .context("Invalid InstallExtension request")?;

        extensions
            .update(&mut cx, |extensions, cx| {
                extensions.install_extension(
                    ExtensionVersion {
                        id: extension.id,
                        version: extension.version,
                        dev: extension.dev,
                    },
                    PathBuf::from_proto(envelope.payload.tmp_dir),
                    cx,
                )
            })?
            .await?;

        Ok(proto::Ack {})
    }
}