feature(pedm): SQL initialization

- adds version tracking in the `pedm_schema_version` table
- adds table initialization
  - error checking is strict and initialization only happens if the
    version table is not found
- adds SQLite error handling for microsecond timestamp conversion to/from
  Chrono
- fixes a bug in the libSQL schema where datetimes were stored
  incorrectly
  - they were being stored fractionally, instead of the full
    microseconds since epoch
- adds `/about` endpoint to get basic info about the running application
  - I used this to ensure that the libsql date handling is functional

A few other changes are included:
- libsql feature set is changed (default is now false, "stream" removed)

Author: allan2

Cargo.lock

@@ -11,38 +11,58 @@ publish = false
 
 [dependencies]
 anyhow = "1.0"
-axum = { version = "0.8", default-features = false, features = ["http1", "json", "tokio", "query", "tracing", "tower-log", "form", "original-uri", "matched-path"] }
+axum = { version = "0.8", default-features = false, features = [
+  "http1",
+  "json",
+  "tokio",
+  "query",
+  "tracing",
+  "tower-log",
+  "form",
+  "original-uri",
+  "matched-path",
+] }
 base16ct = { version = "0.2", features = ["std", "alloc"] }
 base64 = "0.22"
+chrono = { version = "0.4", features = ["serde"] }
 digest = "0.10"
 hyper = { version = "1.3", features = ["server"] }
 hyper-util = { version = "0.1", features = ["tokio"] }
-schemars = "0.8"
+schemars = { version = "0.8", features = ["chrono"] }
 serde = "1.0"
 serde_json = "1.0"
 sha1 = "0.10"
 sha2 = "0.10"
 tokio = { version = "1.44", features = ["net", "rt-multi-thread"] }
 tower-service = "0.3"
 win-api-wrappers = { path = "../win-api-wrappers" }
-devolutions-pedm-shared = { path = "../devolutions-pedm-shared", features = ["policy"]}
+devolutions-pedm-shared = { path = "../devolutions-pedm-shared", features = [
+  "policy",
+] }
 devolutions-gateway-task = { path = "../devolutions-gateway-task" }
 devolutions-agent-shared = { path = "../devolutions-agent-shared" }
 camino = { version = "1", features = ["serde1"] }
 async-trait = "0.1"
 tracing = "0.1"
-aide = { version = "0.14", features = ["axum", "axum-extra", "axum-json", "axum-tokio"] }
+aide = { version = "0.14", features = [
+  "axum",
+  "axum-extra",
+  "axum-json",
+  "axum-tokio",
+] }
 tower-http = { version = "0.5", features = ["timeout"] }
 parking_lot = "0.12"
 cfg-if = "1.0"
 uuid = "1"
 dunce = "1.0"
 tower = "0.5"
 futures-util = "0.3"
-libsql = { version = "0.9", optional = true, features = [ "core", "stream"] }
-tokio-postgres = { version = "0.7", optional = true }
-bb8 = { version = "0.9.0", optional = true }
-bb8-postgres = { version = "0.9.0", optional = true }
+libsql = { version = "0.9", optional = true, default-features = false, features = ["core", "sync"] }
+tokio-postgres = { version = "0.7", optional = true, features = [
+  "with-chrono-0_4",
+] }
+bb8 = { version = "0.9", optional = true }
+bb8-postgres = { version = "0.9", optional = true }
 
 [features]
 default = ["libsql"]

crates/devolutions-pedm/Cargo.toml

@@ -1,20 +1,39 @@
-/* In SQLite, we store time as integer with microsecond precision. This is the same precision used by TIMESTAMPTZ in Postgres. */
+/* In SQLite, we store time as an 8-byte integer (i64) with microsecond precision. This matches TIMESTAMPTZ in Postgres.
+   Use `chrono::DateTime::timestamp_micros` when inserting or fetching timestamps in Rust.
+*/
 
