all lowercase

Reviewed-on: #1

.devcontainer/docker-compose.yml

@@ -8,7 +8,7 @@ services:
       - ..:/workspace:cached
     command: sleep infinity
     ports:
-      - 1234:8080
+      - 8282:8282
 
   db:
     image: postgres

.env

@@ -1,2 +1,3 @@
 DATABASE_URL=postgres://dev:dev@db/iot
 RUST_LOG=debug
+FIRMWARE_FOLDER=/temp/firmware

.gitea/workflows/build_docker.yaml

@@ -1,24 +1,21 @@
 name: Build Project
 
-on: [push]
+on:
+  push:
+    branches: ["main"]
 jobs:
   test:
     runs-on: ubuntu-latest
     container:
-      image: rust:bullseye
-    
-    # Service containers to run with `container-job`
+      image: gitea.tobiasmaier.me/tobimai/devcontainer-rust:2.2
+      options: --user root
     services:
-      # Label used to access the service container
       db:
-        # Docker Hub image
         image: postgres:latest
-        # Provide the password for postgres
         env:
           POSTGRES_USER: dev
           POSTGRES_PASSWORD: dev
           POSTGRES_DB: iot
-        # Set health checks to wait until postgres has started
         options: >-
           --health-cmd pg_isready
           --health-interval 10s
@@ -28,14 +25,10 @@ jobs:
           - 5432:5432
 
     steps:
-    - name: Install necessary dependencies
-      run: apt update && apt install nodejs pkg-config -y
-    - name: Install docker
-      run: curl -fsSL get.docker.com -o get-docker.sh && sh get-docker.sh
     - name: Checkout Code
       uses: actions/checkout@v2
     - name: Run migrations
-      run: cargo install sqlx-cli && sqlx migrate run
+      run: sqlx migrate run
     - name: Build server binary
       run: cargo build --release
     - name: Set up Docker Buildx
@@ -43,14 +36,38 @@ jobs:
     - name: Log in to Docker Registry
       uses: docker/login-action@v1 
       with:
-        registry: gitea.maiertobi.de  # Replace with your Docker registry URL
+        registry: gitea.tobiasmaier.me
         username: tobimai
         password: ${{ secrets.docker_registry_key }}
-    - name: Build and Push Docker Image
+    - name: Extract Version from Cargo.toml
+      run: |
+        VERSION_PATCH=$(grep '^version' Cargo.toml | head -n 1 | cut -d ' ' -f 3 | tr -d '"')
+        VERSION_MINOR=$(grep '^version' Cargo.toml | head -n 1 | cut -d ' ' -f 3 | tr -d '"' | cut -d '.' -f 1,2)
+        VERSION_MAJOR=$(grep '^version' Cargo.toml | head -n 1 | cut -d ' ' -f 3 | tr -d '"' | cut -d '.' -f 1)
+        echo "VERSION_PATCH=$VERSION_PATCH" >> $GITEA_ENV
+        echo "VERSION_MINOR=$VERSION_MINOR" >> $GITEA_ENV
+        echo "VERSION_MAJOR=$VERSION_MAJOR" >> $GITEA_ENV
+        echo "Patch Version: $VERSION_PATCH"
+        echo "Minor Version: $VERSION_MINOR"
+        echo "Major Version: $VERSION_MAJOR"
+    - name: Build and Push Docker Image for Patch version
       uses: docker/build-push-action@v2
       with:
         context: .
         push: true
-        tags: gitea.maiertobi.de/tobimai/iot  # Replace with your Docker image's details
+        tags: gitea.tobiasmaier.me/tobimai/iot:${{ env.VERSION_PATCH }}
+    - name: Build and Push Docker Image for minor version
+      uses: docker/build-push-action@v2
+      with:
+        context: .
+        push: true
+        tags: gitea.tobiasmaier.me/tobimai/iot:${{ env.VERSION_MINOR }}
+    - name: Build and Push Docker Image for major version
+      uses: docker/build-push-action@v2
+      with:
+        context: .
+        push: true
+        tags: gitea.tobiasmaier.me/tobimai/iot:${{ env.VERSION_MAJOR }}
+
 
 

.gitea/workflows/test_build.yaml

@@ -1,24 +1,20 @@
-name: Build Project
+name: Build and Test Project
 
-on: []
+on:
+  pull_request:
 jobs:
   test:
     runs-on: ubuntu-latest
     container:
-      image: rust:latest
-    
-    # Service containers to run with `container-job`
+      image: gitea.tobiasmaier.me/tobimai/devcontainer-rust:2.2
+      options: --user root
     services:
-      # Label used to access the service container
       db:
-        # Docker Hub image
         image: postgres:latest
-        # Provide the password for postgres
         env:
           POSTGRES_USER: dev
           POSTGRES_PASSWORD: dev
           POSTGRES_DB: iot
-        # Set health checks to wait until postgres has started
         options: >-
           --health-cmd pg_isready
           --health-interval 10s
@@ -28,11 +24,12 @@ jobs:
           - 5432:5432
 
     steps:
-    - run: apt update && apt install nodejs pkgconfig -y
     - name: Checkout Code
       uses: actions/checkout@v2
     - name: Run migrations
