Update ipfs uploading

Author: lucemans

engine/src/handlers/car/mod.rs

@@ -106,9 +106,20 @@ impl CarHandler {
             if let Some(ipfs_cid) = payload.cid {
                 Deployment::update_ipfs_cid(&state.database, &payload.deployment_id, &ipfs_cid)
                     .await
-                    .ok();
+                    .ok(); 
             }
 
+            // if let Some(file_path) = payload.file_path {
+            //     // todo pin file using ipfs-cluster
+
+            //     const ipfs_cluster_url = "http://0.0.0.0:44685";
+
+            //     // download from ipfs
+
+            //     // POST /add?local=true&format=car
+            //     // form-data, file: deploy.car                
+            // }
+
             delivery.ack(BasicAckOptions::default()).await.unwrap();
         }
         tracing::error!("Consumer stream ended");

engine/src/ipfs/mod.rs

@@ -1,24 +1,11 @@
-
-pub struct IPFSCluster {
+#[derive(Debug, Clone)]
+pub struct IPFSModule {
     pub cluster_url: String,
+    pub public_cluster_url: String,
 }
 
-impl IPFSCluster {
-    pub fn new(cluster_url: String) -> Self {
-        Self { cluster_url }
+impl IPFSModule {
+    pub fn new(cluster_url: String, public_cluster_url: String) -> Self {
+        Self { cluster_url, public_cluster_url }
     }
-
-    // pub async fn add_car(&self, car_path: &str) -> Result<String, Box<dyn std::error::Error>> {
-    //     let client = reqwest::Client::new();
-    //     let response = client.post(&format!("{}/add", self.cluster_url))
-    //         .header("Content-Type", "multipart/form-data")
-    //         .body(format!("file=@{}", car_path))
-    //         .send()
-    //         .await?;
-
-    //     let body = response.text().await?;
-    //     let json: serde_json::Value = serde_json::from_str(&body)?;
-    //     let cid = json["Hash"].as_str().ok_or("CID not found in response")?;
-    //     Ok(cid.to_string())
-    // }
 }

engine/src/routes/mod.rs

@@ -24,9 +24,10 @@ pub mod invite;
 pub mod site;
 pub mod team;
 pub mod user;
+pub mod system;
 
 fn get_api() -> impl OpenApi {
-    (site::api_routes(), UserApi, AuthApi, team::api_routes(), invite::api_routes())
+    (site::api_routes(), UserApi, AuthApi, team::api_routes(), invite::api_routes(), system::SystemApi)
 }
 
 #[derive(Tags)]
@@ -41,6 +42,8 @@ pub enum ApiTags {
     Team,
     /// User-related endpoints
     User,
+    /// System-related endpoints
+    System,
     /// Authentication-related endpoints
     #[oai(rename = "Authentication")]
     Auth,
@@ -157,6 +160,7 @@ pub async fn serve(state: State) {
         "User",
         "Team",
         "Invite",
+        "System",
     ];
 
     // Reorder tags according to the specified order

engine/src/routes/system.rs

@@ -0,0 +1,31 @@
+use crate::state::State;
+use poem::web::Data;
+use poem_openapi::{payload::Json, ApiResponse, Object, OpenApi};
+
+pub struct SystemApi;
+
+#[derive(Debug, Clone, Object)]
+pub struct IPFSStatus {
+    pub public_cluster_url: String,
+}
+
+#[derive(ApiResponse)]
+pub enum IPFSStatusResponse {
+    #[oai(status = 200)]
+    Ok(Json<IPFSStatus>),
+    #[oai(status = 100)]
+    FeatureDisabled(Json<String>),
+}
+
+#[OpenApi]
+impl SystemApi {
+    #[oai(path = "/system/ipfs", method = "get")]
+    async fn status(&self, state: Data<&State>) -> IPFSStatusResponse {
+        match &state.ipfs {
+            Some(ipfs) => IPFSStatusResponse::Ok(Json(IPFSStatus {
+                public_cluster_url: ipfs.public_cluster_url.clone(),
+            })),
+            None => IPFSStatusResponse::FeatureDisabled(Json("IPFS is not enabled".to_string())),
+        }
+    }
+}

engine/src/state.rs

@@ -4,7 +4,7 @@ use color_eyre::eyre::Result;
 use figment::{Figment, providers::Env};
 use serde::Deserialize;
 
-use crate::{cache::Cache, database::Database, handlers::TaskRabbit, storage::Storage};
+use crate::{cache::Cache, database::Database, handlers::TaskRabbit, ipfs::IPFSModule, storage::Storage};
 
 pub type State = Arc<AppState>;
 
@@ -15,6 +15,7 @@ pub struct AppState {
     pub storage: Storage,
     pub cache: Cache,
     pub rabbit: Option<TaskRabbit>,
+    pub ipfs: Option<IPFSModule>,
 }
 
 #[derive(Deserialize, Debug)]
@@ -26,6 +27,7 @@ pub struct AppConfig {
     pub s3_car: Option<S3CarConfig>,
     pub github_app: Option<GithubAppConfig>,
     pub amqp: Option<AMQPConfig>,
+    pub ipfs: Option<IPFSConfig>,
 }
 
 #[derive(Deserialize, Debug)]
@@ -79,6 +81,12 @@ pub struct GithubAppConfig {
     pub client_secret: String,
 }
 
+#[derive(Deserialize, Debug)]
+pub struct IPFSConfig {
+    pub cluster_url: String,
+    pub public_cluster_url: String,
+}
+
 impl AppState {
     pub async fn new() -> Result<Self> {
         // let config = Config::builder()
@@ -100,6 +108,8 @@ impl AppState {
                 .map(|key| format!("github_app.{}", key.as_str().to_lowercase()).into()))
             .merge(Env::prefixed("AMQP_")
                 .map(|key| format!("amqp.{}", key.as_str().to_lowercase()).into()))
+            .merge(Env::prefixed("IPFS_")
+                .map(|key| format!("ipfs.{}", key.as_str().to_lowercase()).into()))
             .extract::<AppConfig>()
             .expect("Failed to load AppConfig configuration");
 
@@ -115,12 +125,19 @@ impl AppState {
             None
         };
 
+        let ipfs = if let Some(ipfs) = &config.ipfs {
+            Some(IPFSModule::new(ipfs.cluster_url.clone(), ipfs.public_cluster_url.clone()))
+        } else {
+            None
+        };
+
         Ok(Self {
             config,
             database,
             storage,
             cache,
             rabbit,
+            ipfs,
         })
     }
 }

services/ipfs-cart/.env.example

@@ -8,3 +8,6 @@ S3_REGION=us-east-1
 S3_BUCKET_NAME=car
 S3_ACCESS_KEY=minioadmin
 S3_SECRET_KEY=minioadmin
+
+# IPFS Cluster Configuration
+IPFS_CLUSTER_URL=http://0.0.0.0:9094

services/ipfs-cart/src/config.ts

@@ -19,6 +19,9 @@ const defaultConfig: Config = {
     bucket: process.env.S3_BUCKET_NAME || "",
     endpoint: process.env.S3_ENDPOINT_URL || undefined,
   },
+  ipfsCluster: {
+    url: process.env.IPFS_CLUSTER_URL || "http://0.0.0.0:9094",
+  },
 };
 
 export default defaultConfig; 

services/ipfs-cart/src/rabbitmq.ts

@@ -10,6 +10,7 @@ import AdmZip from "adm-zip";
 import { readdir } from "fs/promises";
 import { createCarFile } from "./car";
 import { mkdir } from "fs/promises";
+import { uploadCar } from "./uploadCar";
 
 // Maximum number of retries before marking a message as failed
 const MAX_RETRIES = 4;
@@ -222,6 +223,13 @@ async function processCarRequest(
         console.log("CAR file uploaded to S3:", s3Key);
         console.log("Root CID:", rootCID);
 
+        // upload to ipfs cluster
+        console.log("Uploading CAR file to IPFS cluster...");
+        
+        const ipfsResponse = await uploadCar(carFilePath);
+
+        console.log("IPFS response:", ipfsResponse);
+
         outputFilePath = s3Key;
         outputCID = rootCID;
     } catch (error) {

services/ipfs-cart/src/types.ts

@@ -13,6 +13,9 @@ export interface Config {
         bucket: string;
         endpoint?: string;
     };
+    ipfsCluster: {
+        url: string;
+    };
 }
 
 export interface CarRequest {

services/ipfs-cart/src/uploadCar.ts

@@ -0,0 +1,31 @@
+import config from "./config";
+
+type IpfsClusterResponse = {
+    name: string; // ""
+    cid: string; // "bafy..."
+    bytes: number; // 108925
+    allocations: string[]; // ["12D3..."]
+}
+
+export const uploadCar = async (carFilePath: string) => {
+    if (!config.ipfsCluster.url) {
+        throw new Error("IPFS cluster URL is not set, cannot upload CAR file ::uploadCar.ts");
+    }
+
+    const url = `${config.ipfsCluster.url}/add?local=true&format=car`;
+
+    const file = await Bun.file(carFilePath).arrayBuffer();
+    const formData = new FormData();
+    formData.append("file", new Blob([file]));
+
+    const response = await fetch(url, {
+        method: "POST",
+        body: formData,
+    });
+
+    const data = await response.json() as IpfsClusterResponse;
+
+    console.log("IPFS response:", response);
+
+    return data;
+};