Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 41 additions & 0 deletions src/dirs.rs
Original file line number Diff line number Diff line change
Expand Up @@ -46,3 +46,44 @@ pub fn get_log_dir() -> Result<PathBuf> {
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<PathBuf> {
#[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)
}
}
214 changes: 214 additions & 0 deletions src/lease.rs
Original file line number Diff line number Diff line change
@@ -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<Utc>,
pub expiry_after_seconds: i64,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct DeliveryReceipt {
pub last_successful: Option<DateTime<Utc>>,
pub last_payload: Option<serde_json::Value>,
}

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<PathBuf> {
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<Option<Lease>> {
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);
}
}
49 changes: 48 additions & 1 deletion src/main.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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<aw_client_rust::blocking::AwClient> = OnceLock::new();
static HOSTNAME: OnceLock<String> = 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<Mutex<lease::Lease>> = OnceLock::new();
static NOTIFICATION_TX: OnceLock<crossbeam_channel::Sender<QueuedNotification>> = OnceLock::new();

// Host/port defaults live in aw-notify-client so the client and server can't
Expand Down Expand Up @@ -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::<QueuedNotification>();
NOTIFICATION_TX
Expand Down Expand Up @@ -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
}
Expand Down Expand Up @@ -623,6 +645,17 @@ fn threshold_alerts(shutdown_rx: Receiver<()>, alert_configs: Vec<AlertConfig>)
}
}

// 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(_) => {
Expand Down Expand Up @@ -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(())
}

Expand Down
Loading