-CREATE TABLE pedm_run (
-    id         INTEGER PRIMARY KEY AUTOINCREMENT,
-    start_time INTEGER NOT NULL DEFAULT (CAST(strftime('%f', 'now') * 1000000 AS INTEGER)),
-    pipe_name  TEXT NOT NULL
+CREATE TABLE IF NOT EXISTS pedm_schema_version
+(
+    version    integer PRIMARY KEY,
+    updated_at integer NOT NULL DEFAULT (
+        CAST(strftime('%s', 'now') AS integer) * 1000000 + CAST(strftime('%f', 'now') * 1000000 AS integer) % 1000000
+        )
 );
 
-CREATE TABLE http_request (
-    id          INTEGER PRIMARY KEY,
-    at          INTEGER NOT NULL DEFAULT (CAST(strftime('%f', 'now') * 1000000 AS INTEGER)),
-    method      TEXT NOT NULL,
-    path        TEXT NOT NULL,
-    status_code INTEGER NOT NULL
+CREATE TABLE IF NOT EXISTS pedm_run
+(
+    id         integer PRIMARY KEY AUTOINCREMENT,
+    start_time integer NOT NULL DEFAULT (
+        CAST(strftime('%s', 'now') AS integer) * 1000000 + CAST(strftime('%f', 'now') * 1000000 AS integer) % 1000000
+        ),
+    pipe_name  text    NOT NULL
 );
 
-CREATE TABLE elevate_tmp_request (
-    req_id  INTEGER PRIMARY KEY,
-    seconds INTEGER NOT NULL
+CREATE TABLE IF NOT EXISTS http_request
+(
+    id          integer PRIMARY KEY,
+    at          integer NOT NULL DEFAULT (
+        CAST(strftime('%s', 'now') AS integer) * 1000000 + CAST(strftime('%f', 'now') * 1000000 AS integer) % 1000000
+        ),
+    method      text    NOT NULL,
+    path        text    NOT NULL,
+    status_code integer NOT NULL
 );
+
+CREATE TABLE IF NOT EXISTS elevate_tmp_request
+(
+    req_id  integer PRIMARY KEY,
+    seconds integer NOT NULL
+);
+
+INSERT INTO pedm_schema_version (version) VALUES (1) ON CONFLICT DO NOTHING;

crates/devolutions-pedm/schema/libsql.sql

@@ -1,22 +1,31 @@
+CREATE TABLE IF NOT EXISTS pedm_schema_version
+(
+    version  smallint PRIMARY KEY,
+    add_time timestamptz NOT NULL DEFAULT NOW()
+);
+
 /* The startup of the server */
-CREATE TABLE pedm_run
+CREATE TABLE IF NOT EXISTS pedm_run
 (
     id         int PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
     start_time timestamptz NOT NULL DEFAULT NOW(),
     pipe_name  text        NOT NULL
 );
 
-CREATE TABLE http_request
+CREATE TABLE IF NOT EXISTS http_request
 (
     id          integer PRIMARY KEY,
     at          timestamptz NOT NULL DEFAULT NOW(),
     method      text        NOT NULL,
-    path       text        NOT NULL,
+    path        text        NOT NULL,
     status_code smallint    NOT NULL
 );
 
-CREATE TABLE elevate_tmp_request
+/* The request ID is `http_request(id)` but the http_request INSERT only executes in middleware after the response, so we don't use a FK. */
+CREATE TABLE IF NOT EXISTS elevate_tmp_request
 (
-    req_id  integer PRIMARY KEY,  /* this is http_request but the http_request INSERT only executes in middleware after the response, so we don't use a FK */
+    req_id  integer PRIMARY KEY,
     seconds int NOT NULL
-);
+);
+
+INSERT INTO pedm_schema_version (version) VALUES (1) ON CONFLICT DO NOTHING;

crates/devolutions-pedm/schema/pg.sql

@@ -0,0 +1,24 @@
+use std::sync::atomic::Ordering;
+
+use aide::NoApi;
+use axum::extract::State;
+use axum::Json;
+
+use super::err::HandlerError;
+use super::state::AppState;
+use crate::db::Db;
+use crate::model::AboutData;
+
+/// Gets info about the current state of the application.
+pub(crate) async fn about(
+    NoApi(State(state)): NoApi<State<AppState>>,
+    NoApi(Db(db)): NoApi<Db>,
+) -> Result<Json<AboutData>, HandlerError> {
+    let requests_received = state.req_counter.load(Ordering::Relaxed) - state.startup_info.request_count;
+
+    Ok(Json(AboutData {
+        startup_info: state.startup_info,
+        requests_received,
+        last_request_time: db.get_last_request_time().await?,
+    }))
+}

crates/devolutions-pedm/src/api/about.rs

@@ -9,11 +9,11 @@ use parking_lot::RwLock;
 use schemars::JsonSchema;
 use serde::{Deserialize, Serialize};
 
+use crate::db::Db;
 use crate::elevations;
 use crate::policy::Policy;
 
 use super::err::HandlerError;
-use super::state::Db;
 use super::NamedPipeConnectInfo;
 
 #[derive(Deserialize, Serialize, JsonSchema, Debug)]

crates/devolutions-pedm/src/api/elevate_temporary.rs

@@ -32,18 +32,12 @@ use win_api_wrappers::token::Token;
 use win_api_wrappers::undoc::PIPE_ACCESS_FULL_CONTROL;
 use win_api_wrappers::utils::Pipe;
 
-use elevate_session::elevate_session;
-use elevate_temporary::elevate_temporary;
-use launch::post_launch;
-use revoke::post_revoke;
-use state::{AppState, AppStateError};
-use status::get_status;
-
 use crate::config::Config;
-use crate::db::DbError;
+use crate::db::{Db, DbError, InitSchemaError};
 use crate::error::{Error, ErrorResponse};
 use crate::utils::AccountExt;
 
+mod about;
 mod elevate_session;
 mod elevate_temporary;
 mod err;
@@ -53,6 +47,14 @@ mod revoke;
 pub(crate) mod state;
 mod status;
 
+use about::about;
+use elevate_session::elevate_session;
+use elevate_temporary::elevate_temporary;
+use launch::post_launch;
+use revoke::post_revoke;
+use state::{AppState, AppStateError};
+use status::get_status;
+
 #[derive(Debug, Clone)]
 struct NamedPipeConnectInfo {
     pub(crate) user: User,
@@ -132,6 +134,7 @@ fn create_pipe(pipe_name: &str) -> anyhow::Result<NamedPipeServer> {
 
 pub(crate) fn api_router() -> ApiRouter<AppState> {
     ApiRouter::new()
+        .api_route("/about", aide::axum::routing::get(about))
         .api_route("/elevate/temporary", aide::axum::routing::post(elevate_temporary))
         .api_route("/elevate/session", aide::axum::routing::post(elevate_session))
         .api_route("/launch", aide::axum::routing::post(post_launch))
@@ -165,8 +168,12 @@ async fn health_check() -> &'static str {
     "OK"
 }
 
+/// Initializes the appliation and starts the named pipe server.
 pub async fn serve(config: Config) -> Result<(), ServeError> {
-    let state = AppState::load(&config).await?;
+    let db = Db::new(&config).await?;
+    db.init_schema().await?;
+
+    let state = AppState::new(db, &config.pipe_name).await?;
 
     // a plain Axum router
     let hello_router = Router::new().route("/health", axum::routing::get(health_check));
@@ -182,11 +189,11 @@ pub async fn serve(config: Config) -> Result<(), ServeError> {
     let mut server = create_pipe(pipe_name)?;
 
     // Log the server startup.
-    let run_id = state.db.log_server_startup(pipe_name).await?;
-    info!("Started server at {pipe_name} with run ID {run_id}");
+    info!("Started named pipe server with name `{pipe_name}`");
     info!(
-        "Starting request ID counter at {}",
-        state.req_counter.load(Ordering::Relaxed)
+        "Run ID is {run_id}, request ID counter is {req_count}",
+        run_id = state.startup_info.run_id,
+        req_count = state.req_counter.load(Ordering::Relaxed)
     );
 
     loop {
@@ -219,6 +226,7 @@ pub enum ServeError {
     TokioIo(tokio::io::Error),
     AppState(AppStateError),
     Db(DbError),
+    InitSchema(InitSchemaError),
     Other(anyhow::Error),
 }
 
@@ -228,6 +236,7 @@ impl core::error::Error for ServeError {
             Self::TokioIo(e) => Some(e),
             Self::AppState(e) => Some(e),
             Self::Db(e) => Some(e),
+            Self::InitSchema(e) => Some(e),
             Self::Other(e) => Some(e.as_ref()),
         }
     }
@@ -239,6 +248,7 @@ impl fmt::Display for ServeError {
             Self::TokioIo(e) => e.fmt(f),
             Self::AppState(e) => e.fmt(f),
             Self::Db(e) => e.fmt(f),
+            Self::InitSchema(e) => e.fmt(f),
             Self::Other(e) => e.fmt(f),
         }
     }
@@ -259,6 +269,11 @@ impl From<DbError> for ServeError {
         Self::Db(e)
     }
 }
+impl From<InitSchemaError> for ServeError {
+    fn from(e: InitSchemaError) -> Self {
+        Self::InitSchema(e)
+    }
+}
 impl From<anyhow::Error> for ServeError {
     fn from(e: anyhow::Error) -> Self {
         Self::Other(e)