api: Update 'missing_content' to give upload endpoint(s)

lann · lann · commit 79f9a3536fc4 · 2023-07-14T10:55:56.000-04:00
diff --git a/crates/api/src/v1/package.rs b/crates/api/src/v1/package.rs
@@ -12,7 +12,7 @@ use warg_protocol::{
 
 /// Represents the supported kinds of content sources.
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
-#[serde(tag = "type", rename_all = "camelCase")]
+#[serde(tag = "type", rename_all = "kebab-case")]
 pub enum ContentSource {
     /// The content is located at an anonymous HTTP URL.
     Http {
@@ -21,6 +21,26 @@ pub enum ContentSource {
     },
 }
 
+/// Represents the supported kinds of content upload endpoints.
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(tag = "type", rename_all = "kebab-case")]
+pub enum UploadEndpoint {
+    /// Content may be uploaded via HTTP POST to the given URL.
+    /// If the endpoint responds with "201 Created" and a Location header, that
+    /// header's value will be the content source.
+    HttpPost {
+        /// The URL to POST content to.
+        url: String,
+    },
+}
+
+/// Information about missing content.
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct MissingContent {
+    /// Upload endpoint(s) that may be used to provide missing content.
+    pub upload: Vec<UploadEndpoint>,
+}
+
 /// Represents a request to publish a record to a package log.
 #[derive(Serialize, Deserialize)]
 #[serde(rename = "camelCase")]
@@ -55,13 +75,13 @@ impl PackageRecord {
         }
     }
 
-    /// Gets the missing content digests of the record.
-    pub fn missing_content(&self) -> &[AnyHash] {
+    /// Gets the missing content of the record.
+    pub fn missing_content(&self) -> impl Iterator<Item = (&AnyHash, &MissingContent)> {
         match &self.state {
             PackageRecordState::Sourcing {
                 missing_content, ..
-            } => missing_content,
-            _ => &[],
+            } => Box::new(missing_content.iter()) as Box<dyn Iterator<Item = _>>,
+            _ => Box::new(std::iter::empty()) as Box<dyn Iterator<Item = _>>,
         }
     }
 }
