Added callers moulde (#120)

* Added callers moulde

* Code formatting

* Added uuid dependency

* Added queue_song endpoint

* Added endpoint to router

* Added dependencies for postgres

* Added migrations directory

* Added more code

* Added migrations

* Removed migration

* Docker fixes

* Updated gitignore

* Updated env sample file

* Endpoint url fix

* Updated env sample

* Updated docker file to copy migrations

* Might undo later

* Updated env sample and docker compose

* Updated cargo dependencies

* Saving changes

* Workflow fix

* Will need to take another look at the dependencies again

* Added code

* Test is not finished yet

* Warning fix for now

* Updated workflow to setup database

* Workflow fix

* Workflow fix attempt

* Warning fix

* Warning fix

* Updated workflow

* Added test files

Might change later

* Added more crates

* Added more test code

Not finished

* Formatting

* Close, but far

* I believe it is now working

* Formatting

* Added more crates

* Code cleanup

* Code reformatting

* Code cleanup

* Crate cleanup

* Cleanup

.env.sample

@@ -1,2 +1,6 @@
-DATABASE_URL=postgres://icarus:password@localhost/icarus
-SECRET_KEY=refero34o8rfhfjn983thf39fhc943rf923n3h
+SECRET_MAIN_KEY=refero34o8rfhfjn983thf39fhc943rf923n3h
+POSTGRES_MAIN_USER=icarus
+POSTGRES_MAIN_PASSWORD=password
+POSTGRES_MAIN_DB=icarus_db
+POSTGRES_MAIN_HOST=main_db
+DATABASE_URL=postgres://${POSTGRES_MAIN_USER}:${POSTGRES_MAIN_PASSWORD}@${POSTGRES_MAIN_HOST}:5432/${POSTGRES_MAIN_DB}

.github/workflows/workflow.yml

@@ -2,25 +2,58 @@ name: Rust CI
 
 on:
   push:
-    branches: [ "v0.2" ] # Changed from main to master
+    branches: [ "v0.2" ]
   pull_request:
-    branches: [ "v0.2" ] # Changed from main to master
+    branches: [ "v0.2" ]
 
 jobs:
   build:
-
     runs-on: ubuntu-24.04
+    services:
+      postgres:
+        image: postgres:17.4
+        env:
+          POSTGRES_PASSWORD: ${{ secrets.POSTGRES_PASSWORD }}
+          POSTGRES_USER: ${{ secrets.POSTGRES_USER }}
+          POSTGRES_DB: ${{ secrets.POSTGRES_DB }}
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd pg_isready
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
 
     steps:
     - uses: actions/checkout@v4
+    
     - name: Install Rust 1.86.0
       uses: actions-rs/toolchain@v1
       with:
         toolchain: 1.86.0
         components: clippy, rustfmt
         override: true
+    
     - name: Cache dependencies
       uses: Swatinem/rust-cache@v2
+    
+    - name: Install libpq
+      run: sudo apt-get install -y libpq-dev
+      
+    - name: Setup test database
+      env:
+        DATABASE_URL: "postgres://${{ secrets.POSTGRES_USER }}:${{ secrets.POSTGRES_PASSWORD }}@localhost:5432/${{ secrets.POSTGRES_DB }}"
+      run: |
+        # Wait for PostgreSQL to be ready
+        for i in {1..10}; do
+          pg_isready -U ${{ secrets.POSTGRES_USER }} -d ${{ secrets.POSTGRES_DB }} && break
+          sleep 2
+        done
+        
+        # Run database migrations (if you use sqlx migrations)
+        cargo install sqlx-cli --no-default-features --features native-tls,postgres
+        sqlx migrate run --database-url $DATABASE_URL
+    
     - name: Build
       run: |
         mkdir -p ~/.ssh
@@ -30,9 +63,14 @@ jobs:
         eval $(ssh-agent -s)
         ssh-add -v ~/.ssh/gitea_deploy_key
         cargo build --verbose --release
+    
     - name: Run tests
+      env:
+        DATABASE_URL: "postgres://${{ secrets.POSTGRES_USER }}:${{ secrets.POSTGRES_PASSWORD }}@localhost:5432/${{ secrets.POSTGRES_DB }}"
       run: cargo test --verbose
+    
     - name: Run clippy
       run: cargo clippy -- -D warnings
+    
     - name: Run rustfmt
       run: cargo fmt --all -- --check

.gitignore

@@ -5,3 +5,4 @@ Cargo.lock
 .env
 
 .DS_STORE
+Storage/

Cargo.toml

@@ -4,10 +4,23 @@ version = "0.2.0"
 edition = "2024"
 
 [dependencies]
-axum = { version = "0.8.3" }
+axum = { version = "0.8.3", features = ["multipart"] }
 serde = { version = "1.0.219", features = ["derive"] }
 serde_json = { version = "1.0.140" }
-tokio = { version = "1.44.2", features = ["rt-multi-thread"] }
+tower = { version = "0.5.2" }
+tokio = { version = "1.44.2", features = ["full"] }
+tokio-util = { version = "0.7.14" }
+tower-http = { version = "0.6.2", features = ["timeout"] }
 tracing-subscriber = "0.3.19"
+# mime = { version = "0.3.17" }
+# mime_guess = { version = "2.0.5" }
+futures = { version = "0.3.31" }
+uuid = { version = "1.16.0", features = ["v4", "serde"] }
+sqlx = { version = "0.8.5", features = ["postgres", "runtime-tokio-native-tls", "time", "uuid"] }
+dotenvy = { version = "0.15.7" }
 icarus_meta = { git = "ssh://git@git.kundeng.us/phoenix/icarus_meta.git", tag = "v0.1.30" }
 icarus_models = { git = "ssh://git@git.kundeng.us/phoenix/icarus_models.git", tag = "v0.4.3" }
+
+[dev-dependencies]
+common-multipart-rfc7578 = { version = "0.7.0" }
+url = { version = "2.5.4" }

Dockerfile

@@ -35,6 +35,7 @@ RUN --mount=type=ssh mkdir src && \
 COPY src ./src
 # If you have other directories like `templates` or `static`, copy them too
 COPY .env ./.env
+COPY migrations ./migrations
 
 # << --- SSH MOUNT ADDED HERE --- >>
 # Build *only* dependencies to leverage Docker cache
@@ -60,6 +61,7 @@ COPY --from=builder /usr/src/app/target/release/icarus .
 # Copy other necessary files like .env (if used for runtime config) or static assets
 # It's generally better to configure via environment variables in Docker though
 COPY --from=builder /usr/src/app/.env .
+COPY --from=builder /usr/src/app/migrations ./migrations
 
 # Expose the port your Axum app listens on (e.g., 3000 or 8000)
 EXPOSE 3000

docker-compose.yaml

@@ -48,9 +48,9 @@ services:
     container_name: icarus_db # Optional: Give the container a specific name
     environment:
       # These MUST match the user, password, and database name in the DATABASE_URL above
-      POSTGRES_USER: ${POSTGRES_USER:-icarus}
-      POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:-password}
-      POSTGRES_DB: ${POSTGRES_DB:-icarus_db}
+      POSTGRES_USER: ${POSTGRES_MAIN_USER:-icarus}
+      POSTGRES_PASSWORD: ${POSTGRES_MAIN_PASSWORD:-password}
+      POSTGRES_DB: ${POSTGRES_MAIN_DB:-icarus_db}
     volumes:
       # Persist database data using a named volume
       - postgres_data:/var/lib/postgresql/data

