Start on ipfs-cart

Author: lucemans

bun.lockb

@@ -1,6 +1,6 @@
 {
   "name": "bunshot",
-  "version": "1.0.0",
+  "version": "0.0.1",
   "description": "A Bun service that takes screenshots of websites using Puppeteer and uploads them to S3",
   "main": "src/index.ts",
   "type": "module",

bunshot/package.json

@@ -15,6 +15,13 @@ S3_PREVIEWS_BUCKET_NAME=screenshots
 S3_PREVIEWS_ACCESS_KEY=minioadmin
 S3_PREVIEWS_SECRET_KEY=minioadmin
 
+# S3 Car (Defaults)
+S3_CAR_ENDPOINT_URL=http://localhost:9000
+S3_CAR_REGION=us-east-1
+S3_CAR_BUCKET_NAME=car
+S3_CAR_ACCESS_KEY=minioadmin
+S3_CAR_SECRET_KEY=minioadmin
+
 GITHUB_APP_CLIENT_ID=*******************
 GITHUB_APP_CLIENT_SECRET=*******************************
 # RUST_LOG=info,sqlx=trace

engine/.env.example

@@ -32,6 +32,7 @@ services:
         mc alias set minio http://127.0.0.1:9000 minioadmin minioadmin     # setup Minio client
         mc mb minio/edgeserver || true                                    # create a test bucket
         mc mb minio/screenshots || true                                    # create a test bucket
+        mc mb minio/car || true                                    # create a test bucket
         mc anonymous set public minio/edgeserver                          # make the test bucket public
         mc anonymous set public minio/screenshots                          # make the test bucket public
         kill -s INT $(cat /tmp/minio.pid) && rm /tmp/minio.pid             # stop Minio

engine/compose.yaml

@@ -1,11 +1,6 @@
 use crate::{models::deployment::{DeploymentFile, DeploymentFileEntry}, state::State};
 use serde::{Deserialize, Serialize};
 
-#[derive(Debug, Serialize, Deserialize)]
-pub struct CarRequest {
-    pub deployment_id: String,
-    pub files: Vec<DeploymentFileEntry>,
-}
 
 pub struct IPFSCluster {
     pub cluster_url: String,
@@ -16,17 +11,6 @@ impl IPFSCluster {
         Self { cluster_url }
     }
 
-    pub async fn pin_deployment(&self, deployment_id: String, state: &State) -> Result<String, Box<dyn std::error::Error>> {
-        let deployment_files = DeploymentFile::get_deployment_files(&state.database, &deployment_id).await?;
-
-        let car_request = CarRequest {
-            deployment_id,
-            files: deployment_files,
-        };
-
-        todo!()
-    }
-
     // 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))

engine/src/ipfs/mod.rs

@@ -84,16 +84,14 @@ impl Deployment {
     // }
 
     #[tracing::instrument(name = "upload_files", skip(self, state, file))]
-    pub async fn upload_files(&self, state: &State, file: Upload) -> Result<(), sqlx::Error> {
+    pub async fn upload_files(&self, state: &State, file: Vec<u8>) -> Result<(), sqlx::Error> {
         let span = info_span!("Deployment::upload_files");
         span.set_parent(Context::current());
         let _guard = span.enter();
 
-        let file_stream = file.into_vec().await.unwrap();
-
         // TODO: Read file stream, extract zip file (contains multiple files), upload each file to s3 at the correct relevant path relative to deployment.deployment_id + '/'
 
-        let zip = ZipFileReader::new(file_stream).await.unwrap();
+        let zip = ZipFileReader::new(file).await.unwrap();
 
         for index in 0..zip.file().entries().len() {
             let file = zip.file().entries().get(index).unwrap();

engine/src/models/deployment/mod.rs

@@ -13,6 +13,7 @@ pub struct TaskRabbit {
     channel: Channel,
 
     previews_queue_key: String,
+    car_queue_key: String,
 }
 
 #[derive(Debug, Serialize, Deserialize)]
@@ -22,6 +23,12 @@ pub struct BunshotPayload {
     pub domain: String,
 }
 
+#[derive(Debug, Serialize, Deserialize)]
+pub struct CarRequest {
+    pub deployment_id: String,
+    pub file_path: String,
+}
+
 impl TaskRabbit {
     pub async fn init(config: &AMQPConfig) -> TaskRabbit {
         let connection = Connection::connect(config.addr.as_str(), ConnectionProperties::default())
@@ -53,10 +60,26 @@ impl TaskRabbit {
 
         info!("Bunshot queue declared");
 
+        let car_channel = connection.create_channel().await.unwrap();
+
+        let car_queue_key = config.car_queue.as_deref().unwrap_or("car").to_string();
+
+        car_channel
+            .queue_declare(
+                &car_queue_key,
+                QueueDeclareOptions {
+                    durable: true,
+                    ..QueueDeclareOptions::default()
+                },
+                FieldTable::default(),
+            )
+            .await
+            .unwrap();
         TaskRabbit {
             connection,
             channel: bunshot_channel,
             previews_queue_key,
+            car_queue_key,
         }
     }
 
@@ -89,4 +112,29 @@ impl TaskRabbit {
             info!("Failed to publish message to screenshots queue");
         }
     }
+
+    pub async fn queue_car(&self, request: CarRequest) {
+        let payload = serde_json::to_string(&request).unwrap();
+
+        // Publish to the car queue
+        let confirm = self
+            .channel
+            .basic_publish(
+                "",
+                &self.car_queue_key,
+                BasicPublishOptions::default(),
+                payload.as_bytes(),
+                BasicProperties::default().with_delivery_mode(2),
+            )
+            .await
+            .unwrap()
+            .await
+            .unwrap();
+
+        if confirm == Confirmation::NotRequested || confirm == Confirmation::Ack(None) {
+            info!("Message successfully published to car queue");
+        } else {
+            info!("Failed to publish message to car queue");
+        }
+    }
 }

engine/src/rabbit.rs

@@ -3,12 +3,11 @@ use poem_openapi::{param::Path, payload::Json, OpenApi};
 use tracing::info;
 
 use crate::{
-    middlewares::auth::UserAuth,
-    models::{
-        deployment::{preview::DeploymentPreview, Deployment, DeploymentFile, DeploymentFileEntry}, domain::Domain, site::{Site, SiteId}
-    },
-    routes::{error::HttpError, ApiTags},
-    state::State,
+    middlewares::auth::UserAuth, models::{
+        deployment::{preview::DeploymentPreview, Deployment, DeploymentFile, DeploymentFileEntry},
+        domain::Domain,
+        site::{Site, SiteId},
+    }, rabbit::CarRequest, routes::{error::HttpError, ApiTags}, state::State
 };
 
 use super::UploadPayload;
@@ -113,6 +112,22 @@ impl SiteDeploymentsApi {
         info!("Deployment complete");
 
         if let Some(data) = payload.data {
+            let data = data.into_vec().await.unwrap();
+
+            if let Some(car_bucket) = &state.storage.car_bucket {
+                let path = format!("{}/car.zip", deployment.deployment_id);
+                let _ = car_bucket.put_object(&path, &data).await.unwrap();
+                
+                info!("Car uploaded to: {:?}", path);
+
+                if let Some(rabbit) = &state.rabbit {
+                    rabbit.queue_car(CarRequest {
+                        deployment_id: deployment.deployment_id.clone(),
+                        file_path: path.clone(),
+                    }).await;
+                }
+            }
+
             Deployment::upload_files(&deployment, &state, data)
                 .await
                 .unwrap();
@@ -158,6 +173,22 @@ impl SiteDeploymentsApi {
         info!("Deployment complete");
 
         if let Some(data) = payload.data {
+            let data = data.into_vec().await.unwrap();
+
+            if let Some(car_bucket) = &state.storage.car_bucket {
+                let path = format!("{}/car.zip", deployment.deployment_id);
+                let _ = car_bucket.put_object(&path, &data).await.unwrap();
+
+                if let Some(rabbit) = &state.rabbit {
+                    rabbit.queue_car(CarRequest {
+                        deployment_id: deployment_id.clone(),
+                        file_path: path.clone(),
+                    }).await;
+                }
+
+                info!("Car uploaded to: {:?}", path);
+            }
+
             Deployment::upload_files(&deployment, &state, data)
                 .await
                 .unwrap();
@@ -182,7 +213,9 @@ impl SiteDeploymentsApi {
             if let Some(rabbit) = &state.rabbit {
                 info!("Queueing bunshot for domain: {:?}", domain);
                 let domain = domain.domain();
-                rabbit.queue_bunshot(&site_id.0, &deployment_id, &domain).await;
+                rabbit
+                    .queue_bunshot(&site_id.0, &deployment_id, &domain)
+                    .await;
             }
         } else {
             info!("No domain was found for this site");
@@ -246,7 +279,9 @@ impl SiteDeploymentsApi {
 
             if let Some(domain) = domain {
                 info!("Queueing bunshot for domain: {:?}", domain.domain());
-                rabbit.queue_bunshot(&site_id.0, &deployment_id.0, &domain.domain()).await;
+                rabbit
+                    .queue_bunshot(&site_id.0, &deployment_id.0, &domain.domain())
+                    .await;
             } else {
                 info!("No domain was found for this site");
             }

engine/src/routes/site/deployments/mod.rs

@@ -23,6 +23,7 @@ pub struct AppConfig {
     pub database_url: String,
     pub s3: S3Config,
     pub s3_previews: Option<S3PreviewsConfig>,
+    pub s3_car: Option<S3CarConfig>,
     pub github_app: Option<GithubAppConfig>,
     pub amqp: Option<AMQPConfig>,
 }
@@ -45,10 +46,20 @@ pub struct S3PreviewsConfig {
     pub secret_key: String,
 }
 
+#[derive(Deserialize, Debug)]
+pub struct S3CarConfig {
+    pub endpoint_url: String,
+    pub region: String,
+    pub bucket_name: String,
+    pub access_key: String,
+    pub secret_key: String,
+}
+
 #[derive(Deserialize, Debug)]
 pub struct AMQPConfig {
     pub addr: String,
     pub previews_queue: Option<String>,
+    pub car_queue: Option<String>,
 }
 
 /// Github App Config
@@ -82,6 +93,8 @@ impl AppState {
                 .map(|key| format!("s3.{}", key.as_str().to_lowercase()).into()))
             .merge(Env::prefixed("S3_PREVIEWS_")
                 .map(|key| format!("s3_previews.{}", key.as_str().to_lowercase()).into()))
+            .merge(Env::prefixed("S3_CAR_")
+                .map(|key| format!("s3_car.{}", key.as_str().to_lowercase()).into()))
             .merge(Env::prefixed("GITHUB_APP_")
                 .map(|key| format!("github_app.{}", key.as_str().to_lowercase()).into()))
             .merge(Env::prefixed("AMQP_")

engine/src/state.rs

@@ -8,6 +8,7 @@ use crate::state::AppConfig;
 pub struct Storage {
     pub bucket: Box<Bucket>,
     pub previews_bucket: Option<Box<Bucket>>,
+    pub car_bucket: Option<Box<Bucket>>,
 }
 
 impl Storage {
@@ -50,7 +51,29 @@ impl Storage {
             None
         };
 
-        Self { bucket, previews_bucket }
+        let car_bucket = if let Some(car_config) = &config.s3_car {
+            let car_region = Region::Custom {
+                region: car_config.region.clone(),
+                endpoint: car_config.endpoint_url.clone(),
+            };
+            let car_credentials = Credentials::new(
+                Some(&car_config.access_key),
+                Some(&car_config.secret_key),
+                None,
+                None,
+                None,
+            )
+            .unwrap();
+            let car_bucket = Bucket::new(&car_config.bucket_name, car_region, car_credentials)
+                .unwrap()
+                .with_path_style();
+
+            Some(car_bucket)
+        } else {
+            None
+        };
+
+        Self { bucket, previews_bucket, car_bucket }
     }
 
     pub async fn upload(