Initial implementation of HTTP registry client

Cargo.toml

@@ -53,7 +53,7 @@ deno_task_shell = "0.13"
 derive_builder = "0.12"
 directories = "5"
 dunce = "1"
-fs4 = "0.7"
+fs4 = { version = "0.7", features = ["tokio"] }
 fs_extra = "1.3.0"
 futures = { version = "0.3", default-features = false, features = ["std", "async-await"] }
 gix = "0.54"

scarb/src/core/registry/client/http.rs

@@ -0,0 +1,163 @@
+use std::path::PathBuf;
+use std::sync::Arc;
+
+use anyhow::{Context, Result};
+use async_trait::async_trait;
+use fs4::tokio::AsyncFileExt;
+use futures::StreamExt;
+use reqwest::StatusCode;
+use tokio::fs::OpenOptions;
+use tokio::io;
+use tokio::io::BufWriter;
+use tokio::sync::OnceCell;
+use tracing::{debug, trace};
+
+use crate::core::registry::client::RegistryClient;
+use crate::core::registry::index::{IndexConfig, IndexRecords};
+use crate::core::{Config, Package, PackageId, PackageName, SourceId};
+use crate::flock::{FileLockGuard, Filesystem};
+
+// TODO(mkaput): Honour ETag and Last-Modified headers.
+// TODO(mkaput): Progressbar.
+// TODO(mkaput): Request timeout.
+
+/// Remote registry served by the HTTP-based registry API.
+pub struct HttpRegistryClient<'c> {
+    source_id: SourceId,
+    config: &'c Config,
+    cached_index_config: OnceCell<IndexConfig>,
+    dl_fs: Filesystem<'c>,
+}
+
+impl<'c> HttpRegistryClient<'c> {
+    pub fn new(source_id: SourceId, config: &'c Config) -> Result<Self> {
+        let dl_fs = config
+            .dirs()
+            .registry_dir()
+            .into_child("dl")
+            .into_child(source_id.ident());
+
+        Ok(Self {
+            source_id,
+            config,
+            cached_index_config: Default::default(),
+            dl_fs,
+        })
+    }
+
+    async fn index_config(&self) -> Result<&IndexConfig> {
+        // TODO(mkaput): Cache config locally, honouring ETag and Last-Modified headers.
+
+        async fn load(source_id: SourceId, config: &Config) -> Result<IndexConfig> {
+            let index_config_url = source_id
+                .url
+                .join(IndexConfig::WELL_KNOWN_PATH)
+                .expect("Registry config URL should always be valid.");
+            debug!("fetching registry config: {index_config_url}");
+
+            let index_config = config
+                .http()?
+                .get(index_config_url)
+                .send()
+                .await?
+                .error_for_status()?
+                .json::<IndexConfig>()
+                .await?;
+
+            trace!(index_config = %serde_json::to_string(&index_config).unwrap());
+
+            Ok(index_config)
+        }
+
+        self.cached_index_config
+            .get_or_try_init(|| load(self.source_id, self.config))
+            .await
+            .context("failed to fetch registry config")
+    }
+}
+
+#[async_trait]
+impl<'c> RegistryClient for HttpRegistryClient<'c> {
+    fn is_offline(&self) -> bool {
+        false
+    }
+
+    async fn get_records(&self, package: PackageName) -> Result<Option<Arc<IndexRecords>>> {
+        let index_config = self.index_config().await?;
+        let records_url = index_config.index.expand(package.into())?;
+
+        let response = self
+            .config
+            .http()?
+            .get(records_url)
+            .send()
+            .await?
+            .error_for_status();
+
+        if let Err(err) = &response {
+            if let Some(status) = err.status() {
+                if status == StatusCode::NOT_FOUND {
+                    return Ok(None);
+                }
+            }
+        }
+
+        let records = response?
+            .json()
+            .await
+            .context("failed to deserialize index records")?;
+
+        Ok(Some(Arc::new(records)))
+    }
+
+    async fn is_downloaded(&self, _package: PackageId) -> bool {
+        // TODO(mkaput): Cache downloaded packages.
+        false
+    }
+
+    async fn download(&self, package: PackageId) -> Result<PathBuf> {
+        let dl_url = self.index_config().await?.dl.expand(package.into())?;
+
+        let response = self
+            .config
+            .http()?
+            .get(dl_url)
+            .send()
+            .await?
+            .error_for_status()?;
+
+        let output_path = self.dl_fs.path_existent()?.join(package.tarball_name());
+        let output_file = OpenOptions::new()
+            .read(true)
+            .write(true)
+            .truncate(true)
+            .create(true)
+            .open(&output_path)
+            .await
+            .with_context(|| format!("failed to open: {output_path}"))?;
+
+        output_file
+            .lock_exclusive()
+            .with_context(|| format!("failed to lock file: {output_path}"))?;
+
+        let mut stream = response.bytes_stream();
+        let mut writer = BufWriter::new(output_file);
+        while let Some(chunk) = stream.next().await {
+            let chunk = chunk.context("failed to read response chunk")?;
+            io::copy_buf(&mut &*chunk, &mut writer)
+                .await
+                .context("failed to save response chunk on disk")?;
+        }
+
+        Ok(output_path.into_std_path_buf())
+    }
+
+    async fn supports_publish(&self) -> Result<bool> {
+        // TODO(mkaput): Publishing to HTTP registries is not implemented yet.
+        Ok(false)
+    }
+
+    async fn publish(&self, _package: Package, _tarball: FileLockGuard) -> Result<()> {
+        todo!("Publishing to HTTP registries is not implemented yet.")
+    }
+}

