initial implementation of the Nx remote cache API

.gitignore

@@ -0,0 +1,2 @@
+/target
+.env

Cargo.toml

@@ -0,0 +1,14 @@
+[package]
+name = "enex-rcache"
+version = "0.1.0"
+edition = "2024"
+
+[dependencies]
+actix-web = "4.10.2"
+actix-web-httpauth = "0.8.2"
+config = "0.15.11"
+dotenvy = "0.15.7"
+futures-util = "0.3.31"
+serde = "1.0.219"
+tracing = "0.1.41"
+tracing-subscriber = "0.3.19"

README.md

@@ -0,0 +1,3 @@
+# enex-rcache - remote build cache implementation for Nx
+
+This is an implementation of the [Nx remote cache OpenAPI specification](https://nx.dev/recipes/running-tasks/self-hosted-caching#open-api-specification).

src/access.rs

@@ -0,0 +1,35 @@
+use tracing::debug;
+
+use crate::config::AppConfig;
+
+#[derive(PartialEq)]
+pub enum AccessLevel {
+    Read,
+    ReadWrite,
+}
+
+pub fn get_access_level(config: &AppConfig, token: &str) -> Option<AccessLevel> {
+    if config.read_write_tokens.contains(token) {
+        Some(AccessLevel::ReadWrite)
+    } else if config.read_tokens.contains(token) {
+        Some(AccessLevel::Read)
+    } else {
+        None
+    }
+}
+
+pub fn has_access_level(
+    config: &AppConfig,
+    token: &str,
+    requested_access_level: AccessLevel,
+) -> bool {
+    if let Some(access_level) = get_access_level(&config, token) {
+        match access_level {
+            AccessLevel::ReadWrite => true,
+            AccessLevel::Read => requested_access_level == AccessLevel::Read,
+        }
+    } else {
+        debug!("Missing access token");
+        false
+    }
+}

src/config.rs

@@ -0,0 +1,28 @@
+use std::collections::HashSet;
+
+use serde::Deserialize;
+
+use config::{Config, Environment};
+
+#[derive(Debug, Deserialize)]
+pub struct AppConfig {
+    pub cache_dir: String,
+    pub read_tokens: HashSet<String>,
+    pub read_write_tokens: HashSet<String>,
+}
+
+pub fn load_config() -> AppConfig {
+    let settings = Config::builder()
+        .add_source(
+            Environment::with_prefix("ENEX")
+                .try_parsing(true)
+                .list_separator(",")
+                .with_list_parse_key("read_tokens")
+                .with_list_parse_key("read_write_tokens"),
+        )
+        .build()
+        .expect("Could not initialize config.");
+    settings
+        .try_deserialize::<AppConfig>()
+        .expect("Could not parse config.")
+}

src/handlers.rs

@@ -0,0 +1,86 @@
+use std::{io::Write, path::PathBuf};
+
+use actix_web::{
+    HttpResponse, Responder, get, put,
+    web::{Data, Path, Payload},
+};
+use actix_web_httpauth::extractors::bearer::BearerAuth;
+use futures_util::StreamExt;
+use tracing::{debug, error, trace};
+
+use crate::{
+    access::{AccessLevel, has_access_level},
+    config::AppConfig,
+};
+
+fn get_cache_entry_path(config: &AppConfig, hash: &str) -> PathBuf {
+    PathBuf::from(&config.cache_dir).join(hash)
+}
+
+#[get("/v1/cache/{hash}")]
+pub async fn get_cache_item(
+    app_data: Data<AppConfig>,
+    auth: BearerAuth,
+    hash: Path<String>,
+) -> impl Responder {
+    trace!("Requested cache item {}", hash.as_str());
+    if !has_access_level(&app_data, auth.token(), AccessLevel::Read) {
+        debug!(
+            "Tried to read cache item {} without valid read access token.",
+            hash.as_str()
+        );
+        return HttpResponse::Unauthorized()
+            .content_type("text/plain")
+            .body("Please provide a valid access token with at least read-level access.");
+    }
+    let path = get_cache_entry_path(&app_data, &hash);
+    if !path.exists() {
+        trace!("Cache item not found: {}", hash.as_str());
+        HttpResponse::NotFound().body("The record was not found.")
+    } else {
+        trace!("Returning cache item {}", hash.as_str());
+        HttpResponse::Ok().body(std::fs::read(path).unwrap())
+    }
+}
+
+#[put("/v1/cache/{hash}")]
+pub async fn put_cache_item(
+    app_data: Data<AppConfig>,
+    auth: BearerAuth,
+    hash: Path<String>,
+    mut body: Payload,
+) -> impl Responder {
+    trace!("Received cache item {}", hash.as_str());
+    if !has_access_level(&app_data, auth.token(), AccessLevel::ReadWrite) {
+        debug!(
+            "Tried to write cache item {} without valid read-write access token.",
+            hash.as_str()
+        );
+        return HttpResponse::Unauthorized()
+            .content_type("text/plain")
+            .body("Please provide a valid access token with read-write access.");
+    }
+    let path = get_cache_entry_path(&app_data, &hash);
+    let file = std::fs::File::create_new(&path);
+    match file {
+        Ok(mut file) => {
+            while let Some(chunk) = body.next().await {
+                match chunk {
+                    Ok(chunk) => file.write_all(&chunk).expect("This should actually work"),
+                    Err(e) => {
+                        error!("Could not write cache item chunk: {}", e);
+                        drop(file);
+                        std::fs::remove_file(path).unwrap(); // Clean up to make sure the block doesn't get half-written with the wrong content
+                        break;
+                    }
+                }
+            }
+            debug!("Created cache item {}", hash.as_str());
+            HttpResponse::Accepted().finish()
+        }
+        Err(_) => {
+            trace!("Tried to overwrite existing cache item {}", hash.as_str());
+            HttpResponse::Conflict().body("Cannot overwrite an existing record.")
+        }
+    }
+}

src/main.rs

@@ -0,0 +1,32 @@
+use actix_web::{App, HttpServer, web::Data};
+use dotenvy::dotenv;
+use tracing::Level;
+use tracing_subscriber::FmtSubscriber;
+
+mod access;
+mod config;
+mod handlers;
+
+#[actix_web::main]
+async fn main() -> std::io::Result<()> {
+    let subscriber = FmtSubscriber::builder()
+        .with_max_level(Level::TRACE)
+        .finish();
+
+    tracing::subscriber::set_global_default(subscriber).expect("Could not initialize logging.");
+
+    dotenv().expect("Could not load .env file.");
+
+    HttpServer::new(|| {
+        let config = config::load_config();
+        App::new()
+            .app_data(Data::new(config))
+            .wrap(actix_web::middleware::Logger::default())
+            .service(handlers::get_cache_item)
+            .service(handlers::put_cache_item)
+    })
+    .bind(("127.0.0.1", 8080))?
+    .bind(("::1", 8080))?
+    .run()
+    .await
+}