Refactor database code around transactions

- Force usage of database transactions for all operations around models.
- Add atmospheric data ingest (but no egest? yet)
- Update most dependencies

Cargo.toml

@@ -1,33 +1,31 @@
 [package]
 name = "ganymede"
-version = "2.0.0"
-authors = ["David Bourgault <[email protected]>"]
+version = "3.0.0"
+authors = ["David Bourgault <[email protected]"]
 edition = "2021"
 
 [dependencies]
-async-trait = "0.1.68"
-chrono = "0.4.26"
-chrono-tz = "0.8.2"
-env_logger = "0.10.0"
-futures = "0.3.28"
-jsonwebtoken = { version = "8.3.0", features = ["use_pem"] }
-log = "0.4.19"
-prost = "0.11.9"
-prost-types = "0.11.9"
-regex = "1.8.4"
-serde = "1.0.164"
-serde_json = "1.0.96"
-sqlx = { version = "0.6.3", features = ["runtime-tokio-rustls", "postgres", "uuid", "json"] }
-time = "0.3.26"
-tokio = { version = "1.28.2", features = ["macros", "rt-multi-thread"] }
-toml = "0.5.8"
-tonic = "0.9.2"
-tonic-reflection = "0.9.2"
-uuid = { version = "1.3.4", features = ["v4", "macro-diagnostics"] }
-
-[build-dependencies]
-tonic-build = "0.9.2"
+chrono = "0.4.38"
+chrono-tz = "0.9.0"
+clap = { version = "4.5.17", features = ["derive"] }
+env_logger = "0.11.5"
+futures-core = "0.3.30"
+jsonwebtoken = { version = "9.3.0", features = ["use_pem"] }
+log = "0.4.22"
+prost = "0.13.2"
+prost-types = "0.13.2"
+serde = "1.0.209"
+serde_json = "1.0.128"
+sqlx = { version = "0.8.2", features = ["postgres", "uuid", "json", "chrono", "runtime-tokio-rustls"] }
+tokio = { version = "1.40.0", features = ["macros", "rt-multi-thread"] }
+toml = "0.8.19"
+tonic = "0.12.2"
+tonic-reflection = "0.12.2"
+uuid = { version = "1.10.0", features = ["v4"] }
 
 [[bin]]
 name = "ganymede"
 path = "src/main.rs"
+
+[build-dependencies]
+tonic-build = "0.12.2"

api/ganymede/v2/device.proto

@@ -1,6 +1,7 @@
 syntax = "proto3";
 
 import "google/protobuf/empty.proto";
+import "google/protobuf/duration.proto";
 
 package ganymede.v2;
 
@@ -55,13 +56,17 @@ message PollRequest {
 }
 
 message PollResponse {
-    string uid = 1;
-    string display_name = 2;
+    string device_uid = 1;
+    string device_display_name = 2;
 
-    // Output only. Current device's offset from UTC (including DST if applicable)
-    int64 timezone_offset_minutes = 10;
+    string config_uid = 11;
+    string config_display_name = 12;
 
-    Config config = 20;
+    // Current device's offset from UTC (including DST if applicable)
+    int64 timezone_offset_minutes = 20;
+    google.protobuf.Duration poll_period = 21;
+
+    LightConfig light_config = 101;
 }
 
 message CreateConfigRequest {
@@ -129,7 +134,7 @@ message Luminaire {
     }
 
     uint32 port = 1;
-    bool use_pwm = 2;
+    bool active_high = 2;
 
     repeated DailySchedule photo_period = 3;
 }