-      run: cargo install sqlx-cli && sqlx migrate run
+      run: sqlx migrate run
     - name: Running cargo build
       run: cargo build --release
+    - name: Running cargo build
+      run: cargo test
 

.gitignore

@@ -19,3 +19,5 @@ Cargo.lock
 # Added by cargo
 
 /target
+/fw
+/firmware

Cargo.toml

@@ -1,17 +1,25 @@
 [package]
 name = "iot-cloud"
-version = "0.1.0"
+version = "0.2.1"
 edition = "2021"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-actix-web = "4.4.0"
-chrono = { version = "0.4.31", features = ["serde"] }
+actix-service = "2.0"
+actix-web = "4.11"
+chrono = { version = "0.4", features = ["serde"] }
 dotenvy = "0.15"
 env_logger = "0.11"
-log = "0.4.20"
+log = "0.4"
+semver = "1.0.25"
 serde = "1.0.188"
 sqlx = { version = "0.8", features = [ "runtime-tokio", "tls-rustls", "postgres", "migrate", "chrono", "mac_address"] }
 sqlx-cli = "0.8"
-thiserror = "1.0"
+strum = { version = "0.27", features = ["derive"] }
+thiserror = "2.0"
+tokio = { version = "1", features = ["fs", "rt-multi-thread"] }
+
+
+[lints.clippy]
+pedantic = "warn"

Dockerfile

@@ -1,4 +1,4 @@
-FROM debian:bullseye
+FROM debian:bookworm
 ADD target/release/iot-cloud /iot-cloud
 RUN apt update && apt upgrade -y && apt install openssl -y
 

README.md

@@ -1,2 +1,5 @@
 # iot-cloud
 
+## TODO
+- Env var for fw dir
+- basic auth for uploading

dev/IoT-API/GET firmwares.bru

@@ -0,0 +1,15 @@
+meta {
+  name: GET firmwares
+  type: http
+  seq: 4
+}
+
+get {
+  url: http://localhost:8282/firmware/waterlevel
+  body: multipartForm
+  auth: none
+}
+
+body:multipart-form {
+  : @file(/home/tobi/git/iot-cloud-api/target/debug/iot-cloud)
+}

dev/IoT-API/Get Devices.bru

@@ -0,0 +1,11 @@
+meta {
+  name: Get Devices
+  type: http
+  seq: 2
+}
+
+get {
+  url: http://localhost:8282/device
+  body: none
+  auth: none
+}

dev/IoT-API/PUT firmware.bru

@@ -0,0 +1,15 @@
+meta {
+  name: PUT firmware
+  type: http
+  seq: 3
+}
+
+put {
+  url: http://localhost:8282/firmware/waterlevel/INA233/1.0.1
+  body: multipartForm
+  auth: none
+}
+
+body:multipart-form {
+  : @file(/home/tobi/git/waterlevel-software/.pio/build/ESP32/firmware.bin)
+}

dev/IoT-API/bruno.json

@@ -0,0 +1,9 @@
+{
+  "version": "1",
+  "name": "IoT-API",
+  "type": "collection",
+  "ignore": [
+    "node_modules",
+    ".git"
+  ]
+}

src/database.rs

@@ -1,10 +1,15 @@
 use actix_web::web;
 use chrono::Utc;
 use log::{error, info};
-use sqlx::{migrate, postgres::PgPoolOptions, query, query_as, types::mac_address::MacAddress, Pool, Postgres};
+use sqlx::{
+    migrate, postgres::PgPoolOptions, query, query_as, types::mac_address::MacAddress, Pool,
+    Postgres,
+};
 use thiserror::Error;
 
-use crate::schemas::{TelemetryMessage, TelemetryMessageFromDevice, ValueMessageFromDevice, ValueMessage, Device};
+use crate::schemas::{
+    Device, TelemetryMessage, TelemetryMessageFromDevice, ValueMessage, ValueMessageFromDevice,
+};
 
 #[derive(Clone)]
 pub struct Database {
@@ -29,13 +34,14 @@ impl Database {
                 Database { conn_pool: pool }
             }
             Err(err) => {
-                error!("Failed to connect to the database: {:?}", err);
+                error!("Failed to connect to the database: {err:?}");
                 std::process::exit(1);
             }
         }
     }
 