migrations/20250420185217_init_migration.sql

@@ -0,0 +1,9 @@
+-- Add migration script here
+CREATE EXTENSION IF NOT EXISTS pgcrypto;
+
+-- Table to store queued songs to process
+CREATE TABLE IF NOT EXISTS "songQueue" (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    filename TEXT NOT NULL,
+    data BYTEA NOT NULL
+);

src/callers/mod.rs

@@ -0,0 +1,5 @@
+pub mod song;
+
+pub mod endpoints {
+    pub const QUEUESONG: &str = "/api/v2/song/queue";
+}

src/callers/song.rs

@@ -0,0 +1,106 @@
+pub mod request {
+    use serde::{Deserialize, Serialize};
+
+    #[derive(Default, Deserialize, Serialize)]
+    pub struct Request {
+        pub message: String,
+    }
+}
+
+pub mod response {
+    use serde::{Deserialize, Serialize};
+
+    #[derive(Default, Deserialize, Serialize)]
+    pub struct Response {
+        pub message: String,
+        pub data: Vec<uuid::Uuid>,
+    }
+}
+
+mod song_queue {
+    use sqlx::Row;
+
+    #[derive(Debug, serde::Serialize, sqlx::FromRow)]
+    pub struct InsertedData {
+        pub id: uuid::Uuid,
+    }
+
+    pub async fn insert(
+        pool: &sqlx::PgPool,
+        data: &Vec<u8>,
+        filename: &String,
+    ) -> Result<uuid::Uuid, sqlx::Error> {
+        let result = sqlx::query(
+            r#"
+            INSERT INTO "songQueue" (data, filename) VALUES($1, $2) RETURNING id;
+            "#,
+        )
+        .bind(data)
+        .bind(filename)
+        .fetch_one(pool)
+        .await
+        .map_err(|e| {
+            eprintln!("Error inserting: {}", e);
+        });
+
+        match result {
+            Ok(row) => {
+                let id: uuid::Uuid = row
+                    .try_get("id")
+                    .map_err(|_e| sqlx::Error::RowNotFound)
+                    .unwrap();
+                Ok(id)
+            }
+            Err(_err) => Err(sqlx::Error::RowNotFound),
+        }
+    }
+}
+
+pub mod endpoint {
+    use axum::{Json, http::StatusCode};
+    use std::io::Write;
+
+    use crate::callers::song::song_queue;
+
+    pub async fn queue_song(
+        axum::Extension(pool): axum::Extension<sqlx::PgPool>,
+        mut multipart: axum::extract::Multipart,
+    ) -> (StatusCode, Json<super::response::Response>) {
+        let mut results: Vec<uuid::Uuid> = Vec::new();
+        let mut response = super::response::Response::default();
+
+        while let Some(field) = multipart.next_field().await.unwrap() {
+            let name = field.name().unwrap().to_string();
+            let file_name = field.file_name().unwrap().to_string();
+            let content_type = field.content_type().unwrap().to_string();
+            let data = field.bytes().await.unwrap();
+
+            println!(
+                "Received file '{}' (name = '{}', content-type = '{}', size = {})",
+                file_name,
+                name,
+                content_type,
+                data.len()
+            );
+
+            // Save the file to disk
+            let mut file = std::fs::File::create(&file_name).unwrap();
+            file.write_all(&data).unwrap();
+
+            let raw_data: Vec<u8> = data.to_vec();
+            let queue_repo = song_queue::insert(&pool, &raw_data, &file_name)
+                .await
+                .unwrap();
+            results.push(queue_repo);
+        }
+
+        response.data = results;
+        response.message = if response.data.is_empty() {
+            String::from("Error")
+        } else {
+            String::from("Success")
+        };
+
+        (StatusCode::OK, Json(response))
+    }
+}

