tasks: implement execIfModified

Author: domenkozar

Cargo.lock

@@ -9,6 +9,7 @@ members = [
     "http-client-tls",
     "nix-conf-parser",
     "xtask",
+    "devenv-cache-core",
 ]
 
 [workspace.package]
@@ -23,6 +24,7 @@ devenv = { path = "devenv" }
 devenv-eval-cache = { path = "devenv-eval-cache" }
 devenv-run-tests = { path = "devenv-run-tests" }
 devenv-tasks = { path = "devenv-tasks" }
+devenv-cache-core = { path = "devenv-cache-core" }
 http-client-tls = { path = "http-client-tls" }
 nix-conf-parser = { path = "nix-conf-parser" }
 xtask = { path = "xtask" }
@@ -83,6 +85,7 @@ which = "7.0.2"
 whoami = "1.5.1"
 xdg = "2.5.2"
 tokio-tar = "0.3.1"
+walkdir = "2.3"
 
 # The version of rustls must match the version used by reqwest to set up rustls-platform-verifier.
 # If you encounter an error, lock these versions down.

Cargo.toml

@@ -0,0 +1,38 @@
+[package]
+name = "devenv-cache-core"
+version = "0.1.0"
+edition = "2021"
+description = "Core utilities for file tracking and caching in devenv"
+license = "MIT"
+
+[dependencies]
+# Database
+sqlx = { workspace = true, features = [
+    "runtime-tokio",
+    "tls-rustls",
+    "sqlite",
+    "migrate",
+    "macros",
+] }
+
+# Error handling
+thiserror.workspace = true
+miette.workspace = true
+eyre.workspace = true
+
+# Hashing
+blake3.workspace = true
+
+# File operations
+walkdir.workspace = true
+
+# Async runtime
+tokio = { workspace = true, features = ["fs", "macros", "time"] }
+
+# Logging
+tracing.workspace = true
+serde_json.workspace = true
+
+[dev-dependencies]
+tempfile.workspace = true
+tokio = { workspace = true, features = ["rt", "macros"] }

devenv-cache-core/Cargo.toml

