pub mod extension_settings;pub mod headless_host;pub mod wasm_host; #[cfg(test)]mod extension_store_test; use anyhow::{anyhow, bail, Context as _, Result};use async_compression::futures::bufread::GzipDecoder;use async_tar::Archive;use client::{proto, telemetry::Telemetry, Client, ExtensionMetadata, GetExtensionsResponse};use collections::{btree_map, BTreeMap, HashMap, HashSet};use extension::extension_builder::{CompileExtensionOptions, ExtensionBuilder};pub use extension::ExtensionManifest;use extension::{    ExtensionContextServerProxy, ExtensionGrammarProxy, ExtensionHostProxy,    ExtensionIndexedDocsProviderProxy, ExtensionLanguageProxy, ExtensionLanguageServerProxy,    ExtensionSlashCommandProxy, ExtensionSnippetProxy, ExtensionThemeProxy,};use fs::{Fs, RemoveOptions};use futures::{    channel::{        mpsc::{unbounded, UnboundedSender},        oneshot,    },    io::BufReader,    select_biased, AsyncReadExt as _, Future, FutureExt as _, StreamExt as _,};use gpui::{    actions, AppContext, AsyncAppContext, Context, EventEmitter, Global, Model, ModelContext, Task,    WeakModel,};use http_client::{AsyncBody, HttpClient, HttpClientWithUrl};use language::{    LanguageConfig, LanguageMatcher, LanguageName, LanguageQueries, LoadedLanguage, Rope,    QUERY_FILENAME_PREFIXES,};use node_runtime::NodeRuntime;use project::ContextProviderWithTasks;use release_channel::ReleaseChannel;use remote::SshRemoteClient;use semantic_version::SemanticVersion;use serde::{Deserialize, Serialize};use settings::Settings;use std::ops::RangeInclusive;use std::str::FromStr;use std::{    cmp::Ordering,    path::{self, Path, PathBuf},    sync::Arc,    time::{Duration, Instant},};use url::Url;use util::ResultExt;use wasm_host::{    wit::{is_supported_wasm_api_version, wasm_api_version_range},    WasmExtension, WasmHost,}; pub use extension::{    ExtensionLibraryKind, GrammarManifestEntry, OldExtensionManifest, SchemaVersion,};pub use extension_settings::ExtensionSettings; pub const RELOAD_DEBOUNCE_DURATION: Duration = Duration::from_millis(200);const FS_WATCH_LATENCY: Duration = Duration::from_millis(100); /// The current extension [`SchemaVersion`] supported by Zed.const CURRENT_SCHEMA_VERSION: SchemaVersion = SchemaVersion(1); /// Returns the [`SchemaVersion`] range that is compatible with this version of Zed.pub fn schema_version_range() -> RangeInclusive<SchemaVersion> {    SchemaVersion::ZERO..=CURRENT_SCHEMA_VERSION} /// Returns whether the given extension version is compatible with this version of Zed.pub fn is_version_compatible(    release_channel: ReleaseChannel,    extension_version: &ExtensionMetadata,) -> bool {    let schema_version = extension_version.manifest.schema_version.unwrap_or(0);    if CURRENT_SCHEMA_VERSION.0 < schema_version {        return false;    }     if let Some(wasm_api_version) = extension_version        .manifest        .wasm_api_version        .as_ref()        .and_then(|wasm_api_version| SemanticVersion::from_str(wasm_api_version).ok())    {        if !is_supported_wasm_api_version(release_channel, wasm_api_version) {            return false;        }    }     true} pub struct ExtensionStore {    pub proxy: Arc<ExtensionHostProxy>,    pub builder: Arc<ExtensionBuilder>,    pub extension_index: ExtensionIndex,    pub fs: Arc<dyn Fs>,    pub http_client: Arc<HttpClientWithUrl>,    pub telemetry: Option<Arc<Telemetry>>,    pub reload_tx: UnboundedSender<Option<Arc<str>>>,    pub reload_complete_senders: Vec<oneshot::Sender<()>>,    pub installed_dir: PathBuf,    pub outstanding_operations: BTreeMap<Arc<str>, ExtensionOperation>,    pub index_path: PathBuf,    pub modified_extensions: HashSet<Arc<str>>,    pub wasm_host: Arc<WasmHost>,    pub wasm_extensions: Vec<(Arc<ExtensionManifest>, WasmExtension)>,    pub tasks: Vec<Task<()>>,    pub ssh_clients: HashMap<String, WeakModel<SshRemoteClient>>,    pub ssh_registered_tx: UnboundedSender<()>,} #[derive(Clone, Copy)]pub enum ExtensionOperation {    Upgrade,    Install,    Remove,} #[derive(Clone)]pub enum Event {    ExtensionsUpdated,    StartedReloading,    ExtensionInstalled(Arc<str>),    ExtensionFailedToLoad(Arc<str>),} impl EventEmitter<Event> for ExtensionStore {} struct GlobalExtensionStore(Model<ExtensionStore>); impl Global for GlobalExtensionStore {} #[derive(Debug, Deserialize, Serialize, Default, PartialEq, Eq)]pub struct ExtensionIndex {    pub extensions: BTreeMap<Arc<str>, ExtensionIndexEntry>,    pub themes: BTreeMap<Arc<str>, ExtensionIndexThemeEntry>,    pub languages: BTreeMap<LanguageName, ExtensionIndexLanguageEntry>,} #[derive(Clone, PartialEq, Eq, Debug, Deserialize, Serialize)]pub struct ExtensionIndexEntry {    pub manifest: Arc<ExtensionManifest>,    pub dev: bool,} #[derive(Clone, PartialEq, Eq, PartialOrd, Ord, Debug, Deserialize, Serialize)]pub struct ExtensionIndexThemeEntry {    pub extension: Arc<str>,    pub path: PathBuf,} #[derive(Clone, PartialEq, Eq, PartialOrd, Ord, Debug, Deserialize, Serialize)]pub struct ExtensionIndexLanguageEntry {    pub extension: Arc<str>,    pub path: PathBuf,    pub matcher: LanguageMatcher,    pub hidden: bool,    pub grammar: Option<Arc<str>>,} actions!(zed, [ReloadExtensions]); pub fn init(    extension_host_proxy: Arc<ExtensionHostProxy>,    fs: Arc<dyn Fs>,    client: Arc<Client>,    node_runtime: NodeRuntime,    cx: &mut AppContext,) {    ExtensionSettings::register(cx);     let store = cx.new_model(move |cx| {        ExtensionStore::new(            paths::extensions_dir().clone(),            None,            extension_host_proxy,            fs,            client.http_client().clone(),            client.http_client().clone(),            Some(client.telemetry().clone()),            node_runtime,            cx,        )    });     cx.on_action(|_: &ReloadExtensions, cx| {        let store = cx.global::<GlobalExtensionStore>().0.clone();        store.update(cx, |store, cx| drop(store.reload(None, cx)));    });     cx.set_global(GlobalExtensionStore(store));} impl ExtensionStore {    pub fn try_global(cx: &AppContext) -> Option<Model<Self>> {        cx.try_global::<GlobalExtensionStore>()            .map(|store| store.0.clone())    }     pub fn global(cx: &AppContext) -> Model<Self> {        cx.global::<GlobalExtensionStore>().0.clone()    }     #[allow(clippy::too_many_arguments)]    pub fn new(        extensions_dir: PathBuf,        build_dir: Option<PathBuf>,        extension_host_proxy: Arc<ExtensionHostProxy>,        fs: Arc<dyn Fs>,        http_client: Arc<HttpClientWithUrl>,        builder_client: Arc<dyn HttpClient>,        telemetry: Option<Arc<Telemetry>>,        node_runtime: NodeRuntime,        cx: &mut ModelContext<Self>,    ) -> Self {        let work_dir = extensions_dir.join("work");        let build_dir = build_dir.unwrap_or_else(|| extensions_dir.join("build"));        let installed_dir = extensions_dir.join("installed");        let index_path = extensions_dir.join("index.json");         let (reload_tx, mut reload_rx) = unbounded();        let (connection_registered_tx, mut connection_registered_rx) = unbounded();        let mut this = Self {            proxy: extension_host_proxy.clone(),            extension_index: Default::default(),            installed_dir,            index_path,            builder: Arc::new(ExtensionBuilder::new(builder_client, build_dir)),            outstanding_operations: Default::default(),            modified_extensions: Default::default(),            reload_complete_senders: Vec::new(),            wasm_host: WasmHost::new(                fs.clone(),                http_client.clone(),                node_runtime,                extension_host_proxy,                work_dir,                cx,            ),            wasm_extensions: Vec::new(),            fs,            http_client,            telemetry,            reload_tx,            tasks: Vec::new(),             ssh_clients: HashMap::default(),            ssh_registered_tx: connection_registered_tx,        };         // The extensions store maintains an index file, which contains a complete        // list of the installed extensions and the resources that they provide.        // This index is loaded synchronously on startup.        let (index_content, index_metadata, extensions_metadata) =            cx.background_executor().block(async {                futures::join!(                    this.fs.load(&this.index_path),                    this.fs.metadata(&this.index_path),                    this.fs.metadata(&this.installed_dir),                )            });         // Normally, there is no need to rebuild the index. But if the index file        // is invalid or is out-of-date according to the filesystem mtimes, then        // it must be asynchronously rebuilt.        let mut extension_index = ExtensionIndex::default();        let mut extension_index_needs_rebuild = true;        if let Ok(index_content) = index_content {            if let Some(index) = serde_json::from_str(&index_content).log_err() {                extension_index = index;                if let (Ok(Some(index_metadata)), Ok(Some(extensions_metadata))) =                    (index_metadata, extensions_metadata)                {                    if index_metadata                        .mtime                        .bad_is_greater_than(extensions_metadata.mtime)                    {                        extension_index_needs_rebuild = false;                    }                }            }        }         // Immediately load all of the extensions in the initial manifest. If the        // index needs to be rebuild, then enqueue        let load_initial_extensions = this.extensions_updated(extension_index, cx);        let mut reload_future = None;        if extension_index_needs_rebuild {            reload_future = Some(this.reload(None, cx));        }         cx.spawn(|this, mut cx| async move {            if let Some(future) = reload_future {                future.await;            }            this.update(&mut cx, |this, cx| this.auto_install_extensions(cx))                .ok();            this.update(&mut cx, |this, cx| this.check_for_updates(cx))                .ok();        })        .detach();         // Perform all extension loading in a single task to ensure that we        // never attempt to simultaneously load/unload extensions from multiple        // parallel tasks.        this.tasks.push(cx.spawn(|this, mut cx| {            async move {                load_initial_extensions.await;                 let mut index_changed = false;                let mut debounce_timer = cx                    .background_executor()                    .spawn(futures::future::pending())                    .fuse();                loop {                    select_biased! {                        _ = debounce_timer => {                            if index_changed {                                let index = this                                    .update(&mut cx, |this, cx| this.rebuild_extension_index(cx))?                                    .await;                                this.update(&mut cx, |this, cx| this.extensions_updated(index, cx))?                                    .await;                                index_changed = false;                            }                             Self::update_ssh_clients(&this, &mut cx).await?;                        }                        _ = connection_registered_rx.next() => {                            debounce_timer = cx                                .background_executor()                                .timer(RELOAD_DEBOUNCE_DURATION)                                .fuse();                        }                        extension_id = reload_rx.next() => {                            let Some(extension_id) = extension_id else { break; };                            this.update(&mut cx, |this, _| {                                this.modified_extensions.extend(extension_id);                            })?;                            index_changed = true;                            debounce_timer = cx                                .background_executor()                                .timer(RELOAD_DEBOUNCE_DURATION)                                .fuse();                        }                    }                }                 anyhow::Ok(())            }            .map(drop)        }));         // Watch the installed extensions directory for changes. Whenever changes are        // detected, rebuild the extension index, and load/unload any extensions that        // have been added, removed, or modified.        this.tasks.push(cx.background_executor().spawn({            let fs = this.fs.clone();            let reload_tx = this.reload_tx.clone();            let installed_dir = this.installed_dir.clone();            async move {                let (mut paths, _) = fs.watch(&installed_dir, FS_WATCH_LATENCY).await;                while let Some(events) = paths.next().await {                    for event in events {                        let Ok(event_path) = event.path.strip_prefix(&installed_dir) else {                            continue;                        };                         if let Some(path::Component::Normal(extension_dir_name)) =                            event_path.components().next()                        {                            if let Some(extension_id) = extension_dir_name.to_str() {                                reload_tx.unbounded_send(Some(extension_id.into())).ok();                            }                        }                    }                }            }        }));         this    }     pub fn reload(        &mut self,        modified_extension: Option<Arc<str>>,        cx: &mut ModelContext<Self>,    ) -> impl Future<Output = ()> {        let (tx, rx) = oneshot::channel();        self.reload_complete_senders.push(tx);        self.reload_tx            .unbounded_send(modified_extension)            .expect("reload task exited");        cx.emit(Event::StartedReloading);         async move {            rx.await.ok();        }    }     fn extensions_dir(&self) -> PathBuf {        self.installed_dir.clone()    }     pub fn outstanding_operations(&self) -> &BTreeMap<Arc<str>, ExtensionOperation> {        &self.outstanding_operations    }     pub fn installed_extensions(&self) -> &BTreeMap<Arc<str>, ExtensionIndexEntry> {        &self.extension_index.extensions    }     pub fn dev_extensions(&self) -> impl Iterator<Item = &Arc<ExtensionManifest>> {        self.extension_index            .extensions            .values()            .filter_map(|extension| extension.dev.then_some(&extension.manifest))    }     /// Returns the names of themes provided by extensions.    pub fn extension_themes<'a>(        &'a self,        extension_id: &'a str,    ) -> impl Iterator<Item = &'a Arc<str>> {        self.extension_index            .themes            .iter()            .filter_map(|(name, theme)| theme.extension.as_ref().eq(extension_id).then_some(name))    }     pub fn fetch_extensions(        &self,        search: Option<&str>,        cx: &mut ModelContext<Self>,    ) -> Task<Result<Vec<ExtensionMetadata>>> {        let version = CURRENT_SCHEMA_VERSION.to_string();        let mut query = vec![("max_schema_version", version.as_str())];        if let Some(search) = search {            query.push(("filter", search));        }         self.fetch_extensions_from_api("/extensions", &query, cx)    }     pub fn fetch_extensions_with_update_available(        &mut self,        cx: &mut ModelContext<Self>,    ) -> Task<Result<Vec<ExtensionMetadata>>> {        let schema_versions = schema_version_range();        let wasm_api_versions = wasm_api_version_range(ReleaseChannel::global(cx));        let extension_settings = ExtensionSettings::get_global(cx);        let extension_ids = self            .extension_index            .extensions            .iter()            .filter(|(id, entry)| !entry.dev && extension_settings.should_auto_update(id))            .map(|(id, _)| id.as_ref())            .collect::<Vec<_>>()            .join(",");        let task = self.fetch_extensions_from_api(            "/extensions/updates",            &[                ("min_schema_version", &schema_versions.start().to_string()),                ("max_schema_version", &schema_versions.end().to_string()),                (                    "min_wasm_api_version",                    &wasm_api_versions.start().to_string(),                ),                ("max_wasm_api_version", &wasm_api_versions.end().to_string()),                ("ids", &extension_ids),            ],            cx,        );        cx.spawn(move |this, mut cx| async move {            let extensions = task.await?;            this.update(&mut cx, |this, _cx| {                extensions                    .into_iter()                    .filter(|extension| {                        this.extension_index.extensions.get(&extension.id).map_or(                            true,                            |installed_extension| {                                installed_extension.manifest.version != extension.manifest.version                            },                        )                    })                    .collect()            })        })    }     pub fn fetch_extension_versions(

Lines 1–500 of 1579. Load more