@@ -144,6 +149,8 @@ message Config {
 
     string display_name = 2;
 
+    google.protobuf.Duration poll_period = 3;
+
     reserved 10, 11;
     reserved "pomp_config", "solution_config";
     LightConfig light_config = 12;

api/ganymede/v2/measurements.proto

@@ -29,13 +29,19 @@ message GetMeasurementsResponse {
 
 // Primitive message types
 message AtmosphericMeasurements {
+    float temperature = 1;       // In Celsius
+    float relative_humidity = 2; // In RH [0.0, 1.0]
+}
+
+message SoilMeasurements { 
     float temperature = 1;  // In Celsius
-    float humidity = 2;     // In RH %
+    float humidity = 2;     // In mass fraction [0.0, 1.0]
 }
 
 message Measurement {
     string device_id = 1;
-    google.protobuf.Timestamp timestamp = 2;
+    google.protobuf.Timestamp timestamp = 2; // UTC
 
     AtmosphericMeasurements atmosphere = 10;
+    SoilMeasurements soil = 11;
 }

build.rs

@@ -24,10 +24,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
             "#[derive(serde::Serialize, serde::Deserialize)] #[serde(rename_all = \"camelCase\")]",
         )
         .compile(
-            &[
-                "api/ganymede/v2/device.proto",
-                "api/ganymede/v2/measurements.proto",
-            ],
+            &["api/ganymede/v2/device.proto", "api/ganymede/v2/measurements.proto"],
             &["api/"],
         )?;
     Ok(())

docker-compose.yml

@@ -10,7 +10,7 @@ services:
       - ./Ganymede.toml:/app/Ganymede.toml
 
   postgres:
-    image: postgres:15.3
+    image: timescale/timescaledb:latest-pg15
     environment:
       POSTGRES_PASSWORD: postgres
       POSTGRES_DB: ganymede

fixtures/config.sql

@@ -0,0 +1,13 @@
+INSERT INTO config (
+    config_id,
+    domain_id,
+    display_name,
+    poll_period,
+    light_config
+) VALUES (
+    '00000000-0000-0000-0000-000000000000'::UUID,
+    '00000000-0000-0000-0000-000000000000'::UUID,
+    'Test Config',
+    '1H'::INTERVAL,
+    '{}'::JSONB
+)

fixtures/device.sql

@@ -0,0 +1,19 @@
+INSERT INTO device (
+    device_id,
+    domain_id,
+    display_name,
+    mac,
+    config_id,
+    description,
+    timezone,
+    last_poll
+) VALUES (
+    '00000000-0000-0000-0000-000000000000'::UUID,
+    '00000000-0000-0000-0000-000000000000'::UUID,
+    'Test Display',
+    '00:00:00:00:00:00',
+    '00000000-0000-0000-0000-000000000000'::UUID,
+    'This describes a device',
+    'America/Montreal',
+    NULL
+)

fixtures/domain.sql

@@ -0,0 +1,9 @@
+INSERT INTO domain(
+    domain_id,
+    name,
+    display_name
+) VALUES (
+    '00000000-0000-0000-0000-000000000000'::UUID,
+    'test-domain',
+    'Test Domain'
+)

kubernetes/ganymede.yaml

@@ -8,7 +8,7 @@ metadata:
     app.kubernetes.io/component: app
 data:
   Ganymede.toml: |
-    postgres_uri = "postgres://postgres:postgres@localhost:5432/ganymede"
+    postgres_uri = "postgresql://postgres:postgres@database:5432/ganymede"
     port = 3000
 ---
 apiVersion: apps/v1
@@ -33,7 +33,8 @@ spec:
     spec:
       containers:
         - name: ganymede
-          image: ghcr.io/ngc7293/ganymede:latest
+          image: ghcr.io/ngc7293/ganymede:dev
+          imagePullPolicy: Always
           ports:
             - containerPort: 3000
           resources:
@@ -45,7 +46,8 @@ spec:
               memory: 512M
           volumeMounts:
             - name: config-volume
-              mountPath: /app
+              mountPath: /app/Ganymede.toml
+              subPath: Ganymede.toml
       volumes:
         - name: config-volume
           configMap:

migrations/20240411230607_Add_last_poll_tracking.sql

@@ -0,0 +1 @@
+ALTER TABLE device ADD COLUMN last_poll TIMESTAMP WITH TIME ZONE;

migrations/20240412183033_Initial_timescale_schema.sql

@@ -0,0 +1,25 @@
+CREATE EXTENSION IF NOT EXISTS timescaledb;
+
+CREATE TABLE "atmosphere" (
+    observed_on       TIMESTAMP WITH TIME ZONE DEFAULT (NOW() AT TIME ZONE 'UTC') NOT NULL,
+    domain_id         UUID                                                        NOT NULL,
+    device_id         UUID                                                        NOT NULL,
+    temperature       FLOAT,
+    relative_humidity FLOAT,
+
+    FOREIGN KEY (domain_id) REFERENCES domain(domain_id) ON DELETE CASCADE,
+    FOREIGN KEY (device_id) REFERENCES device(device_id) ON DELETE CASCADE
+);
+SELECT create_hypertable('atmosphere', by_range('observed_on'));
+
+CREATE TABLE "soil" (
+    observed_on    TIMESTAMP WITH TIME ZONE DEFAULT (NOW() AT TIME ZONE 'UTC') NOT NULL,
+    domain_id      UUID                                                        NOT NULL,
+    device_id      UUID                                                        NOT NULL,
+    temperature    FLOAT,
+    humidity       FLOAT,
+
+    FOREIGN KEY (domain_id) REFERENCES domain(domain_id) ON DELETE CASCADE,
+    FOREIGN KEY (device_id) REFERENCES device(device_id) ON DELETE CASCADE
+);
+SELECT create_hypertable('soil', by_range('observed_on'));

migrations/20240829001800_Add_poll_period_to_config.sql

@@ -0,0 +1 @@
+ALTER TABLE config ADD COLUMN poll_period INTERVAL NOT NULL DEFAULT '1H';

migrations/20240907005328_Rename_device_mac_unique_constraint.sql

@@ -0,0 +1 @@
+ALTER TABLE device RENAME CONSTRAINT device_mac_key TO device_mac_unique;

rustfmt.toml

@@ -0,0 +1,2 @@
+max_width = 120
+chain_width = 120

src/database/database.rs

@@ -0,0 +1,52 @@
+use std::sync::Arc;
+
+use crate::{Error, Result};
+
+use super::transaction::DomainDatabaseTransaction;
+
+#[derive(Clone)]
+pub struct Database {
+    connection_pool: Arc<sqlx::Pool<sqlx::Postgres>>,
+}
+
+impl Database {
+    pub async fn try_from_uri(uri: &str) -> Result<Self> {
+        let pool = sqlx::postgres::PgPoolOptions::new().max_connections(5).connect(uri).await.map_err(|err| {
+            log::error!("Failed to connect to postgres: {err}");
+            Error::DatabaseError(err.to_string())
+        })?;
+
+        Ok(Database::new(pool))
+    }
+
+    pub fn new(connection_pool: sqlx::Pool<sqlx::Postgres>) -> Self {
+        Database {
+            connection_pool: Arc::new(connection_pool),
+        }
+    }
+
+    pub fn for_domain(&self, domain_id: uuid::Uuid) -> DomainDatabase {
+        DomainDatabase::new(self.connection_pool.clone(), domain_id)
+    }
+}
+
+pub struct DomainDatabase {
+    connection_pool: Arc<sqlx::Pool<sqlx::Postgres>>,
+    domain_id: uuid::Uuid,
+}
+
+impl DomainDatabase {
+    pub fn new(connection_pool: Arc<sqlx::Pool<sqlx::Postgres>>, domain_id: uuid::Uuid) -> Self {
+        DomainDatabase {
+            connection_pool,
+            domain_id,
+        }
+    }
+
+    pub async fn begin(&self) -> Result<DomainDatabaseTransaction> {
+        match self.connection_pool.begin().await {
+            Ok(tx) => Ok(DomainDatabaseTransaction::new(tx, self.domain_id.clone())),
+            Err(err) => Err(Error::DatabaseError(err.to_string())),
+        }
+    }
+}

src/database/mod.rs

@@ -0,0 +1,6 @@
+pub mod database;
+pub mod models;
+pub mod transaction;
+
+pub type Database = database::Database;
+pub type DomainDatabaseTransaction = transaction::DomainDatabaseTransaction;

src/device/device/mod.rs → src/database/models/atmosphere_data/mod.rs

@@ -1,4 +1,3 @@
-pub mod errors;
 pub mod model;
 pub mod operations;
-pub mod protobuf;
+pub mod tonic;

src/database/models/atmosphere_data/model.rs

@@ -0,0 +1,19 @@
+use chrono::{DateTime, Utc};
+use sqlx::FromRow;
+
+use crate::types::{Celsius, RelativeHumidity};
+
+#[derive(Debug, Clone, PartialEq, FromRow)]
+pub struct AtmosphereDataModel {
+    // Timestamp for data measurement
+    pub observed_on: DateTime<Utc>,
+
+    // UUID of the device that produced the measurement
+    pub device_id: uuid::Uuid,
+
+    // Air temperature
+    pub temperature: Celsius,
+
+    // Air humidity (0-1)
+    pub relative_humidity: RelativeHumidity,
+}