@@ -0,0 +1,177 @@
+use crate::error::{CacheError, CacheResult};
+use sqlx::migrate::MigrateDatabase;
+use sqlx::sqlite::{
+    SqliteConnectOptions, SqliteJournalMode, SqlitePool, SqlitePoolOptions, SqliteSynchronous,
+};
+use std::path::PathBuf;
+use std::str::FromStr;
+use std::time::Duration;
+use tracing::{debug, error, info};
+
+/// Database connection manager
+#[derive(Debug, Clone)]
+pub struct Database {
+    pool: SqlitePool,
+    path: PathBuf,
+}
+
+impl Database {
+    /// Create a new database connection
+    pub async fn new(path: PathBuf) -> CacheResult<Self> {
+        // Ensure parent directory exists
+        if let Some(parent) = path.parent() {
+            if !parent.exists() {
+                std::fs::create_dir_all(parent).map_err(|e| {
+                    CacheError::initialization(format!(
+                        "Failed to create database directory: {}",
+                        e
+                    ))
+                })?;
+            }
+        }
+
+        let db_url = format!("sqlite:{}", path.display());
+
+        let options = connection_options(&db_url)?;
+
+        // Create database if it doesn't exist
+        if !sqlx::Sqlite::database_exists(&db_url)
+            .await
+            .unwrap_or(false)
+        {
+            info!("Creating new database at {}", path.display());
+            sqlx::Sqlite::create_database(&db_url).await?;
+        }
+
+        let pool = SqlitePoolOptions::new()
+            .max_connections(5)
+            .connect_with(options)
+            .await?;
+
+        info!("Connected to database at {}", path.display());
+
+        Ok(Self { pool, path })
+    }
+
+    /// Get a reference to the connection pool
+    pub fn pool(&self) -> &SqlitePool {
+        &self.pool
+    }
+
+    /// Initialize the database with the given schema
+    pub async fn init_schema(&self, schema: &str) -> CacheResult<()> {
+        debug!("Initializing database schema");
+        sqlx::query(schema).execute(&self.pool).await?;
+        Ok(())
+    }
+
+    /// Run migrations from the embed_migrations! macro
+    pub async fn run_migrations<M>(&self, migrations: M) -> CacheResult<()>
+    where
+        M: 'static
+            + Send
+            + Sync
+            + Fn(
+                &SqlitePool,
+            ) -> std::pin::Pin<
+                Box<dyn std::future::Future<Output = Result<(), sqlx::Error>> + Send>,
+            >,
+    {
+        debug!("Running database migrations");
+
+        // Attempt to run migrations
+        if let Err(err) = migrations(&self.pool).await {
+            error!("Failed to migrate database: {}", err);
+
+            // If the database exists and migrations failed, try to recreate it
+            if sqlx::Sqlite::database_exists(&format!("sqlite:{}", self.path.display())).await? {
+                // Close pool and recreate database
+                self.pool.close().await;
+
+                let db_url = format!("sqlite:{}", self.path.display());
+                sqlx::Sqlite::drop_database(&db_url).await?;
+                sqlx::Sqlite::create_database(&db_url).await?;
+
+                // Reconnect and try migrations again
+                let options = connection_options(&db_url)?;
+                let pool = SqlitePoolOptions::new()
+                    .max_connections(5)
+                    .connect_with(options)
+                    .await?;
+
+                let new_db = Self {
+                    pool,
+                    path: self.path.clone(),
+                };
+
+                migrations(&new_db.pool).await?;
+                return Ok(());
+            }
+
+            return Err(err.into());
+        }
+
+        Ok(())
+    }
+
+    /// Close the database connection
+    pub async fn close(self) {
+        self.pool.close().await;
+    }
+}
+
+/// Create SQLite connection options
+fn connection_options(db_url: &str) -> CacheResult<SqliteConnectOptions> {
+    let options = SqliteConnectOptions::from_str(db_url)?
+        .journal_mode(SqliteJournalMode::Wal)
+        .synchronous(SqliteSynchronous::Normal)
+        .busy_timeout(Duration::from_secs(10))
+        .create_if_missing(true)
+        .foreign_keys(true)
+        .pragma("wal_autocheckpoint", "1000")
+        .pragma("journal_size_limit", (64 * 1024 * 1024).to_string()) // 64 MB
+        .pragma("mmap_size", "134217728") // 128 MB
+        .pragma("cache_size", "2000"); // 2000 pages
+
+    Ok(options)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tempfile::TempDir;
+
+    #[tokio::test]
+    async fn test_database_creation() {
+        let temp_dir = TempDir::new().unwrap();
+        let db_path = temp_dir.path().join("test.db");
+
+        let db = Database::new(db_path.clone()).await.unwrap();
+
+        // Test that the database file was created
+        assert!(db_path.exists());
+
+        // Test schema initialization
+        let schema = "CREATE TABLE test (id INTEGER PRIMARY KEY, name TEXT)";
+        db.init_schema(schema).await.unwrap();
+
+        // Test that we can execute queries
+        sqlx::query("INSERT INTO test (name) VALUES (?)")
+            .bind("test_value")
+            .execute(db.pool())
+            .await
+            .unwrap();
+
+        // Test query
+        let row: (i64, String) = sqlx::query_as("SELECT id, name FROM test WHERE name = ?")
+            .bind("test_value")
+            .fetch_one(db.pool())
+            .await
+            .unwrap();
+
+        assert_eq!(row.1, "test_value");
+
+        // Close database
+        db.close().await;
+    }
+}

devenv-cache-core/src/db.rs

@@ -0,0 +1,46 @@
+use miette::Diagnostic;
+use std::path::PathBuf;
+use thiserror::Error;
+
+/// Common error type for cache operations
+#[derive(Error, Diagnostic, Debug)]
+pub enum CacheError {
+    #[error("Database error: {0}")]
+    Database(#[from] sqlx::Error),
+
+    #[error("I/O error: {0}")]
+    Io(#[from] std::io::Error),
+
+    #[error("Failed to initialize cache: {0}")]
+    Initialization(String),
+
+    #[error("JSON serialization error: {0}")]
+    Json(#[from] serde_json::Error),
+
+    #[error("File not found: {0}")]
+    FileNotFound(PathBuf),
+
+    #[error("Environment variable not set: {0}")]
+    MissingEnvVar(String),
+
+    #[error("Invalid path: {0}")]
+    InvalidPath(PathBuf),
+
+    #[error("Content hash calculation failed for {path}: {reason}")]
+    HashFailure { path: PathBuf, reason: String },
+}
+
+impl CacheError {
+    /// Create a new initialization error
+    pub fn initialization<S: ToString>(message: S) -> Self {
+        Self::Initialization(message.to_string())
+    }
+
+    /// Create a new missing environment variable error
+    pub fn missing_env_var<S: ToString>(var_name: S) -> Self {
+        Self::MissingEnvVar(var_name.to_string())
+    }
+}
+
+/// A specialized result type for cache operations
+pub type CacheResult<T> = std::result::Result<T, CacheError>;