-    pub async fn init_from_pool(pool: Pool<Postgres>) -> Database {
+    #[cfg(test)]
+    pub fn init_from_pool(pool: Pool<Postgres>) -> Database {
         Database { conn_pool: pool }
     }
 
@@ -43,9 +49,9 @@ impl Database {
     pub async fn init_db(&self) {
         info!("Checking if required tables exist");
         match migrate!().run(&self.conn_pool).await {
-            Ok(_) => {}
+            Ok(()) => {}
             Err(e) => {
-                error!("Error when running migrations {}", e);
+                error!("Error when running migrations {e}");
                 std::process::exit(1);
             }
         };
@@ -59,15 +65,27 @@ impl Database {
         Ok(())
     }
 
-    pub async fn add_display_name(&self, device_id: &MacAddress, display_name: &str) -> Result<(), DatabaseError> {
-        info!("Adding Displayname {display_name} to Device with ID {device_id}");
-        query!("UPDATE Devices SET display_name = $1 WHERE id = $2;", display_name, device_id)
-            .execute(&self.conn_pool)
-            .await?;
+     #[allow(dead_code)]
+    pub async fn update_display_name(
+        &self,
+        device_id: &MacAddress,
+        display_name: &str,
+    ) -> Result<(), DatabaseError> {
+        info!("Updating Displayname to {display_name} for Device with ID {device_id}");
+        query!(
+            "UPDATE Devices SET display_name = $1 WHERE id = $2;",
+            display_name,
+            device_id
+        )
+        .execute(&self.conn_pool)
+        .await?;
         Ok(())
     }
 
-    pub async fn create_device_if_not_exists(&self, device_id: &MacAddress) -> Result<(), DatabaseError> {
+    pub async fn create_device_if_not_exists(
+        &self,
+        device_id: &MacAddress,
+    ) -> Result<(), DatabaseError> {
         info!("Checking if device with the ID {} exists", &device_id);
         let exists_result = query!("SELECT count(*) FROM devices WHERE ID = $1;", device_id)
             .fetch_one(&self.conn_pool)
@@ -76,7 +94,7 @@ impl Database {
         let exists = match exists_result {
             Ok(res) => res.count > Some(0),
             Err(err) => {
-                error!("Error checking table existence: {:?}", err);
+                error!("Error checking table existence: {err:?}");
                 std::process::exit(1);
             }
         };
@@ -125,20 +143,31 @@ impl Database {
 
     pub async fn add_value(
         &self,
-        msg: &web::Json<ValueMessageFromDevice>,
+        msg: &ValueMessageFromDevice,
         device_id: &MacAddress,
     ) -> Result<(), DatabaseError> {
         info!("Adding value to DB");
         let current_timestamp = Utc::now().naive_utc();
-        query!("
+        query!(
+            "
         INSERT INTO values (timestamp, value, device_id, active_errors, value_id)
         VALUES ($1, $2, $3, $4, $5);",
-                current_timestamp, msg.value, device_id, msg.active_errors, msg.value_id).execute(&self.conn_pool).await?;
+            current_timestamp,
+            msg.value,
+            device_id,
+            msg.active_errors,
+            msg.value_id
+        )
+        .execute(&self.conn_pool)
+        .await?;
 
         Ok(())
     }
 
-    pub async fn get_values_for_id(&self, device_id: &MacAddress) -> Result<Vec<ValueMessage>, DatabaseError> {
+    pub async fn get_values_for_id(
+        &self,
+        device_id: &MacAddress,
+    ) -> Result<Vec<ValueMessage>, DatabaseError> {
         info!("Getting values for {} from DB", &device_id);
         let values = query_as!(
             ValueMessage,
@@ -170,23 +199,46 @@ impl Database {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use dotenvy::dotenv;
     use sqlx::PgPool;
 
     #[sqlx::test]
-    async fn add_device(pool: PgPool) {
-        dotenv().ok();
-        let db = Database::init_from_pool(pool).await;
+    async fn add_device_and_display_name(pool: PgPool) {
+        let db = Database::init_from_pool(pool);
 
-        let test_device = Device{
+        let test_device = Device {
             display_name: Some("Waterlevel daheim".to_owned()),
-            id: MacAddress::from([0x1A, 0x2B, 0x3C, 0x4D, 0x5E, 0x6F])
+            id: MacAddress::from([0x1A, 0x2B, 0x3C, 0x4D, 0x5E, 0x6F]),
         };
-        db.add_device(&MacAddress::from([0x1A, 0x2B, 0x3C, 0x4D, 0x5E, 0x6F])).await.unwrap();
-        db.add_display_name(&MacAddress::from([0x1A, 0x2B, 0x3C, 0x4D, 0x5E, 0x6F]), "Waterlevel daheim").await.unwrap();
+        db.add_device(&MacAddress::from([0x1A, 0x2B, 0x3C, 0x4D, 0x5E, 0x6F]))
+            .await
+            .unwrap();
+        db.update_display_name(
+            &MacAddress::from([0x1A, 0x2B, 0x3C, 0x4D, 0x5E, 0x6F]),
+            "Waterlevel daheim",
+        )
+        .await
+        .unwrap();
 
         let devices = db.get_devices().await.unwrap();
         assert_eq!(test_device, devices[0]);
+    }
 
+    #[sqlx::test]
+    async fn add_value(pool: PgPool) {
+        let db = Database::init_from_pool(pool);
+        let device_id = MacAddress::from([0x1A, 0x2B, 0x3C, 0x4D, 0x5E, 0x6F]);
+
+        db.add_device(&device_id).await.unwrap();
+        let msg = ValueMessageFromDevice {
+            active_errors: 0,
+            value: 112.0,
+            value_id: 1,
+        };
+        db.add_value(&msg, &device_id).await.unwrap();
+
+        let values = db.get_values_for_id(&device_id).await.unwrap();
+
+        assert!((values[0].value - msg.value).abs() < 1e-5);
+        assert_eq!(values[0].value_id, msg.value_id);
     }
 }

src/device_telemetry_api.rs

@@ -0,0 +1,137 @@
+use actix_web::{get, post, put, web, HttpResponse, Responder};
+use log::{error, info};
+use sqlx::types::mac_address::MacAddress;
+
+use crate::{
+    schemas::{AppState, DeviceMetadata, TelemetryMessageFromDevice, ValueMessageFromDevice},
+    util::parse_mac_address,
+};
+
+#[post("/telemetry/{device_id}")]
+async fn receive_telemetry(
+    device_id: web::Path<String>,
+    data: web::Data<AppState>,
+    telemetry_message: web::Json<TelemetryMessageFromDevice>,
+) -> impl Responder {
+    info!("POST - telementry - Processing device id {device_id}");
+    let Ok(mac_converted) = parse_mac_address(&device_id) else {
+        return HttpResponse::InternalServerError();
+    };
+    let mac_converted = MacAddress::from(mac_converted);
+    match data.db.create_device_if_not_exists(&mac_converted).await {
+        Ok(()) => {}
+        Err(e) => {
+            error!("Error creating new device: {e}");
+            return HttpResponse::InternalServerError();
+        }
+    }
+
+    match data
+        .db
+        .add_telemetry(&telemetry_message, &mac_converted)
+        .await
+    {
+        Ok(()) => HttpResponse::Created(),
+        Err(e) => {
+            error!("adding Telemetry message to DB failed \n{e}");
+            HttpResponse::InternalServerError()
+        }
+    }
+}
+
+#[get("/telemetry/{device_id}")]
+async fn get_telemetry(device_id: web::Path<String>, data: web::Data<AppState>) -> impl Responder {
+    info!("GET - telementry - Processing device id {device_id}");
+    let Ok(mac_converted) = parse_mac_address(&device_id) else {
+        return HttpResponse::InternalServerError().finish();
+    };
+    let mac_converted = MacAddress::from(mac_converted);
+    let messages = match data.db.get_telemetry_for_id(&mac_converted).await {
+        Ok(msgs) => msgs,
+        Err(e) => {
+            error!("Getting Telemetry Messages from DB failed \n{e}");
+            return HttpResponse::InternalServerError().finish();
+        }
+    };
+    HttpResponse::Ok().json(messages)
+}
+
+#[post("/value/{device_id}")]
+async fn receive_value(
+    device_id: web::Path<String>,
+    data: web::Data<AppState>,
+    value_message: web::Json<ValueMessageFromDevice>,
+) -> impl Responder {
+    info!("POST - value - Processing device id {device_id}");
+    let Ok(mac_converted) = parse_mac_address(&device_id) else {
+        return HttpResponse::InternalServerError();
+    };
+    let mac_converted = MacAddress::from(mac_converted);
+    match data.db.create_device_if_not_exists(&mac_converted).await {
+        Ok(()) => {}
+        Err(e) => {
+            error!("Error creating new device: {e}");
+            return HttpResponse::InternalServerError();
+        }
+    }
+
+    match data.db.add_value(&value_message, &mac_converted).await {
+        Ok(()) => HttpResponse::Created(),
+        Err(e) => {
+            error!("adding Telemetry message to DB failed \n{e}");
+            HttpResponse::InternalServerError()
+        }
+    }
+}
+
+#[get("/value/{device_id}")]
+async fn get_value(device_id: web::Path<String>, data: web::Data<AppState>) -> impl Responder {
+    info!("GET - value - Processing device id {device_id}");
+    let Ok(mac_converted) = parse_mac_address(&device_id) else {
+        return HttpResponse::InternalServerError().finish();
+    };
+    let mac_converted = MacAddress::from(mac_converted);
+    let messages = match data.db.get_values_for_id(&mac_converted).await {
+        Ok(msgs) => msgs,
+        Err(e) => {
+            error!("Getting Values from DB failed \n{e}");
+            return HttpResponse::InternalServerError().finish();
+        }
+    };
+    HttpResponse::Ok().json(messages)
+}
+
+#[get("/device")]
+async fn get_devices(data: web::Data<AppState>) -> impl Responder {
+    info!("GET - devices - Processing");
+    let devices = match data.db.get_devices().await {
+        Ok(devs) => devs,
+        Err(e) => {
+            error!("Getting Devices from DB failed \n{e}");
+            return HttpResponse::InternalServerError().finish();
+        }
+    };
+    HttpResponse::Ok().json(devices)
+}
+
+#[put("/device/{device_id}/display_name")]
+async fn set_display_name(
+    data: web::Data<AppState>,
+    device_id: web::Path<String>,
+    device_data: web::Json<DeviceMetadata>,
+) -> impl Responder {
+    info!("PUT - device/{device_id} - Display Name",);
+    let display_name = device_data.0.display_name;
+    let db = &data.db;
+    let Ok(mac_converted) = parse_mac_address(&device_id) else {
+        error!("Failed to convert mac address");
+        return HttpResponse::InternalServerError().finish();
+    };
+    let mac_converted = MacAddress::from(mac_converted);
+    if let Ok(()) = db.update_display_name(&mac_converted, &display_name).await {
+        HttpResponse::Ok().finish()
+    } else {
+        error!("Failed to update/set displayName for device");
+        HttpResponse::InternalServerError().finish()
+    }
+}

src/firmware_api.rs

@@ -0,0 +1,125 @@
+use std::{fs, path::PathBuf};
+
+use actix_web::{get, put, web, HttpResponse, Responder};
+use log::{debug, info, warn};
+use semver::Version;
+use std::str::FromStr;
+
+use crate::{
+    schemas::{AppState, OTAConfigurationList, Services},
+    util::{get_files, prune_files},
+};
+
+// Upload Firmware file
+#[put("/{service}/{device}/{config}/{version}")]
+async fn upload_firmware(
+    data: web::Data<AppState>,
+    path: web::Path<(String, String, String, String)>,
+    body: web::Bytes,
+) -> impl Responder {
+    let (service, device, config, version) = path.into_inner();
+    let Ok(version) = Version::parse(&version) else {
+        return HttpResponse::InternalServerError().body("Failed to parse version");
+    };
+    debug!("{version:?}");
+    let version = format!("{}-{}-{}", version.major, version.minor, version.patch);
+
+    let firmware_root_path = &data.firmwares_path;
+    let Ok(service) = Services::from_str(&service) else {
+        return HttpResponse::NotFound().finish();
+    };
+
+    let firmware_folder = match service {
+        Services::Firmware => firmware_root_path.join(&device),
+        Services::Filesystem => firmware_root_path.join(format!("{}_filesystem", &device))
+    };
+    let firmware_path = firmware_folder
+        .join(format!("{service}_{}_{version}", config.to_lowercase()))
+        .with_extension("bin");
+
+    if firmware_path.is_file() {
+        warn!("{service} with product: {device}, config: {config} and version: {version} at path {} already exists, cant upload", firmware_path.display());
+        return HttpResponse::Conflict().body(format!("{}", firmware_path.display()));
+    }
+
+    info!("Uploading {service} with product: {device}, config: {config} and version: {version} to {}", firmware_path.display());
+
+    fs::create_dir_all(&firmware_folder).unwrap();
+    let x = tokio::fs::write(&firmware_path, &body).await;
+    debug!("{x:?}");
+
+    debug!("pruning now");
+    prune_files(&firmware_folder, &service, 3);
+
+    HttpResponse::Ok().body(format!("Firmware version {version} uploaded successfully"))
+
+}
+
+#[get("/{service}/{device}")]
+async fn get_firmware_json(
+    data: web::Data<AppState>,
+    path: web::Path<(String, String)>,
+) -> impl Responder {
+    let (service, device) = path.into_inner();
+    let Ok(firmware_or_filesystem) = Services::from_str(&service) else {
+        return HttpResponse::NotFound().finish();
+    };
+
+    let fw_path = match firmware_or_filesystem {
+        Services::Firmware => &data.firmwares_path.join(device),
+        Services::Filesystem => &data.firmwares_path.join(format!("{device}_filesystem"))
+    };
+
+    if fw_path.is_dir() {
+        match get_files(fw_path, &data.hostname) {
+            Ok(cfg) => HttpResponse::Ok().json(OTAConfigurationList {
+                configurations: cfg,
+            }),
+            Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
+        }
+    } else {
+        HttpResponse::Ok().json(OTAConfigurationList {
+            configurations: vec![],
+        })
+    }
+}
+
+#[get("/{service}/{product}/{config}/{version}.bin")]
+async fn serve_firmware(
+    path: web::Path<(String, String, String, String)>,
+    data: web::Data<AppState>,
+) -> impl Responder {
+    let (service, product, config, version) = path.into_inner();
+    let version = version.replace(['.', '_'], "-");
+    let service = service.to_lowercase();
+    let config = config.to_lowercase();
+    let Ok(service) = Services::from_str(&service) else {
+        return HttpResponse::NotFound().finish();
+    };
+    let fw_root_path = &data.firmwares_path;
+
+    
+    let file_path = match service {
+        Services::Firmware => PathBuf::from(format!("{product}/firmware_{config}_{version}.bin")),
+        Services::Filesystem => PathBuf::from(format!("{product}/filesystem_{config}_{version}.bin")),
+    };
+
+    let file_path = fw_root_path.join(&file_path);
+    
+
+    info!("Requested firmware for product: {product}, config: {config} and version: {version}, expected to be stored at {}", file_path.display());
+
+    if file_path.exists() {
+        info!("File exists, serving download now");
+        HttpResponse::Ok()
+            .content_type("application/octet-stream")
+            .insert_header((
+                "Content-Disposition",
+                format!("attachment; filename=\"{product}_{config}_{version}.bin\""),
+            ))
+            .body(std::fs::read(file_path).unwrap_or_else(|_| Vec::new()))
+    } else {
+        warn!("File does not exist");
+        HttpResponse::NotFound().body("Firmware version not found")
+    }
+}

src/main.rs

@@ -1,129 +1,23 @@
-use std::{env, process};
+use std::{env, path::PathBuf, process};
 
-use crate::schemas::{TelemetryMessageFromDevice, ValueMessageFromDevice};
-use actix_web::{get, post, web, App, HttpResponse, HttpServer, Responder};
+use actix_web::{web, App, HttpServer};
 use database::Database;
+use dotenvy::dotenv;
 use log::{error, info};
-use sqlx::types::mac_address::MacAddress;
-use util::parse_mac_address;
+use schemas::AppState;
 
 mod database;
+mod device_telemetry_api;
+mod firmware_api;
 mod schemas;
 mod util;
 
-struct AppState {
-    db: Database,
-}
-
-#[post("/telemetry/{device_id}")]
-async fn receive_telemetry(
-    device_id: web::Path<String>,
-    data: web::Data<AppState>,
-    telemetry_message: web::Json<TelemetryMessageFromDevice>,
-) -> impl Responder {
-    info!("POST - telementry - Processing device id {}", device_id);
-    let Ok(mac_converted) = parse_mac_address(&device_id) else {
-        return HttpResponse::InternalServerError();
-    };
-    let mac_converted = MacAddress::from(mac_converted);
-    match data.db.create_device_if_not_exists(&mac_converted).await {
-        Ok(_) => {}
-        Err(e) => {
-            error!("Error creating new device: {}", e);
-            return HttpResponse::InternalServerError();
-        }
-    };
-
-    match data.db.add_telemetry(&telemetry_message, &mac_converted).await {
-        Ok(_) => HttpResponse::Created(),
-        Err(e) => {
-            error!("adding Telemetry message to DB failed \n{}", e);
-            HttpResponse::InternalServerError()
-        }
-    }
-}
-
-#[get("/telemetry/{device_id}")]
-async fn get_telemetry(device_id: web::Path<String>, data: web::Data<AppState>) -> impl Responder {
-    info!("GET - telementry - Processing device id {}", device_id);
-    let Ok(mac_converted) = parse_mac_address(&device_id) else {
-        return HttpResponse::InternalServerError().finish();
-    };
-    let mac_converted = MacAddress::from(mac_converted);
-    let messages = match data.db.get_telemetry_for_id(&mac_converted).await {
-        Ok(msgs) => msgs,
-        Err(e) => {
-            error!("Getting Telemetry Messages from DB failed \n{}", e);
-            return HttpResponse::InternalServerError().finish();
-        }
-    };
-    HttpResponse::Ok().json(messages)
-}
-
-#[post("/value/{device_id}")]
-async fn receive_value(
-    device_id: web::Path<String>,
-    data: web::Data<AppState>,
-    value_message: web::Json<ValueMessageFromDevice>,
-) -> impl Responder {
-    info!("POST - value - Processing device id {}", device_id);
-    let Ok(mac_converted) = parse_mac_address(&device_id) else {
-        return HttpResponse::InternalServerError();
-    };
-    let mac_converted = MacAddress::from(mac_converted);
-    match data.db.create_device_if_not_exists(&mac_converted).await {
-        Ok(_) => {}
-        Err(e) => {
-            error!("Error creating new device: {}", e);
-            return HttpResponse::InternalServerError();
-        }
-    };
-
-    match data.db.add_value(&value_message, &mac_converted).await {
-        Ok(_) => HttpResponse::Created(),
-        Err(e) => {
-            error!("adding Telemetry message to DB failed \n{}", e);
-            HttpResponse::InternalServerError()
-        }
-    }
-}
-
-#[get("/value/{device_id}")]
-async fn get_value(device_id: web::Path<String>, data: web::Data<AppState>) -> impl Responder {
-    info!("GET - value - Processing device id {}", device_id);
-    let Ok(mac_converted) = parse_mac_address(&device_id) else {
-        return HttpResponse::InternalServerError().finish();
-    };
-    let mac_converted = MacAddress::from(mac_converted);
-    let messages = match data.db.get_values_for_id(&mac_converted).await {
-        Ok(msgs) => msgs,
-        Err(e) => {
-            error!("Getting Values from DB failed \n{}", e);
-            return HttpResponse::InternalServerError().finish();
-        }
-    };
-    HttpResponse::Ok().json(messages)
-}
-
-#[get("/device/")]
-async fn get_devices(data: web::Data<AppState>) -> impl Responder {
-    info!("GET - devices - Processing");
-    let devices = match data.db.get_devices().await {
-        Ok(devs) => devs,
-        Err(e) => {
-            error!("Getting Devices from DB failed \n{}", e);
-            return HttpResponse::InternalServerError().finish();
-        }
-    };
-    HttpResponse::Ok().json(devices)
-}
-
 #[actix_web::main]
 async fn main() -> std::io::Result<()> {
+    dotenv().ok();
     env_logger::init();
     info!("Starting");
 
-    
     let db_url = match env::var("DATABASE_URL") {
         Ok(url) => url,
         Err(e) => {
@@ -131,21 +25,34 @@ async fn main() -> std::io::Result<()> {
             process::exit(1);
         }
     };
-    info!("Connecting to Database {}", db_url);
+    let external_url = env::var("URL").unwrap_or("localhost".to_string());
+    let firmware_path = env::var("FIRMWARE_PATH").unwrap_or("/firmware".to_string());
+
+    info!("External URL: {external_url}");
+    info!("Firmware path set to: {firmware_path}");
+    info!("Connecting to Database {db_url}");
 
     let db = Database::init(&db_url).await;
     db.init_db().await;
 
     HttpServer::new(move || {
         App::new()
-            .app_data(web::Data::new(AppState { db: db.clone() }))
-            .service(receive_telemetry)
-            .service(get_telemetry)
-            .service(receive_value)
-            .service(get_value)
-            .service(get_devices)
+            .app_data(web::Data::new(AppState {
+                db: db.clone(),
+                firmwares_path: PathBuf::from(firmware_path.clone()),
+                hostname: external_url.clone(),
+            }))
+            .app_data(web::PayloadConfig::new(2 * 1024 * 1024 * 1024)) //1GB
+            .service(device_telemetry_api::receive_telemetry)
+            .service(device_telemetry_api::get_telemetry)
+            .service(device_telemetry_api::receive_value)
+            .service(device_telemetry_api::get_value)
+            .service(device_telemetry_api::get_devices)
+            .service(firmware_api::upload_firmware)
+            .service(firmware_api::get_firmware_json)
+            .service(firmware_api::serve_firmware)
     })
-    .bind(("0.0.0.0", 8080))?
+    .bind(("0.0.0.0", 8282))?
     .run()
     .await
 }

src/schemas.rs

@@ -1,6 +1,12 @@
+use std::path::PathBuf;
+
 use chrono::NaiveDateTime;
+use semver::Version;
 use serde::{ser::SerializeStruct, Deserialize, Serialize};
 use sqlx::types::mac_address::MacAddress;
+use strum::{Display, EnumString};
+
+use crate::database::Database;
 
 #[derive(Deserialize, Debug, Serialize)]
 pub struct TelemetryMessage {
@@ -19,14 +25,14 @@ pub struct TelemetryMessageFromDevice {
     pub software_version: i32,
 }
 
-#[derive(Deserialize, Debug, Serialize)]
+#[derive(Deserialize, Debug, Serialize, PartialEq)]
 pub struct ValueMessageFromDevice {
     pub value: f64,
     pub value_id: i32,
     pub active_errors: i32,
 }
 
-#[derive(Deserialize, Debug, Serialize)]
+#[derive(Deserialize, Debug, Serialize, PartialEq)]
 pub struct ValueMessage {
     pub value: f64,
     pub value_id: i32,
@@ -36,10 +42,9 @@ pub struct ValueMessage {
 #[derive(Debug, PartialEq)]
 pub struct Device {
     pub display_name: Option<String>,
-    pub id: MacAddress
+    pub id: MacAddress,
 }
 
-
 impl Serialize for Device {
     fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
     where
@@ -50,9 +55,67 @@ impl Serialize for Device {
         // Serialize each field with custom logic
         let bytes = self.id.bytes();
         state.serialize_field("display_name", &self.display_name)?;
-        state.serialize_field("id", &format!("{}{}{}{}{}{}", bytes[0], bytes[1], bytes[2], bytes[3], bytes[4], bytes[5]))?;
+        state.serialize_field(
+            "id",
+            &format!(
+                "{}{}{}{}{}{}",
+                bytes[0], bytes[1], bytes[2], bytes[3], bytes[4], bytes[5]
+            ),
+        )?;
 
         // End the serialization process
         state.end()
     }
 }
+
+#[derive(serde::Serialize)]
+#[serde(rename_all = "PascalCase")]
+pub struct OTAConfigurationList {
+    pub configurations: Vec<OTAConfiguration>,
+}
+
+#[derive(serde::Serialize, PartialEq, Debug)]
+#[serde(rename_all = "PascalCase")]
+pub struct OTAConfiguration {
+    pub version: Version,
+    #[serde(rename = "URL")]
+    pub url: String,
+    pub board: Option<BoardType>,
+    pub config: Option<BoardConfig>,
+}
+
+#[derive(serde::Serialize, EnumString, PartialEq, Debug, Display)]
+#[strum(serialize_all = "snake_case")]
+pub enum BoardType {
+    Waterlevel,
+    WaterlevelFilesystem,
+}
+
+#[derive(serde::Serialize, EnumString, PartialEq, Debug, Display)]
+#[strum(ascii_case_insensitive, serialize_all = "SCREAMING_SNAKE_CASE")]
+pub enum BoardConfig {
+    INA226,
+    INA233,
+    INA226REV2,
+    INA233REV2,
+    Generic,
+}
+
+pub struct AppState {
+    pub db: Database,
+    pub firmwares_path: PathBuf,
+    pub hostname: String,
+}
+
+#[allow(dead_code)]
+#[derive(Serialize, Deserialize, Debug)]
+pub struct DeviceMetadata {
+    pub display_name: String,
+}
+
+#[derive(EnumString, PartialEq, Debug, Display, Serialize)]
+#[strum(ascii_case_insensitive, serialize_all = "snake_case")]
+pub enum Services {
+    Firmware,
+    Filesystem,
+}

src/util.rs

@@ -1,12 +1,15 @@
+use std::{fs, path::PathBuf};
 
-
+use log::{debug, error, info};
+use semver::Version;
+use std::str::FromStr;
 use thiserror::Error;
 
-
+use crate::schemas::{BoardConfig, BoardType, OTAConfiguration, Services};
 
 pub fn parse_mac_address(mac: &str) -> Result<[u8; 6], MacAddressError> {
     if mac.len() != 12 {
-        return Err(MacAddressError::Length(mac.len()))
+        return Err(MacAddressError::Length(mac.len()));
     }
 
     let mut mac_bytes = [0u8; 6];
@@ -26,10 +29,122 @@ pub enum MacAddressError {
     Length(usize),
 }
 
+#[derive(Error, Debug)]
+pub enum GetFilesError {
+    #[error("IO Error while reading files")]
+    IO(#[from] std::io::Error),
+    #[error("Error getting filename")]
+    Filename,
+    #[error("Error getting extension (.bin)")]
+    Extension,
+    #[error("Strum parse Error")]
+    Conversion(#[from] strum::ParseError),
+    #[error("Failed to parse the version")]
+    VersionParse(#[from] semver::Error),
+}
+
+pub fn get_files(
+    root_path: &PathBuf,
+    hostname: &str,
+) -> Result<Vec<OTAConfiguration>, GetFilesError> {
+    info!("Getting all files from path {}", root_path.display());
+    let mut configs = Vec::new();
+    let product_name = root_path
+        .file_name()
+        .ok_or(GetFilesError::Filename)?
+        .to_string_lossy();
+    let entries = fs::read_dir(root_path)?;
+
+    for entry in entries.flatten() {
+        info!("Reading entry: {entry:?}");
+        let path = entry.path();
+        if path.is_file() {
+            info!("processing file: {}", path.display());
+            // Splits the filename at the underscores. This is safe to do as names get sanitized on upload and are only uploaded by the pipeline
+            let split_name: Vec<_> = path
+                .file_name()
+                .ok_or(GetFilesError::Filename)?
+                .to_str()
+                .ok_or(GetFilesError::Filename)?
+                .split('_')
+                .collect();
+            let version = split_name[2]
+                .strip_suffix(".bin")
+                .ok_or(GetFilesError::Extension)?
+                .replace('-', ".");
+            debug!("Version: {:?}", version);
+            debug!("split_name: {:?}", split_name);
+            // TODO this is kinda messy
+            let board_config = BoardConfig::from_str(split_name[1])?;
+            let service = split_name[0];
+            let board_type = BoardType::from_str(&product_name).unwrap();
+            let version = if version.contains('.') {
+                Version::parse(&version)?
+            } else {
+                // Handle simple version number by adding .0.0
+                Version::parse(&format!("{}.0.0", version))?
+            };
+            let version_replaced = format!("{}-{}-{}", version.major, version.minor, version.patch);
+            let fw_url =
+                format!("{hostname}/{service}/{board_type}/{board_config}/{version_replaced}.bin");
+            let cfg = OTAConfiguration {
+                version,
+                url: fw_url,
+                board: Some(board_type),
+                config: Some(board_config),
+            };
+            configs.push(cfg);
+        } else if path.is_dir() {
+            println!("Directory: {}", path.display());
+        }
+    }
+
+    Ok(configs)
+}
+
+pub fn prune_files(path: &PathBuf, service: &Services, keep_last: usize) {
+    let Ok(mut config_list) = get_files(path, "irrelevant") else {
+        error!("failed to get file list for pruning");
+        return
+    };
+    config_list.sort_by_key(|x| x.version.clone());
+    config_list.truncate(config_list.len() - keep_last);
+    for cfg in config_list {
+        println!("{cfg:?}");
+        let path = path.to_string_lossy();
+        let board_type = cfg. config.unwrap();
+        let path_to_remove = format!("{path}/{service}_{}_{}-{}-{}.bin", board_type, cfg.version.major, cfg.version.minor, cfg.version.patch);
+        if let Err(e) = fs::remove_file(&path_to_remove) {
+            error!("Failed to delete {path_to_remove}, {e:?}");
+            return;
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
 
+    #[test]
+    fn test_file_loading() {
+        let expected_1 = OTAConfiguration {
+            version: Version::parse("1.2.3").unwrap(),
+            url: "example.com/firmware/waterlevel/INA233/1-2-3.bin".to_string(),
+            board: Some(BoardType::Waterlevel),
+            config: Some(BoardConfig::INA233),
+        };
+        let expected_2 = OTAConfiguration {
+            version: Version::parse("4.5.6").unwrap(),
+            url: "example.com/firmware/waterlevel/INA226/4-5-6.bin".to_string(),
+            board: Some(BoardType::Waterlevel),
+            config: Some(BoardConfig::INA226),
+        };
+        let loaded_configs = get_files(&PathBuf::from("./test/waterlevel"), "example.com").unwrap();
+
+        assert_eq!(loaded_configs[1], expected_1);
+        assert_eq!(loaded_configs[0], expected_2);
+    }
+
     #[test]
     fn test_valid_mac_address_plain() {
         let mac_str = "001A2B3C4D5E";
@@ -68,4 +183,3 @@ mod tests {
         assert!(parse_mac_address(mac_str).is_err());
     }
 }
-