diff --git a/src/dirs.rs b/src/dirs.rs index 1cb8c58..aecfcae 100644 --- a/src/dirs.rs +++ b/src/dirs.rs @@ -46,3 +46,44 @@ pub fn get_log_dir() -> Result { Ok(dir) } } + +/// Get the state directory for the lease file following ActivityWatch conventions. +/// Uses `~/.local/state/activitywatch/aw-notify/` on Linux, the data-local dir on +/// Windows, and `~/Library/Application Support/...` on macOS. +pub fn get_state_dir() -> Result { + #[cfg(target_os = "linux")] + { + let mut dir = ::dirs::state_dir() + .or_else(|| { + // Fall back to ~/.local/state if the dirs crate can't resolve it + ::dirs::home_dir().map(|h| h.join(".local").join("state")) + }) + .ok_or_else(|| anyhow!("Failed to get state dir"))?; + dir.push("activitywatch"); + dir.push("aw-notify"); + fs::create_dir_all(&dir)?; + Ok(dir) + } + + #[cfg(target_os = "windows")] + { + let mut dir = + ::dirs::data_local_dir().ok_or_else(|| anyhow!("Failed to get local data dir"))?; + dir.push("activitywatch"); + dir.push("aw-notify"); + fs::create_dir_all(&dir)?; + Ok(dir) + } + + #[cfg(not(any(target_os = "linux", target_os = "windows")))] + { + // macOS and other Unix-like systems + let mut dir = ::dirs::home_dir().ok_or_else(|| anyhow!("Failed to get home dir"))?; + dir.push("Library"); + dir.push("Application Support"); + dir.push("activitywatch"); + dir.push("aw-notify"); + fs::create_dir_all(&dir)?; + Ok(dir) + } +} diff --git a/src/lease.rs b/src/lease.rs new file mode 100644 index 0000000..19b2950 --- /dev/null +++ b/src/lease.rs @@ -0,0 +1,214 @@ +//! Background capability lease/receipt record for aw-notify. +//! +//! Implements the "Background Capability Lease Protocol" schema (design v1): +//! a tiny, local-first, JSON-serializable contract that distinguishes +//! **enabled**, **running**, and **actually-delivered** for an opt-in +//! background capability. See the schema design in the Bob workspace: +//! `knowledge/technical/background-capability-lease-schema.md`. +//! +//! The lease is a single JSON object written to a well-known local path +//! (`~/.local/state/activitywatch/aw-notify/lease.json` on Linux). Any monitor +//! can read it and test liveness with no scheduler and no server round-trip. + +use anyhow::{anyhow, Result}; +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; +use std::fs; +use std::path::PathBuf; + +use crate::dirs; + +/// Schema version. Bump on a breaking change; readers ignore unknown fields. +pub const SCHEMA_VERSION: u32 = 1; + +/// Heartbeat staleness threshold in seconds. A lease whose heartbeat is older +/// than this is `stale` (hung/crashed) rather than `running`. +pub const HEARTBEAT_EXPIRY_SECONDS: i64 = 120; + +/// The lease/receipt record for a single background capability. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct Lease { + pub capability: String, + pub schema_version: u32, + pub desired_state: String, + pub effective_state: String, + pub owner: Owner, + pub heartbeat: Heartbeat, + pub delivery_receipt: DeliveryReceipt, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct Owner { + pub process: String, + pub pid: u32, + pub hostname: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct Heartbeat { + pub last: DateTime, + pub expiry_after_seconds: i64, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct DeliveryReceipt { + pub last_successful: Option>, + pub last_payload: Option, +} + +impl Lease { + /// Create a fresh lease for a capability that is enabled and running. + pub fn new(capability: &str, process: &str, hostname: &str) -> Self { + let now = Utc::now(); + Lease { + capability: capability.to_string(), + schema_version: SCHEMA_VERSION, + desired_state: "enabled".to_string(), + effective_state: "running".to_string(), + owner: Owner { + process: process.to_string(), + pid: std::process::id(), + hostname: hostname.to_string(), + }, + heartbeat: Heartbeat { + last: now, + expiry_after_seconds: HEARTBEAT_EXPIRY_SECONDS, + }, + delivery_receipt: DeliveryReceipt { + last_successful: None, + last_payload: None, + }, + } + } + + // Read-side contract for a monitor process; not consumed by the writer binary + // itself, so clippy sees it as dead code here. + #[allow(dead_code)] + /// True if the heartbeat is fresh (now - last < expiry). + pub fn is_running(&self) -> bool { + let now = Utc::now(); + (now - self.heartbeat.last).num_seconds() < self.heartbeat.expiry_after_seconds + } + + // Read-side contract for a monitor process. + #[allow(dead_code)] + /// True if the lease is stale (heartbeat expired) — a hung/crashed process. + pub fn is_stale(&self) -> bool { + !self.is_running() + } + + /// Refresh the heartbeat timestamp. + pub fn touch(&mut self) { + self.heartbeat.last = Utc::now(); + } + + /// Record a successful delivery. `last_successful` is only ever advanced + /// by a success — a failed delivery never moves it (the edge case in the + /// schema: last-successful, not last-attempted). + pub fn record_delivery(&mut self, payload: serde_json::Value) { + self.delivery_receipt.last_successful = Some(Utc::now()); + self.delivery_receipt.last_payload = Some(payload); + } + + // Read-side contract for a monitor process. + #[allow(dead_code)] + /// Mark the capability as cleanly stopped (removes the lease on write). + pub fn mark_stopped(&mut self) { + self.effective_state = "stopped".to_string(); + } +} + +/// Path to the lease file for this capability. +pub fn lease_path() -> Result { + Ok(dirs::get_state_dir()?.join("lease.json")) +} + +/// Write the lease to disk (atomic: write temp then rename). +pub fn write_lease(lease: &Lease) -> Result<()> { + let path = lease_path()?; + let json = serde_json::to_string_pretty(lease)?; + let tmp = path.with_extension("json.tmp"); + fs::write(&tmp, json)?; + fs::rename(&tmp, &path)?; + Ok(()) +} + +// Read-side contract for a monitor process; the writer binary itself never calls +// this, so clippy sees it as dead code here. +#[allow(dead_code)] +/// Read the lease from disk, if present. +pub fn read_lease() -> Result> { + let path = lease_path()?; + if !path.exists() { + return Ok(None); + } + let json = fs::read_to_string(&path)?; + let lease: Lease = serde_json::from_str(&json) + .map_err(|e| anyhow!("failed to parse lease at {}: {}", path.display(), e))?; + Ok(Some(lease)) +} + +/// Remove the lease file (used on clean shutdown when the capability stops). +pub fn remove_lease() -> Result<()> { + let path = lease_path()?; + if path.exists() { + fs::remove_file(&path)?; + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn test_lease() -> Lease { + Lease::new("aw-notify", "aw-notify", "testhost") + } + + #[test] + fn fresh_lease_is_running() { + let lease = test_lease(); + assert!(lease.is_running()); + assert!(!lease.is_stale()); + assert_eq!(lease.desired_state, "enabled"); + assert_eq!(lease.effective_state, "running"); + } + + #[test] + fn stale_lease_detected() { + let mut lease = test_lease(); + // Backdate the heartbeat beyond the expiry window. + lease.heartbeat.last = Utc::now() - chrono::Duration::seconds(HEARTBEAT_EXPIRY_SECONDS + 1); + assert!(lease.is_stale()); + assert!(!lease.is_running()); + } + + #[test] + fn touch_refreshes_heartbeat() { + let mut lease = test_lease(); + lease.heartbeat.last = Utc::now() - chrono::Duration::seconds(HEARTBEAT_EXPIRY_SECONDS + 1); + assert!(lease.is_stale()); + lease.touch(); + assert!(lease.is_running()); + } + + #[test] + fn delivery_receipt_is_last_successful() { + let mut lease = test_lease(); + assert!(lease.delivery_receipt.last_successful.is_none()); + lease.record_delivery(serde_json::json!({"notification_type": "threshold"})); + assert!(lease.delivery_receipt.last_successful.is_some()); + assert_eq!( + lease.delivery_receipt.last_payload, + Some(serde_json::json!({"notification_type": "threshold"})) + ); + } + + #[test] + fn roundtrip_serialization() { + let lease = test_lease(); + let json = serde_json::to_string(&lease).unwrap(); + let parsed: Lease = serde_json::from_str(&json).unwrap(); + assert_eq!(parsed, lease); + } +} diff --git a/src/main.rs b/src/main.rs index 443d719..e9bb3d0 100644 --- a/src/main.rs +++ b/src/main.rs @@ -20,17 +20,22 @@ use std::collections::HashMap; use std::fs; use std::io::{self, Read, Write}; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::OnceLock; +use std::sync::{Mutex, OnceLock}; use std::thread; use std::time; mod dirs; +mod lease; mod logging; static AW_CLIENT: OnceLock = OnceLock::new(); static HOSTNAME: OnceLock = OnceLock::new(); static SERVER_AVAILABLE: AtomicBool = AtomicBool::new(true); static OUTPUT_ONLY: AtomicBool = AtomicBool::new(false); + +/// Global handle to the capability lease, so the worker thread and main loop +/// can refresh the heartbeat and record deliveries. Set once at service start. +static LEASE: OnceLock> = OnceLock::new(); static NOTIFICATION_TX: OnceLock> = OnceLock::new(); // Host/port defaults live in aw-notify-client so the client and server can't @@ -365,6 +370,17 @@ fn run_app(cli: Cli) -> Result<()> { fn start_service(hostname: String, config: NotificationConfig) -> Result<()> { log::info!("Starting notification service..."); + // Write the background-capability lease: enabled + running, with a fresh + // heartbeat. Removed on clean shutdown (see below). + let lease = lease::Lease::new("aw-notify", "aw-notify", &hostname); + if let Err(e) = lease::write_lease(&lease) { + log::warn!( + "Failed to write capability lease: {} (continuing anyway)", + e + ); + } + LEASE.set(Mutex::new(lease)).ok(); + // Initialize notification queue and worker thread (unbounded queue) let (notification_tx, notification_rx) = unbounded::(); NOTIFICATION_TX @@ -452,6 +468,12 @@ fn start_service(hostname: String, config: NotificationConfig) -> Result<()> { // Give background threads a moment to finish cleanup thread::sleep(time::Duration::from_millis(100)); + // Clean shutdown: remove the lease so a monitor sees `stopped` (no lease) + // rather than `stale`. Best-effort; never fatal. + if let Err(e) = lease::remove_lease() { + log::warn!("Failed to remove capability lease on shutdown: {}", e); + } + log::info!("Shutdown complete"); result } @@ -623,6 +645,17 @@ fn threshold_alerts(shutdown_rx: Receiver<()>, alert_configs: Vec) } } + // Refresh the lease heartbeat so a monitor can tell this process is + // alive (running) rather than stale. Best-effort; never fatal. + if let Some(lease) = LEASE.get() { + if let Ok(mut lease) = lease.lock() { + lease.touch(); + if let Err(e) = lease::write_lease(&lease) { + log::warn!("Failed to refresh lease heartbeat: {}", e); + } + } + } + // Wait for shutdown signal or timeout (10 seconds for normal monitoring) match shutdown_rx.recv_timeout(time::Duration::from_secs(10)) { Ok(_) => { @@ -1553,6 +1586,20 @@ fn display_notification(title: &str, message: &str, sender: Option<&str>) -> Res .timeout(5000) .show()?; + // Record a successful delivery in the lease (last-successful, not + // last-attempted). Best-effort; a failed write must not fail the notify. + if let Some(lease) = LEASE.get() { + if let Ok(mut lease) = lease.lock() { + lease.record_delivery(serde_json::json!({ + "title": title, + "sender": sender, + })); + if let Err(e) = lease::write_lease(&lease) { + log::warn!("Failed to record delivery receipt: {}", e); + } + } + } + Ok(()) }