publish from server admin

docs/server_guide.md

@@ -79,8 +79,6 @@ cargo run -p file-service -- --config ./file-server/template.toml
 
 5. Access services via the additional endpoints:
 
-
-
 **Cost and Status API**
 
 Schema is provided at the `server-[]-schema.json`. One can access the graphql playground by navigating to the server's endpoint at `/files-status` or `/files-cost`.
@@ -115,10 +113,16 @@ Available mutations you can make, in addition to Status queries, are to add and
 Curl query will be similar to the above examples, here we provide an example in the GraphQL version
 ```
 mutation{
+  # Add an existing bundle
   addBundles(deployments:["QmeD3dRVV6Gs84TRwiNj3tLt9mBEMVqy3GoWm7WN8oDzGz", "QmeaPp764FjQjPB66M9ijmQKmLhwBpHQhA7dEbH2FA1j3v"], 
     locations:["./example-file", "./example-file"]){
     ipfsHash
   }
+
+  # Create, publish, and serve a bundle
+  publishAndServeBundle(filenames: ["0017686116-5331aab87811944d-f8d105f60fa2e78d-17686021-default.dbin", "0017686115-f8d105f60fa2e78d-7d23a3e458beaff1-17686021-default.dbin"]) {
+    ipfsHash
+  }
 }
 ```
 with configuration 

file-exchange/src/manifest/store.rs

@@ -479,7 +479,6 @@ mod tests {
             main_dir: main_directory.to_string(),
         }))
         .unwrap();
-        let _local_path = Path::from("");
         let local = LocalBundle {
             bundle: bundle.clone(),
             local_path: Path::from(""),

file-exchange/tests/admin.rs

@@ -0,0 +1,147 @@
+#[cfg(test)]
+mod tests {
+    use serde_json::{json, Value};
+    use std::{process::Command, time::Duration};
+    use tempfile::tempdir;
+    use tokio::fs;
+
+    use file_exchange::{
+        config::{DownloaderArgs, LocalDirectory},
+        download_client::Downloader,
+        manifest::ipfs::IpfsClient,
+        test_util::server_ready,
+    };
+
+    #[tokio::test]
+    async fn test_admin_publish_and_serve() {
+        std::env::set_var("RUST_LOG", "off,file_exchange=trace,file_transfer=trace,file_service=trace,indexer_service=trace,indexer_common=trace");
+        file_exchange::config::init_tracing("pretty").unwrap();
+
+        let client = IpfsClient::new("https://ipfs.network.thegraph.com")
+            .expect("Could not create client to thegraph IPFS gateway");
+        // 1. Setup server
+        let mut server_process = Command::new("cargo")
+            .arg("run")
+            .arg("-p")
+            .arg("file-service")
+            .arg("--")
+            .arg("--config")
+            .arg("./tests/test0.toml")
+            .spawn()
+            .expect("Failed to start server");
+        tracing::debug!("Wait 10 seconds");
+        tokio::time::sleep(Duration::from_secs(3)).await;
+        let _ = server_ready("http://localhost:5677").await;
+
+        // 2. admin command
+        // Read status
+        let bundles_query = json!({
+            "query": r#"query {
+              bundles{
+                ipfsHash
+              }
+            }"#
+        });
+        let res = reqwest::Client::new()
+            .post("http://0.0.0.0:5664/admin")
+            .header("Content-Type", "application/json")
+            .header("Authorization", "Bearer slayyy")
+            .json(&bundles_query)
+            .send()
+            .await
+            .expect("GraphQL admin bundle request gets response");
+        assert!(res.status().is_success());
+        let response_body: Value = res.json().await.expect("Status response is a json value");
+        // Navigate through the JSON to extract the `ipfsHash` values
+        let ipfs_hashes: Vec<String> = response_body["data"]["bundles"]
+            .as_array()
+            .expect("Bundles in response")
+            .iter()
+            .filter_map(|bundle| bundle["ipfsHash"].as_str().map(String::from))
+            .collect();
+        assert!(ipfs_hashes.len() == 1);
+
+        // Mutation: publish and serve new files
+        let graphql_query = json!({
+            "query": r#"mutation {
+                publishAndServeBundle(filenames: ["0017686116-5331aab87811944d-f8d105f60fa2e78d-17686021-default.dbin", "0017686115-f8d105f60fa2e78d-7d23a3e458beaff1-17686021-default.dbin"]) {
+                    ipfsHash
+                }
+            }"#
+        });
+
+        let res = reqwest::Client::new()
+            .post("http://0.0.0.0:5664/admin")
+            .header("Content-Type", "application/json")
+            .header("Authorization", "Bearer slayyy")
+            .json(&graphql_query)
+            .send()
+            .await
+            .expect("GraphQL admin mutation request gets a response");
+
+        assert!(res.status().is_success());
+        let response_body: Value = res.json().await.expect("Response is a json value");
+        let ipfs_hash = response_body["data"]["publishAndServeBundle"]["ipfsHash"]
+            .as_str()
+            .expect("Response has IPFS hash");
+        assert!(!ipfs_hashes.contains(&ipfs_hash.to_string())); // should be a new hash
+
+        // Check public status (checking 5665/admin should be the same)
+        let res = reqwest::Client::new()
+            .post("http://0.0.0.0:5677/files-status")
+            .header("Content-Type", "application/json")
+            .json(&bundles_query)
+            .send()
+            .await
+            .expect("GraphQL status bundle request gets response");
+        assert!(res.status().is_success());
+        let response_body: Value = res.json().await.expect("Status response is a json value");
+        // Navigate through the JSON to extract the `ipfsHash` values
+        let ipfs_hashes: Vec<String> = response_body["data"]["bundles"]
+            .as_array()
+            .expect("Bundles in response")
+            .iter()
+            .filter_map(|bundle| bundle["ipfsHash"].as_str().map(String::from))
+            .collect();
+        assert!(ipfs_hashes.contains(&ipfs_hash.to_string())); // should be a new hash
+
+        // 3. Setup downloader
+        let temp_dir = tempdir().unwrap();
+        let main_dir = temp_dir.path().to_path_buf();
+
+        let downloader_args = DownloaderArgs {
+            storage_method: file_exchange::config::StorageMethod::LocalFiles(LocalDirectory {
+                main_dir: main_dir.to_str().unwrap().to_string(),
+            }),
+            ipfs_hash: ipfs_hash.to_string(),
+            indexer_endpoints: [
+                "http://localhost:5679".to_string(),
+                "http://localhost:5677".to_string(),
+            ]
+            .to_vec(),
+            verifier: None,
+            mnemonic: None,
+            free_query_auth_token: Some("Bearer free-token".to_string()),
+            provider: None,
+            provider_concurrency: 2,
+            ..Default::default()
+        };
+
+        let downloader = Downloader::new(client, downloader_args).await;
+
+        // 4. Perform the download
+        let download_result = downloader.download_target().await;
+
+        // 5. Validate the download
+        tracing::info!(
+            result = tracing::field::debug(&download_result),
+            "Download result"
+        );
+        assert!(download_result.is_ok());
+        // Further checks can be added to verify the contents of the downloaded files
+
+        // 6. Cleanup
+        fs::remove_dir_all(temp_dir).await.unwrap();
+        let _ = server_process.kill();
+    }
+}