scarb/src/core/registry/client/mod.rs

@@ -8,6 +8,7 @@ use crate::core::registry::index::IndexRecords;
 use crate::core::{Package, PackageId, PackageName};
 use crate::flock::FileLockGuard;
 
+pub mod http;
 pub mod local;
 
 #[async_trait]

scarb/src/core/registry/index/config.rs

@@ -43,6 +43,10 @@ pub struct IndexConfig {
     pub index: TemplateUrl,
 }
 
+impl IndexConfig {
+    pub const WELL_KNOWN_PATH: &'_ str = "config.json";
+}
+
 #[derive(Copy, Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize)]
 #[serde(into = "u8", try_from = "u8")]
 pub struct IndexVersion;

scarb/src/sources/registry.rs

@@ -2,12 +2,13 @@ use std::collections::HashSet;
 use std::fmt;
 use std::path::PathBuf;
 
-use anyhow::{bail, ensure, Context, Result};
+use anyhow::{anyhow, bail, ensure, Context, Result};
 use async_trait::async_trait;
 use tracing::trace;
 
 use scarb_ui::components::Status;
 
+use crate::core::registry::client::http::HttpRegistryClient;
 use crate::core::registry::client::local::LocalRegistryClient;
 use crate::core::registry::client::RegistryClient;
 use crate::core::registry::index::IndexRecord;
@@ -45,14 +46,25 @@ impl<'c> RegistrySource<'c> {
 
     pub fn create_client(
         source_id: SourceId,
-        _config: &'c Config,
+        config: &'c Config,
     ) -> Result<Box<dyn RegistryClient + 'c>> {
-        if let Ok(path) = source_id.url.to_file_path() {
-            trace!("creating local registry client for: {source_id}");
-            Ok(Box::new(LocalRegistryClient::new(&path)?))
-        } else {
-            // TODO(mkaput): Implement pipelining HTTP client.
-            bail!("unsupported registry protocol: {source_id}")
+        assert!(source_id.is_registry());
+        match source_id.url.scheme() {
+            "file" => {
+                trace!("creating local registry client for: {source_id}");
+                let path = source_id
+                    .url
+                    .to_file_path()
+                    .map_err(|_| anyhow!("url is not a valid path: {}", source_id.url))?;
+                Ok(Box::new(LocalRegistryClient::new(&path)?))
+            }
+            "http" | "https" => {
+                trace!("creating http registry client for: {source_id}");
+                Ok(Box::new(HttpRegistryClient::new(source_id, config)?))
+            }
+            _ => {
+                bail!("unsupported registry protocol: {source_id}")
+            }
         }
     }
 }