src/main.rs

@@ -1,29 +1,90 @@
-use axum::{
-    // Json,
-    Router,
-    // http::StatusCode,
-    // routing::{get, post},
-    routing::get,
-};
-// use serde::{Deserialize, Serialize};
+pub mod callers;
+
+pub mod db {
+
+    use sqlx::postgres::PgPoolOptions;
+    use std::env;
+
+    pub mod connection_settings {
+        pub const MAXCONN: u32 = 10;
+    }
+
+    pub mod keys {
+        pub const DBURL: &str = "DATABASE_URL";
+
+        pub mod error {
+            pub const ERROR: &str = "DATABASE_URL must be set in .env";
+        }
+    }
+
+    pub async fn create_pool() -> Result<sqlx::PgPool, sqlx::Error> {
+        let database_url = get_db_url().await;
+        println!("Database url: {:?}", database_url);
+
+        PgPoolOptions::new()
+            .max_connections(connection_settings::MAXCONN)
+            .connect(&database_url)
+            .await
+    }
+
+    async fn get_db_url() -> String {
+        dotenvy::dotenv().ok();
+        env::var(keys::DBURL).expect(keys::error::ERROR)
+    }
+
+    pub async fn migrations(pool: &sqlx::PgPool) {
+        // Run migrations using the sqlx::migrate! macro
+        // Assumes your migrations are in a ./migrations folder relative to Cargo.toml
+        sqlx::migrate!("./migrations")
+            .run(pool)
+            .await
+            .expect("Failed to run migrations");
+    }
+}
 
 #[tokio::main]
 async fn main() {
     // initialize tracing
     tracing_subscriber::fmt::init();
 
-    // build our application with a route
-    let app = Router::new()
-        // `GET /` goes to `root`
-        .route("/", get(root));
-    // `POST /users` goes to `create_user`
-    // .route("/users", post(create_user));
+    let pool = db::create_pool().await.expect("Failed to create pool");
+    db::migrations(&pool).await;
+
+    // build our application with a route
+    let app = init::app().await;
 
-    // run our app with hyper, listening globally on port 3000
     let listener = tokio::net::TcpListener::bind(get_full()).await.unwrap();
     axum::serve(listener, app).await.unwrap();
 }
 