file-service/src/admin.rs

@@ -4,6 +4,10 @@ use std::sync::Arc;
 use async_graphql::{Context, EmptySubscription, MergedObject, Object, Schema};
 use async_graphql_axum::{GraphQLRequest, GraphQLResponse};
 use axum::{extract::State, routing::get, Router, Server};
+use file_exchange::{
+    config::{BundleArgs, PublisherArgs},
+    publisher::ManifestPublisher,
+};
 use http::HeaderMap;
 use tokio::sync::Mutex;
 
@@ -119,6 +123,94 @@ pub struct StatusMutation;
 
 #[Object]
 impl StatusMutation {
+    // Publish a bundle; the location of files are relative to the main directory of the server
+    #[allow(clippy::too_many_arguments)]
+    async fn publish_and_serve_bundle(
+        &self,
+        ctx: &Context<'_>,
+        filenames: Vec<String>,
+        chunk_size: Option<u64>,
+        bundle_name: Option<String>,
+        file_type: Option<String>,
+        bundle_version: Option<String>,
+        identifier: Option<String>,
+        start_block: Option<u64>,
+        end_block: Option<u64>,
+        description: Option<String>,
+        chain_id: Option<String>,
+    ) -> Result<GraphQlBundle, ServerError> {
+        if ctx.data_opt::<String>()
+            != ctx
+                .data_unchecked::<AdminContext>()
+                .state
+                .admin_auth_token
+                .as_ref()
+        {
+            return Err(ServerError::InvalidAuthentication(format!(
+                "Failed to authenticate: {:#?}",
+                ctx.data_opt::<String>(),
+            )));
+        }
+        // publish bundle
+        let client = ctx.data_unchecked::<AdminContext>().state.client.clone();
+        let publisher = ManifestPublisher::new(
+            client,
+            PublisherArgs {
+                chunk_size: chunk_size.unwrap_or(1048576),
+                filenames,
+                bundle: Some(BundleArgs {
+                    bundle_name,
+                    file_type,
+                    bundle_version,
+                    identifier,
+                    start_block,
+                    end_block,
+                    description,
+                    chain_id,
+                }),
+                storage_method: ctx
+                    .data_unchecked::<AdminContext>()
+                    .state
+                    .store
+                    .storage_method
+                    .clone(),
+                ..Default::default()
+            },
+        );
+
+        let published = publisher
+            .publish()
+            .await
+            .map_err(|e| ServerError::ContextError(e.to_string()))?;
+        let deployment = published
+            .first()
+            .ok_or(ServerError::ContextError("No bundle published".to_string()))?;
+
+        let bundle = match read_bundle(
+            &ctx.data_unchecked::<AdminContext>().state.client,
+            deployment,
+        )
+        .await
+        {
+            Ok(s) => s,
+            Err(e) => return Err(ServerError::RequestBodyError(e.to_string())),
+        };
+        let local_bundle = LocalBundle {
+            bundle: bundle.clone(),
+            //TODO: remove this field
+            local_path: "".into(),
+        };
+
+        ctx.data_unchecked::<AdminContext>()
+            .state
+            .bundles
+            .lock()
+            .await
+            .insert(bundle.ipfs_hash.clone(), local_bundle);
+
+        Ok(GraphQlBundle::from(bundle))
+    }
+
     // Add a bundle
     async fn add_bundle(
         &self,
@@ -419,6 +511,93 @@ impl StatusMutation {
         Ok(resolved_files.unwrap_or_default())
     }
 
+    // Publish a bundle; the location of files are relative to the main directory of the server
+    async fn publish_and_serve_files(
+        &self,
+        ctx: &Context<'_>,
+        filenames: Vec<String>,
+        chunk_size: Option<u64>,
+    ) -> Result<Vec<GraphQlFileManifestMeta>, ServerError> {
+        if ctx.data_opt::<String>()
+            != ctx
+                .data_unchecked::<AdminContext>()
+                .state
+                .admin_auth_token
+                .as_ref()
+        {
+            return Err(ServerError::InvalidAuthentication(format!(
+                "Failed to authenticate: {:#?}",
+                ctx.data_opt::<String>(),
+            )));
+        }
+        // publish bundle
+        let client = ctx.data_unchecked::<AdminContext>().state.client.clone();
+        let file_ref = ctx.data_unchecked::<AdminContext>().state.files.clone();
+        let publisher = ManifestPublisher::new(
+            client.clone(),
+            PublisherArgs {
+                chunk_size: chunk_size.unwrap_or(1048576),
+                filenames: filenames.clone(),
+                storage_method: ctx
+                    .data_unchecked::<AdminContext>()
+                    .state
+                    .store
+                    .storage_method
+                    .clone(),
+                ..Default::default()
+            },
+        );
+
+        let published = publisher
+            .publish()
+            .await
+            .map_err(|e| ServerError::ContextError(e.to_string()))?;
+
+        let files = published
+            .iter()
+            .zip(filenames)
+            .map(|(deployment, file_name)| {
+                let client = client.clone();
+                let file_ref = file_ref.clone();
+
+                async move {
+                    tracing::debug!(deployment, file_name, "Adding file");
+
+                    let file_manifest = fetch_file_manifest_from_ipfs(&client.clone(), deployment)
+                        .await
+                        .map_err(|e| ServerError::ContextError(e.to_string()))?;
+
+                    let meta = FileManifestMeta {
+                        meta_info: FileMetaInfo {
+                            name: file_name.clone(),
+                            hash: deployment.clone(),
+                        },
+                        file_manifest,
+                    };
+                    ctx.data_unchecked::<AdminContext>()
+                        .state
+                        .store
+                        .read_and_validate_file(&meta, None)
+                        .await
+                        .map_err(|e| ServerError::ContextError(e.to_string()))?;
+                    file_ref
+                        .clone()
+                        .lock()
+                        .await
+                        .insert(deployment.clone(), meta.clone());
+
+                    Ok::<_, crate::admin::ServerError>(GraphQlFileManifestMeta::from(meta))
+                }
+            })
+            .collect::<Vec<_>>();
+
+        // Since collect() gathers futures, we need to resolve them. You can use `try_join_all` for this.
+        let resolved_files: Result<Vec<GraphQlFileManifestMeta>, _> =
+            futures::future::try_join_all(files).await;
+
+        Ok(resolved_files.unwrap_or_default())
+    }
+
     async fn remove_file(
         &self,
         ctx: &Context<'_>,