937 lines
35 KiB
Rust
937 lines
35 KiB
Rust
use crate::config::Config;
|
|
use crate::manifest::Manifest;
|
|
use crate::sync::SyncQueue;
|
|
use anyhow::Result;
|
|
use axum::{
|
|
extract::State,
|
|
response::Sse,
|
|
routing::{get, post},
|
|
Json, Router,
|
|
};
|
|
use chrono::{DateTime, Utc};
|
|
use serde::{Deserialize, Serialize};
|
|
use std::{
|
|
convert::Infallible,
|
|
path::PathBuf,
|
|
sync::Arc,
|
|
time::Duration,
|
|
};
|
|
use tokio_stream::wrappers::IntervalStream;
|
|
use tokio_stream::StreamExt;
|
|
use tower_http::cors::{Any, CorsLayer};
|
|
|
|
// ── AppState ──────────────────────────────────────────────────────────────────
|
|
|
|
pub struct AppState {
|
|
pub cfg: Config,
|
|
pub manifest_path: PathBuf,
|
|
}
|
|
|
|
// ── API response types ────────────────────────────────────────────────────────
|
|
|
|
#[derive(Serialize)]
|
|
pub struct NodeStatus {
|
|
pub node_name: String,
|
|
pub role: String,
|
|
pub hot_used_gb: f64,
|
|
pub hot_max_gb: u64,
|
|
pub hot_path: String,
|
|
pub warm_dataset: String,
|
|
pub warm_used_gb: f64,
|
|
pub warm_avail_gb: f64,
|
|
pub warm_snapshot_count: usize,
|
|
pub cold_path: Option<String>,
|
|
pub peer_host: Option<String>,
|
|
pub peer_reachable: bool,
|
|
pub zfs_pool_state: String,
|
|
pub active_project_count: usize,
|
|
pub daemon_uptime_secs: u64,
|
|
/// Number of git-sync notifications waiting for the peer to accept.
|
|
pub sync_queue_depth: usize,
|
|
/// Number of queued jobs that have exceeded the escalation threshold
|
|
/// (~1h of retries). Non-zero means a peer has been unreachable long
|
|
/// enough that we've stopped hiding it in WARN-level logs.
|
|
pub sync_queue_stuck: usize,
|
|
}
|
|
|
|
#[derive(Serialize)]
|
|
pub struct ProjectInfo {
|
|
pub name: String,
|
|
pub warm_path: String,
|
|
pub hot_target_path: String,
|
|
pub is_active: bool,
|
|
pub hot_size_mb: f64,
|
|
pub last_build: Option<DateTime<Utc>>,
|
|
pub last_active: Option<DateTime<Utc>>,
|
|
pub last_sync: Option<DateTime<Utc>>,
|
|
pub is_building: bool,
|
|
pub git_branch: String,
|
|
pub git_ahead: u32,
|
|
}
|
|
|
|
#[derive(Serialize)]
|
|
pub struct SnapshotInfo {
|
|
pub name: String,
|
|
pub kind: String,
|
|
pub timestamp: Option<DateTime<Utc>>,
|
|
pub used_bytes: u64,
|
|
pub refer_bytes: u64,
|
|
}
|
|
|
|
#[derive(Serialize)]
|
|
pub struct HotEntry {
|
|
pub project: String,
|
|
pub size_mb: f64,
|
|
pub last_modified: Option<DateTime<Utc>>,
|
|
}
|
|
|
|
#[derive(Deserialize)]
|
|
pub struct ProjectBody {
|
|
pub project: String,
|
|
}
|
|
|
|
#[derive(Serialize)]
|
|
pub struct OkResponse {
|
|
pub ok: bool,
|
|
#[serde(skip_serializing_if = "Option::is_none")]
|
|
pub error: Option<String>,
|
|
}
|
|
|
|
// ── helpers ───────────────────────────────────────────────────────────────────
|
|
|
|
fn load_manifest(state: &AppState) -> Manifest {
|
|
Manifest::load(&state.manifest_path).unwrap_or_default()
|
|
}
|
|
|
|
/// Returns how long the daemon has been running by reading its start-time file.
|
|
fn daemon_uptime_secs() -> u64 {
|
|
std::fs::read_to_string(crate::daemon::DAEMON_STARTED_PATH)
|
|
.ok()
|
|
.and_then(|s| s.trim().parse::<i64>().ok())
|
|
.map(|started| (chrono::Utc::now().timestamp() - started).max(0) as u64)
|
|
.unwrap_or(0)
|
|
}
|
|
|
|
/// Validates that a project name is a safe `org/repo` slug.
|
|
/// Rejects traversal attempts, shell metacharacters, and malformed names.
|
|
fn validate_project_name(name: &str) -> Result<(), String> {
|
|
let parts: Vec<&str> = name.split('/').collect();
|
|
let ok = parts.len() == 2
|
|
&& parts.iter().all(|p| {
|
|
!p.is_empty()
|
|
&& p.chars().all(|c| c.is_alphanumeric() || c == '-' || c == '_' || c == '.')
|
|
});
|
|
if ok {
|
|
Ok(())
|
|
} else {
|
|
Err(format!(
|
|
"invalid project name '{}': expected org/repo using alphanumeric, -, _, .",
|
|
name
|
|
))
|
|
}
|
|
}
|
|
|
|
fn zfs_used_avail(dataset: &str) -> (f64, f64) {
|
|
let out = std::process::Command::new("zfs")
|
|
.args(["list", "-H", "-p", "-o", "used,avail", dataset])
|
|
.output();
|
|
match out {
|
|
Ok(o) if o.status.success() => {
|
|
let s = String::from_utf8_lossy(&o.stdout);
|
|
let mut parts = s.split_whitespace();
|
|
let used = parts.next().and_then(|v| v.parse::<u64>().ok()).unwrap_or(0);
|
|
let avail = parts.next().and_then(|v| v.parse::<u64>().ok()).unwrap_or(0);
|
|
(
|
|
used as f64 / 1_073_741_824.0,
|
|
avail as f64 / 1_073_741_824.0,
|
|
)
|
|
}
|
|
_ => (0.0, 0.0),
|
|
}
|
|
}
|
|
|
|
fn zfs_pool_state(dataset: &str) -> String {
|
|
let pool = dataset.splitn(2, '/').next().unwrap_or(dataset);
|
|
let out = std::process::Command::new("zpool")
|
|
.args(["list", "-H", "-o", "health", pool])
|
|
.output();
|
|
match out {
|
|
Ok(o) if o.status.success() => String::from_utf8_lossy(&o.stdout).trim().to_string(),
|
|
_ => "UNKNOWN".to_string(),
|
|
}
|
|
}
|
|
|
|
fn zfs_snapshot_count(dataset: &str) -> usize {
|
|
let out = std::process::Command::new("zfs")
|
|
.args(["list", "-H", "-t", "snapshot", "-o", "name", "-r", dataset])
|
|
.output();
|
|
match out {
|
|
Ok(o) if o.status.success() => {
|
|
String::from_utf8_lossy(&o.stdout)
|
|
.lines()
|
|
.filter(|l| !l.trim().is_empty())
|
|
.count()
|
|
}
|
|
_ => 0,
|
|
}
|
|
}
|
|
|
|
fn peer_reachable(peer: &crate::config::PeerConfig) -> bool {
|
|
let target = format!("{}@{}", peer.user, peer.host);
|
|
std::process::Command::new("ssh")
|
|
.args([
|
|
"-o", "ConnectTimeout=3",
|
|
"-o", "BatchMode=yes",
|
|
&target,
|
|
"true",
|
|
])
|
|
.status()
|
|
.map(|s| s.success())
|
|
.unwrap_or(false)
|
|
}
|
|
|
|
fn hot_used_gb(manifest: &Manifest) -> f64 {
|
|
let total: u64 = manifest.projects.iter().map(|p| {
|
|
crate::hot::target_size_bytes(&p.hot_target_path).unwrap_or(0)
|
|
}).sum();
|
|
total as f64 / 1_073_741_824.0
|
|
}
|
|
|
|
fn du_bytes(path: &std::path::Path) -> u64 {
|
|
if !path.exists() { return 0; }
|
|
let out = std::process::Command::new("du")
|
|
.args(["-sb", path.to_str().unwrap_or("")])
|
|
.output();
|
|
match out {
|
|
Ok(o) => String::from_utf8_lossy(&o.stdout)
|
|
.split_whitespace()
|
|
.next()
|
|
.and_then(|s| s.parse().ok())
|
|
.unwrap_or(0),
|
|
_ => 0,
|
|
}
|
|
}
|
|
|
|
fn git_branch(warm_path: &std::path::Path) -> String {
|
|
let out = std::process::Command::new("git")
|
|
.args(["-C", warm_path.to_str().unwrap_or(""), "rev-parse", "--abbrev-ref", "HEAD"])
|
|
.output();
|
|
match out {
|
|
Ok(o) if o.status.success() => String::from_utf8_lossy(&o.stdout).trim().to_string(),
|
|
_ => "unknown".to_string(),
|
|
}
|
|
}
|
|
|
|
/// Returns true if the project's .cargo/config.toml has a claw-store-managed
|
|
/// target-dir pointing at the hot tier (catches projects activated on another node).
|
|
fn has_hot_cargo_config(warm_path: &std::path::Path, hot_path: &std::path::Path) -> bool {
|
|
let config = warm_path.join(".cargo/config.toml");
|
|
if let Ok(content) = std::fs::read_to_string(&config) {
|
|
let hot_str = hot_path.to_string_lossy();
|
|
content.contains("target-dir") && content.contains(hot_str.as_ref())
|
|
} else {
|
|
false
|
|
}
|
|
}
|
|
|
|
fn git_ahead(warm_path: &std::path::Path) -> u32 {
|
|
let out = std::process::Command::new("git")
|
|
.args(["-C", warm_path.to_str().unwrap_or(""), "rev-list", "@{u}..HEAD", "--count"])
|
|
.output();
|
|
match out {
|
|
Ok(o) if o.status.success() => {
|
|
String::from_utf8_lossy(&o.stdout).trim().parse().unwrap_or(0)
|
|
}
|
|
_ => 0,
|
|
}
|
|
}
|
|
|
|
fn is_building(warm_path: &std::path::Path) -> bool {
|
|
let warm_str = match warm_path.to_str() {
|
|
Some(s) => s,
|
|
None => return false,
|
|
};
|
|
// Walk /proc/<pid>/cwd and check if it matches warm_path
|
|
if let Ok(proc) = std::fs::read_dir("/proc") {
|
|
for entry in proc.flatten() {
|
|
let name = entry.file_name();
|
|
let name_str = name.to_string_lossy();
|
|
if !name_str.chars().all(|c| c.is_ascii_digit()) { continue; }
|
|
let cwd_link = entry.path().join("cwd");
|
|
if let Ok(cwd) = std::fs::read_link(&cwd_link) {
|
|
if cwd.to_str().map(|s| s.starts_with(warm_str)).unwrap_or(false) {
|
|
// check if exe is cargo or rustc
|
|
let exe_link = entry.path().join("exe");
|
|
if let Ok(exe) = std::fs::read_link(&exe_link) {
|
|
let exe_str = exe.to_string_lossy();
|
|
if exe_str.contains("cargo") || exe_str.contains("rustc") {
|
|
return true;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
false
|
|
}
|
|
|
|
fn last_modified_time(path: &std::path::Path) -> Option<DateTime<Utc>> {
|
|
path.metadata().ok()
|
|
.and_then(|m| m.modified().ok())
|
|
.map(|t| {
|
|
let secs = t.duration_since(std::time::UNIX_EPOCH)
|
|
.map(|d| d.as_secs())
|
|
.unwrap_or(0);
|
|
DateTime::from_timestamp(secs as i64, 0).unwrap_or_else(Utc::now)
|
|
})
|
|
}
|
|
|
|
// Parse snapshot kind from name like "slab/projects@hourly-2026-06-16-0400"
|
|
fn parse_snapshot_kind(snap_name: &str) -> String {
|
|
// The part after @ is like "hourly-2026-06-16-0400"
|
|
let after_at = snap_name.splitn(2, '@').nth(1).unwrap_or(snap_name);
|
|
// kind is everything before first '-YYYY'
|
|
// Find where the date starts (4-digit year after a dash)
|
|
let parts: Vec<&str> = after_at.splitn(2, '-').collect();
|
|
parts.first().copied().unwrap_or("unknown").to_string()
|
|
}
|
|
|
|
fn parse_snapshot_timestamp(snap_name: &str) -> Option<DateTime<Utc>> {
|
|
// Extract YYYY-MM-DD-HHMM pattern from snapshot name
|
|
let after_at = snap_name.splitn(2, '@').nth(1).unwrap_or(snap_name);
|
|
// Find a 4-digit year segment
|
|
let re_parts: Vec<&str> = after_at.split('-').collect();
|
|
// Look for pattern: YYYY-MM-DD-HHMM
|
|
for i in 0..re_parts.len().saturating_sub(3) {
|
|
if re_parts[i].len() == 4 && re_parts[i].chars().all(|c| c.is_ascii_digit()) {
|
|
let yyyy = re_parts[i];
|
|
let mm = re_parts.get(i + 1).copied().unwrap_or("01");
|
|
let dd = re_parts.get(i + 2).copied().unwrap_or("01");
|
|
let hhmm = re_parts.get(i + 3).copied().unwrap_or("0000");
|
|
let hh = &hhmm.get(..2).unwrap_or("00");
|
|
let mi = &hhmm.get(2..4).unwrap_or("00");
|
|
let s = format!("{}-{}-{}T{}:{}:00Z", yyyy, mm, dd, hh, mi);
|
|
if let Ok(dt) = s.parse::<DateTime<Utc>>() {
|
|
return Some(dt);
|
|
}
|
|
}
|
|
}
|
|
None
|
|
}
|
|
|
|
async fn run_claw_store_cmd(args: &[&str]) -> OkResponse {
|
|
let result = tokio::process::Command::new("/usr/local/bin/claw-store")
|
|
.args(args)
|
|
.output()
|
|
.await;
|
|
match result {
|
|
Ok(o) if o.status.success() => OkResponse { ok: true, error: None },
|
|
Ok(o) => OkResponse {
|
|
ok: false,
|
|
error: Some(String::from_utf8_lossy(&o.stderr).trim().to_string()),
|
|
},
|
|
Err(e) => OkResponse {
|
|
ok: false,
|
|
error: Some(e.to_string()),
|
|
},
|
|
}
|
|
}
|
|
|
|
// ── route handlers ────────────────────────────────────────────────────────────
|
|
|
|
async fn handle_status(State(state): State<Arc<AppState>>) -> Json<NodeStatus> {
|
|
let manifest = load_manifest(&state);
|
|
let cfg = &state.cfg;
|
|
|
|
let (warm_used_gb, warm_avail_gb) = zfs_used_avail(&cfg.warm.zfs_dataset);
|
|
let warm_snapshot_count = zfs_snapshot_count(&cfg.warm.zfs_dataset);
|
|
let zfs_pool_state = zfs_pool_state(&cfg.warm.zfs_dataset);
|
|
let hot_used = hot_used_gb(&manifest);
|
|
let (peer_host, reachable) = match &cfg.peer {
|
|
Some(p) => (Some(p.host.clone()), peer_reachable(p)),
|
|
None => (None, false),
|
|
};
|
|
|
|
let role = match cfg.node.role {
|
|
crate::config::NodeRole::Primary => "primary",
|
|
crate::config::NodeRole::Secondary => "secondary",
|
|
};
|
|
|
|
// Cheap file read; the queue is bounded by pending projects, not job
|
|
// history, so the file stays small in practice.
|
|
let queue = SyncQueue::load(&SyncQueue::default_path()).unwrap_or_default();
|
|
|
|
Json(NodeStatus {
|
|
node_name: cfg.node.name.clone(),
|
|
role: role.to_string(),
|
|
hot_used_gb: hot_used,
|
|
hot_max_gb: cfg.hot.max_gb,
|
|
hot_path: cfg.hot.path.to_string_lossy().to_string(),
|
|
warm_dataset: cfg.warm.zfs_dataset.clone(),
|
|
warm_used_gb,
|
|
warm_avail_gb,
|
|
warm_snapshot_count,
|
|
cold_path: cfg.cold.as_ref().map(|c| c.archive_path.to_string_lossy().to_string()),
|
|
peer_host,
|
|
peer_reachable: reachable,
|
|
zfs_pool_state,
|
|
active_project_count: manifest.projects.len(),
|
|
daemon_uptime_secs: daemon_uptime_secs(),
|
|
sync_queue_depth: queue.depth(),
|
|
sync_queue_stuck: queue.stuck_count(crate::sync::SYNC_STUCK_ATTEMPTS),
|
|
})
|
|
}
|
|
|
|
async fn handle_projects(State(state): State<Arc<AppState>>) -> Json<Vec<ProjectInfo>> {
|
|
let manifest = load_manifest(&state);
|
|
Json(build_projects_list(&state.cfg, &manifest))
|
|
}
|
|
|
|
/// Single source of truth for the project list shape. Used by both
|
|
/// `GET /api/projects` (the 10s dashboard poll) AND the SSE
|
|
/// `/api/events` stream — without this, the two endpoints disagreed on
|
|
/// whether `is_active` should include the cargo-config redirect heuristic.
|
|
/// The dashboard hit both every few seconds, the last-write-won, and rows
|
|
/// visually flashed as their `is_active` flipped between true and false.
|
|
fn build_projects_list(
|
|
cfg: &crate::config::Config,
|
|
manifest: &Manifest,
|
|
) -> Vec<ProjectInfo> {
|
|
let base = &cfg.warm.projects_path;
|
|
let mut projects: Vec<ProjectInfo> = Vec::new();
|
|
|
|
let mut push = |name: String, warm: std::path::PathBuf| {
|
|
let hot_path = cfg.hot.path.join(&name);
|
|
let size_bytes = du_bytes(&hot_path);
|
|
let active = manifest.get(&name);
|
|
projects.push(ProjectInfo {
|
|
name: name.clone(),
|
|
warm_path: warm.to_string_lossy().to_string(),
|
|
hot_target_path: hot_path.to_string_lossy().to_string(),
|
|
// Treat manifest membership AS WELL AS a live cargo-config
|
|
// redirect as "active" — both paths through the codebase
|
|
// used to compute this differently; consolidating here
|
|
// stops the dashboard rows from flashing.
|
|
is_active: active.is_some() || has_hot_cargo_config(&warm, &hot_path),
|
|
hot_size_mb: size_bytes as f64 / 1_048_576.0,
|
|
last_build: active.and_then(|p| p.last_build),
|
|
last_active: active.and_then(|p| p.last_active),
|
|
last_sync: active.and_then(|p| p.last_sync),
|
|
is_building: is_building(&warm),
|
|
git_branch: git_branch(&warm),
|
|
git_ahead: git_ahead(&warm),
|
|
});
|
|
};
|
|
|
|
if let Ok(top_entries) = std::fs::read_dir(base) {
|
|
for org_entry in top_entries.flatten() {
|
|
let org_path = org_entry.path();
|
|
if !org_path.is_dir() { continue; }
|
|
// Skip all symlinks: covers Gitea's repo.git→repo aliases AND
|
|
// cross-org shortcuts (e.g. clawverse/clawhdf5→../quantumclaw/clawhdf5)
|
|
// that resolve to the same git tree and would produce duplicates.
|
|
if org_entry.file_type().map(|t| t.is_symlink()).unwrap_or(false) { continue; }
|
|
let org = org_entry.file_name().to_string_lossy().to_string();
|
|
|
|
if org_path.join(".git").exists() {
|
|
push(org, org_path);
|
|
} else if let Ok(repos) = std::fs::read_dir(&org_path) {
|
|
for repo_entry in repos.flatten() {
|
|
let repo_path = repo_entry.path();
|
|
if !repo_path.is_dir() || !repo_path.join(".git").exists() { continue; }
|
|
if repo_entry.file_type().map(|t| t.is_symlink()).unwrap_or(false) { continue; }
|
|
let repo = repo_entry.file_name().to_string_lossy().to_string();
|
|
let key = format!("{}/{}", org, repo);
|
|
push(key, repo_path);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
projects.sort_by(|a, b| a.name.cmp(&b.name));
|
|
projects
|
|
}
|
|
|
|
async fn handle_snapshots(State(state): State<Arc<AppState>>) -> Json<Vec<SnapshotInfo>> {
|
|
let cfg = &state.cfg;
|
|
let dataset = &cfg.warm.zfs_dataset;
|
|
|
|
let out = std::process::Command::new("zfs")
|
|
.args(["list", "-H", "-p", "-t", "snapshot", "-o", "name,used,refer", "-r", dataset])
|
|
.output();
|
|
|
|
let mut snapshots = Vec::new();
|
|
if let Ok(o) = out {
|
|
let text = String::from_utf8_lossy(&o.stdout);
|
|
for line in text.lines() {
|
|
let parts: Vec<&str> = line.split('\t').collect();
|
|
if parts.len() < 3 { continue; }
|
|
let name = parts[0].trim().to_string();
|
|
let used_bytes = parts[1].trim().parse::<u64>().unwrap_or(0);
|
|
let refer_bytes = parts[2].trim().parse::<u64>().unwrap_or(0);
|
|
let kind = parse_snapshot_kind(&name);
|
|
let timestamp = parse_snapshot_timestamp(&name);
|
|
snapshots.push(SnapshotInfo { name, kind, timestamp, used_bytes, refer_bytes });
|
|
}
|
|
}
|
|
|
|
Json(snapshots)
|
|
}
|
|
|
|
async fn handle_sync_queue(State(_state): State<Arc<AppState>>) -> Json<Vec<crate::sync::SyncJob>> {
|
|
let queue = SyncQueue::load(&SyncQueue::default_path()).unwrap_or_default();
|
|
Json(queue.jobs)
|
|
}
|
|
|
|
async fn handle_hot(State(state): State<Arc<AppState>>) -> Json<Vec<HotEntry>> {
|
|
let hot_base = &state.cfg.hot.path;
|
|
let mut entries = Vec::new();
|
|
|
|
if let Ok(top) = std::fs::read_dir(hot_base) {
|
|
for org_entry in top.flatten() {
|
|
let org_path = org_entry.path();
|
|
if !org_path.is_dir() { continue; }
|
|
let org_name = org_entry.file_name().to_string_lossy().to_string();
|
|
|
|
if let Ok(repos) = std::fs::read_dir(&org_path) {
|
|
for repo_entry in repos.flatten() {
|
|
let repo_path = repo_entry.path();
|
|
if !repo_path.is_dir() { continue; }
|
|
let repo_name = repo_entry.file_name().to_string_lossy().to_string();
|
|
let project = format!("{}/{}", org_name, repo_name);
|
|
let size_bytes = du_bytes(&repo_path);
|
|
let last_modified = last_modified_time(&repo_path);
|
|
entries.push(HotEntry {
|
|
project,
|
|
size_mb: size_bytes as f64 / 1_048_576.0,
|
|
last_modified,
|
|
});
|
|
}
|
|
} else {
|
|
// Flat layout — org_path itself is the project dir
|
|
let size_bytes = du_bytes(&org_path);
|
|
let last_modified = last_modified_time(&org_path);
|
|
entries.push(HotEntry {
|
|
project: org_name,
|
|
size_mb: size_bytes as f64 / 1_048_576.0,
|
|
last_modified,
|
|
});
|
|
}
|
|
}
|
|
}
|
|
|
|
entries.sort_by(|a, b| a.project.cmp(&b.project));
|
|
Json(entries)
|
|
}
|
|
|
|
async fn handle_activate(
|
|
State(_state): State<Arc<AppState>>,
|
|
Json(body): Json<ProjectBody>,
|
|
) -> Json<OkResponse> {
|
|
if let Err(e) = validate_project_name(&body.project) {
|
|
return Json(OkResponse { ok: false, error: Some(e) });
|
|
}
|
|
Json(run_claw_store_cmd(&["activate", &body.project]).await)
|
|
}
|
|
|
|
async fn handle_deactivate(
|
|
State(_state): State<Arc<AppState>>,
|
|
Json(body): Json<ProjectBody>,
|
|
) -> Json<OkResponse> {
|
|
if let Err(e) = validate_project_name(&body.project) {
|
|
return Json(OkResponse { ok: false, error: Some(e) });
|
|
}
|
|
Json(run_claw_store_cmd(&["deactivate", &body.project]).await)
|
|
}
|
|
|
|
async fn handle_sync(
|
|
State(_state): State<Arc<AppState>>,
|
|
Json(body): Json<ProjectBody>,
|
|
) -> Json<OkResponse> {
|
|
if let Err(e) = validate_project_name(&body.project) {
|
|
return Json(OkResponse { ok: false, error: Some(e) });
|
|
}
|
|
Json(run_claw_store_cmd(&["sync", &body.project]).await)
|
|
}
|
|
|
|
async fn handle_gc(State(_state): State<Arc<AppState>>) -> Json<OkResponse> {
|
|
Json(run_claw_store_cmd(&["gc"]).await)
|
|
}
|
|
|
|
async fn handle_snapshot(State(_state): State<Arc<AppState>>) -> Json<OkResponse> {
|
|
Json(run_claw_store_cmd(&["snapshot"]).await)
|
|
}
|
|
|
|
// ── SSE event stream ──────────────────────────────────────────────────────────
|
|
|
|
#[derive(Serialize)]
|
|
struct SsePayload {
|
|
status: NodeStatus,
|
|
projects: Vec<ProjectInfo>,
|
|
}
|
|
|
|
async fn handle_events(
|
|
State(state): State<Arc<AppState>>,
|
|
) -> Sse<impl tokio_stream::Stream<Item = Result<axum::response::sse::Event, Infallible>>> {
|
|
let interval = tokio::time::interval(Duration::from_secs(5));
|
|
let stream = IntervalStream::new(interval).map(move |_| {
|
|
let manifest = load_manifest(&state);
|
|
let cfg = &state.cfg;
|
|
|
|
// Build status
|
|
let (warm_used_gb, warm_avail_gb) = zfs_used_avail(&cfg.warm.zfs_dataset);
|
|
let warm_snapshot_count = zfs_snapshot_count(&cfg.warm.zfs_dataset);
|
|
let zfs_pool_state = zfs_pool_state(&cfg.warm.zfs_dataset);
|
|
let hot_used = hot_used_gb(&manifest);
|
|
let (peer_host, reachable) = match &cfg.peer {
|
|
Some(p) => (Some(p.host.clone()), peer_reachable(p)),
|
|
None => (None, false),
|
|
};
|
|
let role = match cfg.node.role {
|
|
crate::config::NodeRole::Primary => "primary",
|
|
crate::config::NodeRole::Secondary => "secondary",
|
|
};
|
|
|
|
let queue = SyncQueue::load(&SyncQueue::default_path()).unwrap_or_default();
|
|
let status = NodeStatus {
|
|
node_name: cfg.node.name.clone(),
|
|
role: role.to_string(),
|
|
hot_used_gb: hot_used,
|
|
hot_max_gb: cfg.hot.max_gb,
|
|
hot_path: cfg.hot.path.to_string_lossy().to_string(),
|
|
warm_dataset: cfg.warm.zfs_dataset.clone(),
|
|
warm_used_gb,
|
|
warm_avail_gb,
|
|
warm_snapshot_count,
|
|
cold_path: cfg.cold.as_ref().map(|c| c.archive_path.to_string_lossy().to_string()),
|
|
peer_host,
|
|
peer_reachable: reachable,
|
|
zfs_pool_state,
|
|
active_project_count: manifest.projects.len(),
|
|
daemon_uptime_secs: daemon_uptime_secs(),
|
|
sync_queue_depth: queue.depth(),
|
|
sync_queue_stuck: queue.stuck_count(crate::sync::SYNC_STUCK_ATTEMPTS),
|
|
};
|
|
|
|
// v0.2.1 — single-source the project list builder. The previous
|
|
// inline duplicate computed `is_active = active.is_some()` while
|
|
// `handle_projects` also OR'd in `has_hot_cargo_config(...)` —
|
|
// so the same project flipped active/inactive every few seconds
|
|
// as the dashboard alternated between the SSE and HTTP polls.
|
|
let projects = build_projects_list(cfg, &manifest);
|
|
|
|
let payload = SsePayload { status, projects };
|
|
let data = serde_json::to_string(&payload).unwrap_or_else(|_| "{}".to_string());
|
|
Ok(axum::response::sse::Event::default().data(data))
|
|
});
|
|
|
|
Sse::new(stream).keep_alive(
|
|
axum::response::sse::KeepAlive::new()
|
|
.interval(Duration::from_secs(15))
|
|
.text(":keep-alive"),
|
|
)
|
|
}
|
|
|
|
// ── auth middleware ───────────────────────────────────────────────────────────
|
|
|
|
/// If `cfg.api_token` is set, all non-GET requests must supply a matching
|
|
/// `Authorization: Bearer <token>` header. GET requests are always allowed
|
|
/// so the dashboard can load without credentials.
|
|
async fn auth_middleware(
|
|
State(state): State<Arc<AppState>>,
|
|
request: axum::extract::Request,
|
|
next: axum::middleware::Next,
|
|
) -> axum::response::Response {
|
|
if let Some(expected) = &state.cfg.api_token {
|
|
if request.method() != axum::http::Method::GET {
|
|
let provided = request
|
|
.headers()
|
|
.get(axum::http::header::AUTHORIZATION)
|
|
.and_then(|v| v.to_str().ok())
|
|
.and_then(|v| v.strip_prefix("Bearer "));
|
|
if provided != Some(expected.as_str()) {
|
|
return axum::response::Response::builder()
|
|
.status(axum::http::StatusCode::UNAUTHORIZED)
|
|
.header("content-type", "application/json")
|
|
.body(axum::body::Body::from(r#"{"ok":false,"error":"unauthorized"}"#))
|
|
.unwrap_or_default();
|
|
}
|
|
}
|
|
}
|
|
next.run(request).await
|
|
}
|
|
|
|
// ── server entry point ────────────────────────────────────────────────────────
|
|
|
|
/// Builds the Axum Router. Extracted so tests can call it without binding a port.
|
|
pub fn build_app(
|
|
cfg: Config,
|
|
manifest_path: PathBuf,
|
|
static_dir: Option<PathBuf>,
|
|
) -> Router {
|
|
build_app_with_v2(cfg, manifest_path, static_dir, None)
|
|
}
|
|
|
|
pub fn build_app_with_v2(
|
|
cfg: Config,
|
|
manifest_path: PathBuf,
|
|
static_dir: Option<PathBuf>,
|
|
v2_static_dir: Option<PathBuf>,
|
|
) -> Router {
|
|
let state = Arc::new(AppState { cfg, manifest_path });
|
|
|
|
let cors = CorsLayer::new()
|
|
.allow_origin(Any)
|
|
.allow_methods([
|
|
axum::http::Method::GET,
|
|
axum::http::Method::POST,
|
|
])
|
|
.allow_headers([axum::http::header::CONTENT_TYPE]);
|
|
|
|
let mut api: Router<Arc<AppState>> = Router::new()
|
|
.route("/api/status", get(handle_status))
|
|
.route("/api/projects", get(handle_projects))
|
|
.route("/api/snapshots", get(handle_snapshots))
|
|
.route("/api/sync-queue", get(handle_sync_queue))
|
|
.route("/api/hot", get(handle_hot))
|
|
.route("/api/activate", post(handle_activate))
|
|
.route("/api/deactivate", post(handle_deactivate))
|
|
.route("/api/sync", post(handle_sync))
|
|
.route("/api/gc", post(handle_gc))
|
|
.route("/api/snapshot", post(handle_snapshot))
|
|
.route("/api/events", get(handle_events))
|
|
.route_layer(axum::middleware::from_fn_with_state(state.clone(), auth_middleware));
|
|
|
|
if let Some(dir) = static_dir {
|
|
api = api.fallback_service(tower_http::services::ServeDir::new(dir));
|
|
}
|
|
|
|
// dashboard-v2 aggregator backend (docs/dashboard-v2.md).
|
|
// Builds only when [cluster] + [cluster.tls] + peers are
|
|
// present; otherwise v2 API routes are skipped, and the /v2/
|
|
// static mount (if any) still works so operators can see the
|
|
// config-missing error message the SPA renders.
|
|
let v2_state = crate::serve_v2::V2State::from_config(&state.cfg)
|
|
.map(std::sync::Arc::new)
|
|
.ok();
|
|
let mut v2_router: Router<()> = Router::new();
|
|
if let Some(v2s) = v2_state {
|
|
v2_router = v2_router.merge(crate::serve_v2::build(v2s));
|
|
}
|
|
if let Some(dir) = v2_static_dir {
|
|
v2_router = v2_router.nest_service(
|
|
"/v2",
|
|
tower_http::services::ServeDir::new(&dir).fallback(
|
|
tower_http::services::ServeFile::new(dir.join("index.html")),
|
|
),
|
|
);
|
|
}
|
|
|
|
Router::new()
|
|
.merge(api.layer(cors.clone()).with_state(state))
|
|
.merge(v2_router.layer(cors))
|
|
}
|
|
|
|
pub async fn run_server(
|
|
cfg: Config,
|
|
_manifest: Manifest,
|
|
port: u16,
|
|
static_dir: Option<PathBuf>,
|
|
v2_static_dir: Option<PathBuf>,
|
|
) -> Result<()> {
|
|
let manifest_path = Manifest::default_path();
|
|
let app = build_app_with_v2(cfg, manifest_path, static_dir, v2_static_dir);
|
|
|
|
let addr = std::net::SocketAddr::from(([0, 0, 0, 0], port));
|
|
tracing::info!("claw-store API server listening on {}", addr);
|
|
|
|
let listener = tokio::net::TcpListener::bind(addr).await?;
|
|
axum::serve(listener, app).await?;
|
|
Ok(())
|
|
}
|
|
|
|
// ── tests ─────────────────────────────────────────────────────────────────────
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use axum::{
|
|
body::Body,
|
|
http::{Request, StatusCode},
|
|
};
|
|
use tempfile::NamedTempFile;
|
|
use tower::ServiceExt; // for .oneshot()
|
|
|
|
fn test_cfg(api_token: Option<&str>) -> Config {
|
|
let token_line = match api_token {
|
|
Some(t) => format!("api_token = \"{}\"\n", t),
|
|
None => String::new(),
|
|
};
|
|
// api_token is a root-level key — it must appear before any [section] header
|
|
toml::from_str(&format!(
|
|
r#"
|
|
{token_line}
|
|
[node]
|
|
name = "test-node"
|
|
role = "primary"
|
|
[hot]
|
|
path = "/tmp/claw-test-hot"
|
|
max_gb = 10
|
|
stale_hours = 48
|
|
[warm]
|
|
projects_path = "/tmp/claw-test-warm"
|
|
zfs_dataset = "test/dataset"
|
|
snapshot_retain_hours = 24
|
|
snapshot_retain_days = 7
|
|
snapshot_retain_weeks = 4
|
|
"#
|
|
))
|
|
.unwrap()
|
|
}
|
|
|
|
async fn body_bytes(body: Body) -> Vec<u8> {
|
|
use http_body_util::BodyExt;
|
|
body.collect().await.unwrap().to_bytes().to_vec()
|
|
}
|
|
|
|
// ── pure-function tests ───────────────────────────────────────────────────
|
|
|
|
#[test]
|
|
fn test_validate_project_name_accepts_valid() {
|
|
assert!(validate_project_name("org/repo").is_ok());
|
|
assert!(validate_project_name("my-org/my-repo").is_ok());
|
|
assert!(validate_project_name("Org123/Repo.name_v2").is_ok());
|
|
}
|
|
|
|
#[test]
|
|
fn test_validate_project_name_rejects_bad() {
|
|
assert!(validate_project_name("no-slash").is_err());
|
|
assert!(validate_project_name("../../etc/passwd").is_err());
|
|
assert!(validate_project_name("; rm -rf /").is_err());
|
|
assert!(validate_project_name("a/b/c").is_err());
|
|
assert!(validate_project_name("org/").is_err());
|
|
assert!(validate_project_name("/repo").is_err());
|
|
}
|
|
|
|
#[test]
|
|
fn test_daemon_uptime_absent() {
|
|
if std::path::Path::new(crate::daemon::DAEMON_STARTED_PATH).exists() {
|
|
return; // daemon is live — skip rather than assert a live value
|
|
}
|
|
assert_eq!(daemon_uptime_secs(), 0);
|
|
}
|
|
|
|
// ── HTTP handler tests ────────────────────────────────────────────────────
|
|
|
|
#[tokio::test]
|
|
async fn test_get_status_returns_200_json() {
|
|
let tmp = NamedTempFile::new().unwrap();
|
|
let app = build_app(test_cfg(None), tmp.path().to_path_buf(), None);
|
|
let req = Request::builder()
|
|
.uri("/api/status")
|
|
.body(Body::empty())
|
|
.unwrap();
|
|
let resp = app.oneshot(req).await.unwrap();
|
|
assert_eq!(resp.status(), StatusCode::OK);
|
|
let bytes = body_bytes(resp.into_body()).await;
|
|
let val: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
|
|
assert_eq!(val["node_name"], "test-node");
|
|
assert_eq!(val["role"], "primary");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_get_projects_returns_200_array() {
|
|
let tmp = NamedTempFile::new().unwrap();
|
|
let app = build_app(test_cfg(None), tmp.path().to_path_buf(), None);
|
|
let req = Request::builder()
|
|
.uri("/api/projects")
|
|
.body(Body::empty())
|
|
.unwrap();
|
|
let resp = app.oneshot(req).await.unwrap();
|
|
assert_eq!(resp.status(), StatusCode::OK);
|
|
let bytes = body_bytes(resp.into_body()).await;
|
|
let val: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
|
|
assert!(val.is_array(), "expected JSON array, got: {}", val);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_activate_invalid_name_returns_error_json() {
|
|
let tmp = NamedTempFile::new().unwrap();
|
|
let app = build_app(test_cfg(None), tmp.path().to_path_buf(), None);
|
|
let body = serde_json::json!({"project": "../../bad"}).to_string();
|
|
let req = Request::builder()
|
|
.method("POST")
|
|
.uri("/api/activate")
|
|
.header("content-type", "application/json")
|
|
.body(Body::from(body))
|
|
.unwrap();
|
|
let resp = app.oneshot(req).await.unwrap();
|
|
assert_eq!(resp.status(), StatusCode::OK);
|
|
let bytes = body_bytes(resp.into_body()).await;
|
|
let val: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
|
|
assert_eq!(val["ok"], false);
|
|
assert!(val["error"].as_str().unwrap_or("").contains("invalid project name"));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_auth_no_header_returns_401() {
|
|
let tmp = NamedTempFile::new().unwrap();
|
|
let app = build_app(test_cfg(Some("secret123")), tmp.path().to_path_buf(), None);
|
|
let body = serde_json::json!({"project": "org/repo"}).to_string();
|
|
let req = Request::builder()
|
|
.method("POST")
|
|
.uri("/api/activate")
|
|
.header("content-type", "application/json")
|
|
.body(Body::from(body))
|
|
.unwrap();
|
|
let resp = app.oneshot(req).await.unwrap();
|
|
assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_auth_wrong_token_returns_401() {
|
|
let tmp = NamedTempFile::new().unwrap();
|
|
let app = build_app(test_cfg(Some("secret123")), tmp.path().to_path_buf(), None);
|
|
let body = serde_json::json!({"project": "org/repo"}).to_string();
|
|
let req = Request::builder()
|
|
.method("POST")
|
|
.uri("/api/activate")
|
|
.header("content-type", "application/json")
|
|
.header("authorization", "Bearer wrong-token")
|
|
.body(Body::from(body))
|
|
.unwrap();
|
|
let resp = app.oneshot(req).await.unwrap();
|
|
assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_auth_correct_token_passes_through() {
|
|
let tmp = NamedTempFile::new().unwrap();
|
|
let app = build_app(test_cfg(Some("secret123")), tmp.path().to_path_buf(), None);
|
|
let body = serde_json::json!({"project": "org/repo"}).to_string();
|
|
let req = Request::builder()
|
|
.method("POST")
|
|
.uri("/api/activate")
|
|
.header("content-type", "application/json")
|
|
.header("authorization", "Bearer secret123")
|
|
.body(Body::from(body))
|
|
.unwrap();
|
|
let resp = app.oneshot(req).await.unwrap();
|
|
// Auth passed — response is from the handler, not the middleware
|
|
assert_ne!(resp.status(), StatusCode::UNAUTHORIZED);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_get_requests_bypass_auth() {
|
|
// GET endpoints must be accessible even when a token is configured
|
|
let tmp = NamedTempFile::new().unwrap();
|
|
let app = build_app(test_cfg(Some("secret123")), tmp.path().to_path_buf(), None);
|
|
let req = Request::builder()
|
|
.uri("/api/status")
|
|
.body(Body::empty())
|
|
.unwrap();
|
|
let resp = app.oneshot(req).await.unwrap();
|
|
assert_eq!(resp.status(), StatusCode::OK);
|
|
}
|
|
}
|