+pub mod init {
+    use axum::routing::{get, post};
+    use std::time::Duration;
+    use tower_http::timeout::TimeoutLayer;
+
+    pub async fn routes() -> axum::Router {
+        axum::Router::new()
+            .route(crate::ROOT, get(crate::root))
+            .route(
+                crate::callers::endpoints::QUEUESONG,
+                post(crate::callers::song::endpoint::queue_song),
+            )
+    }
+
+    pub async fn app() -> axum::Router {
+        let pool = crate::db::create_pool()
+            .await
+            .expect("Failed to create pool");
+        crate::db::migrations(&pool).await;
+
+        routes()
+            .await
+            .layer(axum::Extension(pool))
+            .layer(axum::extract::DefaultBodyLimit::max(1024 * 1024 * 1024))
+            .layer(TimeoutLayer::new(Duration::from_secs(300)))
+    }
+}
+
 fn get_full() -> String {
     get_address() + ":" + &get_port()
 }
@@ -35,7 +96,153 @@ fn get_port() -> String {
     String::from("3000")
 }
 
+pub const ROOT: &str = "/";
 // basic handler that responds with a static string
-async fn root() -> &'static str {
+pub async fn root() -> &'static str {
     "Hello, World!"
 }
+
+#[cfg(test)]
+mod tests {
+    use crate::db;
+
+    use common_multipart_rfc7578::client::multipart::{
+        Body as MultipartBody, Form as MultipartForm,
+    };
+    use std::{time::Duration, usize};
+    use tower::ServiceExt;
+    use tower_http::timeout::TimeoutLayer;
+
+    mod db_mgr {
+        use std::str::FromStr;
+
+        use crate::db;
+
+        pub const LIMIT: usize = 6;
+
+        pub async fn get_pool() -> Result<sqlx::PgPool, sqlx::Error> {
+            dotenvy::dotenv().ok();
+            let tm_db_url = std::env::var(db::keys::DBURL).expect("DATABASE_URL must be present");
+            let tm_options = sqlx::postgres::PgConnectOptions::from_str(&tm_db_url).unwrap();
+            sqlx::PgPool::connect_with(tm_options).await
+        }
+
+        pub async fn generate_db_name() -> String {
+            let db_name =
+                get_database_name().unwrap() + &"_" + &uuid::Uuid::new_v4().to_string()[..LIMIT];
+            db_name
+        }
+
+        pub async fn connect_to_db(db_name: &str) -> Result<sqlx::PgPool, sqlx::Error> {
+            dotenvy::dotenv().ok();
+            let db_url =
+                std::env::var(db::keys::DBURL).expect("DATABASE_URL must be set for tests");
+            let options = sqlx::postgres::PgConnectOptions::from_str(&db_url)?.database(db_name);
+            sqlx::PgPool::connect_with(options).await
+        }
+
+        pub async fn create_database(
+            template_pool: &sqlx::PgPool,
+            db_name: &str,
+        ) -> Result<(), sqlx::Error> {
+            let create_query = format!("CREATE DATABASE {}", db_name);
+            match sqlx::query(&create_query).execute(template_pool).await {
+                Ok(_) => Ok(()),
+                Err(e) => Err(e),
+            }
+        }
+
+        // Function to drop a database
+        pub async fn drop_database(
+            template_pool: &sqlx::PgPool,
+            db_name: &str,
+        ) -> Result<(), sqlx::Error> {
+            let drop_query = format!("DROP DATABASE IF EXISTS {} WITH (FORCE)", db_name);
+            sqlx::query(&drop_query).execute(template_pool).await?;
+            Ok(())
+        }
+
+        pub fn get_database_name() -> Result<String, Box<dyn std::error::Error>> {
+            dotenvy::dotenv().ok(); // Load .env file if it exists
+
+            match std::env::var(db::keys::DBURL) {
+                Ok(database_url) => {
+                    let parsed_url = url::Url::parse(&database_url)?;
+                    if parsed_url.scheme() == "postgres" || parsed_url.scheme() == "postgresql" {
+                        match parsed_url
+                            .path_segments()
+                            .and_then(|segments| segments.last().map(|s| s.to_string()))
+                        {
+                            Some(sss) => Ok(sss),
+                            None => Err("Error parsing".into()),
+                        }
+                    } else {
+                        // Handle other database types if needed
+                        Err("Error parsing".into())
+                    }
+                }
+                Err(_) => {
+                    // DATABASE_URL environment variable not found
+                    Err("Error parsing".into())
+                }
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn test_song_queue() {
+        let tm_pool = db_mgr::get_pool().await.unwrap();
+        let db_name = db_mgr::generate_db_name().await;
+
+        match db_mgr::create_database(&tm_pool, &db_name).await {
+            Ok(_) => {
+                println!("Success");
+            }
+            Err(err) => {
+                assert!(false, "Error: {:?}", err);
+            }
+        }
+
+        let pool = db_mgr::connect_to_db(&db_name).await.unwrap();
+        db::migrations(&pool).await;
+
+        let app = crate::init::routes()
+            .await
+            .layer(axum::Extension(pool))
+            .layer(axum::extract::DefaultBodyLimit::max(1024 * 1024 * 1024))
+            .layer(TimeoutLayer::new(Duration::from_secs(300)));
+
+        // Create multipart form
+        let mut form = MultipartForm::default();
+        let _ = form.add_file("flac", "tests/Machine_gun/track01.flac");
+
+        // Create request
+        let content_type = form.content_type();
+        let body = MultipartBody::from(form);
+        let req = axum::http::Request::builder()
+            .method(axum::http::Method::POST)
+            .uri(crate::callers::endpoints::QUEUESONG)
+            .header(axum::http::header::CONTENT_TYPE, content_type)
+            .body(axum::body::Body::from_stream(body))
+            .unwrap();
+
+        // Send request
+        match app.oneshot(req).await {
+            Ok(response) => {
+                let body = axum::body::to_bytes(response.into_body(), usize::MAX)
+                    .await
+                    .unwrap();
+                println!("Body: {:?}", body);
+                let resp: crate::callers::song::response::Response =
+                    serde_json::from_slice(&body).unwrap();
+                assert_eq!(false, resp.data.is_empty(), "Should not be empty");
+                assert_eq!(false, resp.data[0].is_nil(), "Should not be empty");
+            }
+            Err(err) => {
+                assert!(false, "Error: {:?}", err);
+            }
+        };
+
+        let _ = db_mgr::drop_database(&tm_pool, &db_name).await;
+    }
+}