command_stream/command_packages/
marketplace.rs

1//! Registry-backed package discovery, with an injectable asynchronous transport.
2use crate::CommandBuilder;
3use serde::{Deserialize, Serialize};
4use serde_json::Value;
5use std::future::Future;
6use tokio::sync::watch;
7
8#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)]
9pub struct PackageMetadata {
10    pub name: String,
11    #[serde(default)]
12    pub version: String,
13    #[serde(default)]
14    pub description: String,
15    #[serde(flatten)]
16    pub extra: serde_json::Map<String, Value>,
17}
18
19#[derive(Clone)]
20pub struct SearchOptions {
21    pub registry: String,
22    pub size: usize,
23    /// Sending true aborts an outstanding request; dropping the sender does not.
24    pub cancel: Option<watch::Receiver<bool>>,
25}
26
27impl Default for SearchOptions {
28    fn default() -> Self {
29        Self {
30            registry: "https://registry.npmjs.org".into(),
31            size: 20,
32            cancel: None,
33        }
34    }
35}
36
37pub struct Marketplace;
38
39impl Marketplace {
40    /// Query npm's real registry. Requires curl on PATH; never invokes a shell.
41    /// The request future owns its child process, so cancellation stops curl.
42    pub async fn search(
43        query: &str,
44        options: SearchOptions,
45    ) -> Result<Vec<PackageMetadata>, String> {
46        Self::search_with(query, options, |url| async move {
47            let result = CommandBuilder::new("curl")
48                .args([
49                    "--fail-with-body",
50                    "--silent",
51                    "--show-error",
52                    "--url",
53                    &url,
54                ])
55                .run()
56                .await
57                .map_err(|error| error.to_string())?;
58            if result.code != 0 {
59                return Err(format!("Marketplace search failed: {}", result.stderr));
60            }
61            serde_json::from_str(&result.stdout).map_err(|error| error.to_string())
62        })
63        .await
64    }
65
66    /// Supply an HTTP transport returning decoded registry JSON for tests or
67    /// applications using their own HTTP client. Transport errors propagate.
68    pub async fn search_with<F, Fut>(
69        query: &str,
70        mut options: SearchOptions,
71        request: F,
72    ) -> Result<Vec<PackageMetadata>, String>
73    where
74        F: FnOnce(String) -> Fut,
75        Fut: Future<Output = Result<Value, String>>,
76    {
77        let registry = options.registry.trim_end_matches('/');
78        let authority = registry
79            .strip_prefix("https://")
80            .or_else(|| registry.strip_prefix("http://"))
81            .ok_or("Marketplace registry must use HTTP or HTTPS")?;
82        if authority.is_empty() || authority.contains(['/', '?', '#', '@']) {
83            return Err("Marketplace registry must be an HTTP origin".into());
84        }
85        let text = encode(&format!("keywords:command-stream-command {query}"));
86        let url = format!(
87            "{registry}/-/v1/search?text={text}&size={}",
88            options.size.clamp(1, 250)
89        );
90        let response = if let Some(cancel) = &mut options.cancel {
91            if *cancel.borrow() {
92                return Err("Marketplace search cancelled".into());
93            }
94            tokio::select! {
95                response = request(url) => response?,
96                () = cancelled(cancel) => return Err("Marketplace search cancelled".into()),
97            }
98        } else {
99            request(url).await?
100        };
101        #[derive(Deserialize)]
102        struct Entry {
103            package: PackageMetadata,
104        }
105        #[derive(Deserialize)]
106        struct Response {
107            objects: Vec<Entry>,
108        }
109        let response: Response =
110            serde_json::from_value(response).map_err(|error| error.to_string())?;
111        Ok(response
112            .objects
113            .into_iter()
114            .map(|entry| entry.package)
115            .collect())
116    }
117}
118
119async fn cancelled(receiver: &mut watch::Receiver<bool>) {
120    while receiver.changed().await.is_ok() {
121        if *receiver.borrow() {
122            return;
123        }
124    }
125    std::future::pending::<()>().await;
126}
127
128fn encode(value: &str) -> String {
129    use std::fmt::Write;
130    let mut encoded = String::new();
131    for byte in value.bytes() {
132        if byte.is_ascii_alphanumeric() || b"-._~".contains(&byte) {
133            encoded.push(char::from(byte));
134        } else {
135            write!(encoded, "%{byte:02X}").expect("writing to a string");
136        }
137    }
138    encoded
139}
command_stream/command_packages/marketplace.rs

command_stream/command_packages/
marketplace.rs