@@ -78,7 +98,7 @@ pub enum PackageRecordState {
     /// The package record needs content sources.
     Sourcing {
         /// The digests of the missing content.
-        missing_content: Vec<AnyHash>,
+        missing_content: HashMap<AnyHash, MissingContent>,
     },
     /// The package record is processing.
     Processing,
diff --git a/crates/client/src/api.rs b/crates/client/src/api.rs
@@ -349,15 +349,9 @@ impl Client {
     /// Uploads package content to the registry.
     pub async fn upload_content(
         &self,
-        log_id: &LogId,
-        record_id: &RecordId,
-        digest: &AnyHash,
+        url: &str,
         content: impl Into<Body>,
     ) -> Result<String, ClientError> {
-        let url = self
-            .url
-            .join(&paths::package_record_content(log_id, record_id, digest))
-            .unwrap();
         tracing::debug!("uploading content to `{url}`");
 
         let response = self.client.post(url).body(content).send().await?;
diff --git a/crates/client/src/lib.rs b/crates/client/src/lib.rs
@@ -13,7 +13,10 @@ use storage::{
 use thiserror::Error;
 use warg_api::v1::{
     fetch::{FetchError, FetchLogsRequest, FetchLogsResponse},
-    package::{PackageError, PackageRecord, PackageRecordState, PublishRecordRequest},
+    package::{
+        MissingContent, PackageError, PackageRecord, PackageRecordState, PublishRecordRequest,
+        UploadEndpoint,
+    },
     proof::{ConsistencyRequest, InclusionRequest},
 };
 use warg_crypto::{
@@ -154,34 +157,33 @@ impl<R: RegistryStorage, C: ContentStorage> Client<R, C> {
                 })
             })?;
 
-        let missing = record.missing_content();
-        if !missing.is_empty() {
-            // Upload the missing content
-            // TODO: parallelize this
-            for digest in record.missing_content() {
-                self.api
-                    .upload_content(
-                        &log_id,
-                        &record.id,
-                        digest,
-                        Body::wrap_stream(self.content.load_content(digest).await?.ok_or_else(
-                            || ClientError::ContentNotFound {
-                                digest: digest.clone(),
-                            },
-                        )?),
-                    )
-                    .await
-                    .map_err(|e| match e {
-                        api::ClientError::Package(PackageError::Rejection(reason)) => {
-                            ClientError::PublishRejected {
-                                id: package.id.clone(),
-                                record_id: record.id.clone(),
-                                reason,
-                            }
+        // TODO: parallelize this
+        for (digest, MissingContent { upload }) in record.missing_content() {
+            // Upload the missing content, if the registry supports it
+            let Some(UploadEndpoint::HttpPost {url}) = upload.first() else {
+                continue;
+            };
+
+            self.api
+                .upload_content(
+                    url,
+                    Body::wrap_stream(self.content.load_content(digest).await?.ok_or_else(
+                        || ClientError::ContentNotFound {
+                            digest: digest.clone(),
+                        },
+                    )?),
+                )
+                .await
+                .map_err(|e| match e {
+                    api::ClientError::Package(PackageError::Rejection(reason)) => {
+                        ClientError::PublishRejected {
+                            id: package.id.clone(),
+                            record_id: record.id.clone(),
+                            reason,
                         }
-                        _ => e.into(),
-                    })?;
-            }
+                    }
+                    _ => e.into(),
+                })?;
         }
 
         Ok(record.id)
diff --git a/crates/server/src/api/v1/package.rs b/crates/server/src/api/v1/package.rs
@@ -16,13 +16,14 @@ use axum::{
     Router,
 };
 use futures::StreamExt;
-use std::path::PathBuf;
 use std::sync::Arc;
+use std::{collections::HashMap, path::PathBuf};
 use tempfile::NamedTempFile;
 use tokio::io::AsyncWriteExt;
 use url::Url;
 use warg_api::v1::package::{
-    ContentSource, PackageError, PackageRecord, PackageRecordState, PublishRecordRequest,
+    ContentSource, MissingContent, PackageError, PackageRecord, PackageRecordState,
+    PublishRecordRequest, UploadEndpoint,
 };
 use warg_crypto::hash::{AnyHash, Sha256};
 use warg_protocol::{
@@ -83,13 +84,32 @@ impl Config {
         self.files_dir.join(self.content_file_name(digest))
     }
 
-    fn content_url(&self, digest: &AnyHash) -> String {
-        self.content_base_url
-            .join("content/")
-            .unwrap()
-            .join(&self.content_file_name(digest))
-            .unwrap()
-            .to_string()
+    fn content_url(&self, log_id: &LogId, record_id: &RecordId, digest: &AnyHash) -> String {
+        format!(
+            "{url}/{log_id}/record/{record_id}/content/{digest}",
+            url = self.content_base_url,
+        )
+    }
+
+    fn missing_content<'a>(
+        &self,
+        log_id: &LogId,
+        record_id: &RecordId,
+        missing_digests: impl IntoIterator<Item = &'a AnyHash>,
+    ) -> HashMap<AnyHash, MissingContent> {
+        missing_digests
+            .into_iter()
+            .map(|digest| {
+                (
+                    digest.clone(),
+                    MissingContent {
+                        upload: vec![UploadEndpoint::HttpPost {
+                            url: self.content_url(log_id, record_id, digest),
+                        }],
+                    },
+                )
+            })
+            .collect()
     }
 }
 
@@ -231,13 +251,12 @@ async fn publish_record(
         ));
     }
 
+    let missing_content = config.missing_content(&log_id, &record_id, missing);
     Ok((
         StatusCode::ACCEPTED,
         Json(PackageRecord {
             id: record_id,
-            state: PackageRecordState::Sourcing {
-                missing_content: missing.into_iter().cloned().collect(),
-            },
+            state: PackageRecordState::Sourcing { missing_content },
         }),
     ))
 }
@@ -254,12 +273,13 @@ async fn get_record(
         .await?;
 
     match record.status {
-        RecordStatus::MissingContent(missing) => Ok(Json(PackageRecord {
-            id: record_id,
-            state: PackageRecordState::Sourcing {
-                missing_content: missing,
-            },
-        })),
+        RecordStatus::MissingContent(missing) => {
+            let missing_content = config.missing_content(&log_id, &record_id, &missing);
+            Ok(Json(PackageRecord {
+                id: record_id,
+                state: PackageRecordState::Sourcing { missing_content },
+            }))
+        }
         // Validated is considered still processing until included in a checkpoint
         RecordStatus::Pending | RecordStatus::Validated => Ok(Json(PackageRecord {
             id: record_id,
@@ -275,11 +295,11 @@ async fn get_record(
                 .as_ref()
                 .contents()
                 .into_iter()
-                .map(|d| {
+                .map(|digest| {
                     (
-                        d.clone(),
+                        digest.clone(),
                         vec![ContentSource::Http {
-                            url: config.content_url(d),
+                            url: config.content_url(&log_id, &record_id, digest),
                         }],
                     )
                 })
@@ -361,13 +381,16 @@ async fn upload_content(
     {
         config
             .core_service
-            .submit_package_record(log_id, record_id.clone())
+            .submit_package_record(log_id.clone(), record_id.clone())
             .await;
     }
 
     Ok((
         StatusCode::CREATED,
-        [(axum::http::header::LOCATION, config.content_url(&digest))],
+        [(
+            axum::http::header::LOCATION,
+            config.content_url(&log_id, &record_id, &digest),
+        )],
     ))
 }
 