scarb/tests/http_registry.rs

@@ -0,0 +1,102 @@
+use assert_fs::prelude::*;
+use assert_fs::TempDir;
+use indoc::indoc;
+use std::fs;
+use std::time::Duration;
+
+use scarb_test_support::command::Scarb;
+use scarb_test_support::project_builder::{Dep, DepBuilder, ProjectBuilder};
+use scarb_test_support::registry::http::HttpRegistry;
+
+#[test]
+fn usage() {
+    let mut registry = HttpRegistry::serve();
+    registry.publish(|t| {
+        ProjectBuilder::start()
+            .name("bar")
+            .version("1.0.0")
+            .lib_cairo(r#"fn f() -> felt252 { 0 }"#)
+            .build(t);
+    });
+
+    let t = TempDir::new().unwrap();
+    ProjectBuilder::start()
+        .name("foo")
+        .version("0.1.0")
+        .dep("bar", Dep.version("1").registry(&registry))
+        .lib_cairo(r#"fn f() -> felt252 { bar::f() }"#)
+        .build(&t);
+
+    // FIXME(mkaput): Why are verbose statuses not appearing here?
+    Scarb::quick_snapbox()
+        .arg("fetch")
+        .current_dir(&t)
+        .timeout(Duration::from_secs(10))
+        .assert()
+        .success()
+        .stdout_matches(indoc! {r#"
+        [..] Downloading bar v1.0.0 ([..])
+        "#});
+}
+
+#[test]
+fn not_found() {
+    let mut registry = HttpRegistry::serve();
+    registry.publish(|t| {
+        // Publish a package so that the directory hierarchy is created.
+        // Note, however, that we declare a dependency on baZ.
+        ProjectBuilder::start()
+            .name("bar")
+            .version("1.0.0")
+            .lib_cairo(r#"fn f() -> felt252 { 0 }"#)
+            .build(t);
+    });
+
+    let t = TempDir::new().unwrap();
+    ProjectBuilder::start()
+        .name("foo")
+        .version("0.1.0")
+        .dep("baz", Dep.version("1").registry(&registry))
+        .build(&t);
+
+    Scarb::quick_snapbox()
+        .arg("fetch")
+        .current_dir(&t)
+        .timeout(Duration::from_secs(10))
+        .assert()
+        .failure()
+        .stdout_matches(indoc! {r#"
+        error: package not found in registry: baz ^1 (registry+http://[..])
+        "#});
+}
+
+#[test]
+fn missing_config_json() {
+    let registry = HttpRegistry::serve();
+    fs::remove_file(registry.child("config.json")).unwrap();
+
+    let t = TempDir::new().unwrap();
+    ProjectBuilder::start()
+        .name("foo")
+        .version("0.1.0")
+        .dep("baz", Dep.version("1").registry(&registry))
+        .build(&t);
+
+    Scarb::quick_snapbox()
+        .arg("fetch")
+        .current_dir(&t)
+        .timeout(Duration::from_secs(10))
+        .assert()
+        .failure()
+        .stdout_matches(indoc! {r#"
+        error: failed to lookup for `baz ^1 (registry+http://[..])` in registry: registry+http://[..]
+
+        Caused by:
+            0: failed to fetch registry config
+            1: HTTP status client error (404 Not Found) for url (http://[..]/config.json)
+        "#});
+}
+
+// TODO(mkaput): Test errors properly when package is in index, but tarball is missing.
+// TODO(mkaput): Test interdependencies.
+// TODO(mkaput): Test offline mode.