Introduce basic service streaming

Author: lucemans

engine/src/main.rs

@@ -2,7 +2,6 @@ use std::env;
 use std::sync::Arc;
 
 use async_std::prelude::FutureExt;
-use async_std::task::JoinHandle;
 use opentelemetry::KeyValue;
 use opentelemetry::{global, trace::TracerProvider};
 use opentelemetry_otlp::WithExportConfig;

engine/src/models/deployment/mod.rs

@@ -260,6 +260,20 @@ impl Deployment {
 
         Ok(())
     }
+
+    pub async fn get_last_by_site_id(db: &Database, site_id: &str) -> Result<Self, sqlx::Error> {
+        let span = info_span!("Deployment::get_last_by_site_id");
+        span.set_parent(Context::current());
+        let _guard = span.enter();
+
+        query_as!(
+            Deployment,
+            "SELECT * FROM deployments WHERE site_id = $1 ORDER BY created_at DESC LIMIT 1",
+            site_id
+        )
+        .fetch_one(&db.pool)
+        .await
+    }
 }
 
 #[derive(Debug, Serialize, Deserialize, Object)]
@@ -297,6 +311,33 @@ impl DeploymentFile {
         .fetch_all(&db.pool)
         .await
     }
+
+    pub async fn get_file_by_path(db: &Database, deployment_id: &str, path: &str) -> Result<DeploymentFileEntry, sqlx::Error> {
+        let span = info_span!("DeploymentFile::get_file_by_path");
+        span.set_parent(Context::current());
+        let _guard = span.enter();
+
+        query_as!(
+            DeploymentFileEntry,
+            r#"
+            SELECT
+                df.deployment_id as "deployment_file_deployment_id!",
+                df.file_id as "deployment_file_file_id!",
+                df.file_path as "deployment_file_file_path!",
+                df.mime_type as "deployment_file_mime_type!",
+                f.file_hash as "file_hash!",
+                f.file_size,
+                f.file_deleted
+            FROM deployment_files df
+            JOIN files f ON df.file_id = f.file_id
+            WHERE df.deployment_id = $1 AND df.file_path = $2
+            "#,
+            deployment_id,
+            path
+        )
+        .fetch_one(&db.pool)
+        .await
+    }
 }
 
 // Add this new struct to represent the joined result

engine/src/server/mod.rs

@@ -3,25 +3,24 @@ use std::sync::Arc;
 use opentelemetry::global;
 use poem::middleware::OpenTelemetryMetrics;
 use poem::web::Data;
-use poem::Response;
-use poem::{
-    endpoint::StaticFilesEndpoint, get, handler, listener::TcpListener, middleware::Cors,
-    web::Html, EndpointExt, Route, Server,
-};
-use poem_openapi::{OpenApi, OpenApiService, Tags};
-use serde_json::{self, Value};
+use poem::{Body, EndpointExt, IntoResponse, Request, Response};
+use poem::{get, handler, listener::TcpListener, middleware::Cors, Route, Server};
+use reqwest::StatusCode;
 use tracing::info;
+use futures::StreamExt;
 
 use crate::middlewares::tracing::TraceId;
+use crate::models::deployment::{Deployment, DeploymentFile};
+use crate::models::domain::Domain;
+use crate::models::site::Site;
+use crate::routes::error::HttpError;
 use crate::state::State;
 
 pub async fn serve(state: State) {
     info!("Serving Router");
 
     let app = Route::new()
-        // .nest("/api", api_service)
-        // .nest("/", file_endpoint)
-        .at("/", get(not_found))
+        .at("*", get(resolve_http))
         .with(Cors::new())
         .with(TraceId::new(Arc::new(global::tracer("edgeserver"))))
         .with(OpenTelemetryMetrics::new())
@@ -33,7 +32,88 @@ pub async fn serve(state: State) {
 }
 
 #[handler]
-async fn not_found() -> Html<&'static str> {
+async fn resolve_http(request: &Request, state: Data<&State>) -> impl IntoResponse {
+    let headers = request.headers();
+    let host = match headers.get("host") {
+        Some(host) => host.to_str().unwrap_or("localhost"),
+        None => "localhost",
+    };
+    let path = request.uri().path();
+    let path = path.trim_start_matches('/');
+
+    info!("Router request at: {} {}", host, path);
+
+    let deployment = get_last_deployment(host, &state.clone()).await;
+
+    if let Ok(deployment) = deployment {
+        info!("Deployment found: {}", deployment.deployment_id);
+
+        let path = if path.is_empty() {
+            "index.html"
+        } else {
+            path
+        };
+
+        let file = DeploymentFile::get_file_by_path(&state.clone().database, &deployment.deployment_id, path)
+            .await.ok();
+
+        if let Some(deployment_file) = file {
+            info!("File found: {}", deployment_file.file_hash);
+
+            // stream file from s3 storage and return it
+            let s3_path = deployment_file.file_hash.clone();
+            if let Ok(s3_data) = state.storage.bucket.get_object_stream(s3_path).await {
+                // Stream the S3 response bytes directly to the client
+                let stream = s3_data.bytes.map(|chunk| {
+                    chunk.map_err(|e| {
+                        info!("Error streaming file: {}", e);
+                        std::io::Error::new(std::io::ErrorKind::Other, e)
+                    })
+                });
+
+                let body = Body::from_bytes_stream(stream);
+                return Response::builder()
+                    .status(StatusCode::OK)
+                    .header("content-type", deployment_file.deployment_file_mime_type.clone())
+                    .body(body);
+            } else {
+                info!("File not found in s3");
+            }
+        } else {
+            info!("No file found");
+        }
+    } else {
+        info!("No deployment found");
+    }
+
     // inline 404 template
-    Html(include_str!("./404.html"))
+    Response::builder()
+        .status(StatusCode::NOT_FOUND)
+        .body(Body::from_string(include_str!("./404.html").to_string()))
+}
+
+async fn get_last_deployment(host: &str, state: &State) -> Result<Deployment, HttpError> {
+        // get domain
+    // get site
+    // get last deployment
+    // get file at path in deployment
+    // otherwise return default /index.html if exists
+    let domain = Domain::existing_domain_by_name(host, &state.clone())
+        .await.ok().flatten();
+
+    if let Some(domain) = domain {
+        let site = Site::get_by_id(&state.clone().database, &domain.site_id)
+            .await.ok();
+
+        if let Some(site) = site {
+            let deployment = Deployment::get_last_by_site_id(&state.clone().database, &site.site_id)
+                .await.ok();
+
+            if let Some(deployment) = deployment {
+                return Ok(deployment);
+            }
+        }
+    }
+
+    Err(HttpError::NotFound)
 }