bestool 2.1.2

//! Backup run registry and the on-demand `run` endpoint.
//!
//! A backup of a given type runs at most once at a time inside the daemon, and
//! across types the daemon runs at most one backup at a time: a batch of due
//! backups is linearised through a single run slot, with a quiet period between
//! consecutive runs, rather than starting together and loading the server.
//! [`BackupRegistry::ensure_run`] is start-or-attach: it starts a run via the
//! injected [`BackupRunner`] when the type is idle, or hands back a subscription
//! to the in-flight run otherwise. A subscriber sees a replay of the latest
//! status, then live status events and a periodic heartbeat, then the run's
//! terminal event. [`BackupRegistry::running`] lists in-flight runs for the
//! daemon's status.
//!
//! This serialisation is a daemon concern only; a manual `bestool canopy backup`
//! invocation drives the backup driver directly and is bounded just by the
//! cross-process lock, not by this slot.
//!
//! The actual backup driver lives in the bestool binary; it's injected here as a
//! [`BackupRunner`] so this crate carries no backup logic of its own.

use std::{
	collections::HashMap,
	sync::{
		Arc,
		atomic::{AtomicBool, Ordering},
	},
	time::Duration,
};

use futures::{StreamExt, future::BoxFuture, stream::BoxStream};
use jiff::Timestamp;
use miette::Result;
use serde_json::{Value, json};
use tokio::sync::{Mutex, Semaphore, broadcast, mpsc};
use tokio_stream::wrappers::BroadcastStream;

use crate::alertd::{
	BackgroundTask, TaskContext, TaskEndpoint, TaskEndpointResponse, tasks::TaskEndpointHandler,
};

/// Runs one backup of `backup_type`, emitting JSON status events into the sink
/// and resolving when the run finishes. Injected by the binary that owns the
/// backup driver. The runner must emit a terminal `done`/`error` event before it
/// returns, so attached clients always see an end.
///
/// The hold switch is shared with the running driver rather than passed by
/// value, so an instruction that arrives after the run started still reaches it:
/// the driver reads the switch when it finishes with its capture, not when it
/// takes one.
pub type BackupRunner = Arc<
	dyn Fn(String, mpsc::UnboundedSender<Value>, Arc<AtomicBool>) -> BoxFuture<'static, ()>
		+ Send
		+ Sync
		+ 'static,
>;

const HEARTBEAT: Duration = Duration::from_secs(5);
const BROADCAST_CAPACITY: usize = 256;
/// Quiet gap held after each backup before the next queued one may start, so a
/// batch of due backups runs spread out rather than hammering the server
/// together. Deliberately coarse for now; tune once we have load data.
const QUIET_PERIOD: Duration = Duration::from_secs(30);

struct RunHandle {
	started_at: Timestamp,
	run_id: Mutex<Option<String>>,
	/// Most recent status event, replayed to a late attacher so it isn't blank.
	latest: Mutex<Value>,
	events: broadcast::Sender<Value>,
	/// Whether this run should keep its capture. Kept here for the run's whole
	/// duration so it can be set while the run is under way.
	hold: Arc<AtomicBool>,
}

/// One in-flight backup, for the daemon's status.
#[derive(Clone, serde::Serialize, serde::Deserialize)]
pub struct RunningBackup {
	pub r#type: String,
	pub run_id: Option<String>,
	pub started_at: String,
	pub latest: Value,
}

/// Tracks in-flight backup runs and fans their status out to subscribers.
pub struct BackupRegistry {
	runner: BackupRunner,
	running: Mutex<HashMap<String, Arc<RunHandle>>>,
	/// The backup types configured on this host, refreshed by the capabilities
	/// task as it (re-)reads the backups dir. Surfaced in the daemon's status so
	/// an operator can see what's registered without listing the config dir.
	configured: Mutex<Vec<String>>,
	/// Daemon-wide single-run slot: at most one backup actually runs at a time,
	/// so a batch of due backups is linearised instead of starting together.
	run_slot: Semaphore,
	/// Quiet gap held (still occupying [`run_slot`]) after each run completes.
	quiet_period: Duration,
}

impl BackupRegistry {
	pub fn new(runner: BackupRunner) -> Arc<Self> {
		Self::with_quiet_period(runner, QUIET_PERIOD)
	}

	fn with_quiet_period(runner: BackupRunner, quiet_period: Duration) -> Arc<Self> {
		Arc::new(Self {
			runner,
			running: Mutex::new(HashMap::new()),
			configured: Mutex::new(Vec::new()),
			run_slot: Semaphore::new(1),
			quiet_period,
		})
	}

	/// Record the configured backup types (called by the capabilities task each
	/// time it reads the backups dir).
	pub async fn set_configured(&self, mut types: Vec<String>) {
		types.sort();
		*self.configured.lock().await = types;
	}

	/// The configured backup types, for the status endpoint.
	pub async fn configured(&self) -> Vec<String> {
		self.configured.lock().await.clone()
	}

	/// Start a run for `backup_type`, or attach to the one already in flight.
	/// Returns a stream of JSON status events ending with the terminal event.
	pub async fn ensure_run(
		self: &Arc<Self>,
		backup_type: String,
		hold: bool,
	) -> BoxStream<'static, Value> {
		let mut running = self.running.lock().await;

		if let Some(handle) = running.get(&backup_type) {
			// Attaching with a hold asked for sets it on the run already going —
			// the same instruction `hold` carries, arriving by another door. The
			// switch is never cleared this way: a hold is released deliberately.
			if hold {
				handle.hold.store(true, Ordering::Relaxed);
			}
			let attached = json!({
				"event": "attached",
				"runId": *handle.run_id.lock().await,
				"startedAt": handle.started_at.to_string(),
				"latest": *handle.latest.lock().await,
			});
			return subscription(Some(attached), handle.events.subscribe());
		}

		let (events, receiver) = broadcast::channel(BROADCAST_CAPACITY);
		let hold = Arc::new(AtomicBool::new(hold));
		let handle = Arc::new(RunHandle {
			started_at: Timestamp::now(),
			run_id: Mutex::new(None),
			latest: Mutex::new(json!({ "event": "starting" })),
			events: events.clone(),
			hold: hold.clone(),
		});
		running.insert(backup_type.clone(), handle.clone());
		drop(running);

		let (sink, run_rx) = mpsc::unbounded_channel::<Value>();
		let runner = (self.runner)(backup_type.clone(), sink, hold);
		let registry = self.clone();
		tokio::spawn(async move {
			// Wait for the daemon-wide run slot: only one backup runs at a time,
			// so multiple due backups queue here rather than loading the server
			// all at once. The type stays registered (and shows as running) while
			// queued, so a repeat request still attaches instead of double-starting.
			let queued = json!({ "event": "queued" });
			*handle.latest.lock().await = queued.clone();
			let _ = events.send(queued);
			let Ok(_permit) = registry.run_slot.acquire().await else {
				return; // semaphore closed; daemon shutting down
			};

			tokio::spawn(runner);
			registry
				.clone()
				.pump(backup_type, handle, run_rx, events)
				.await;

			// Hold the slot through a quiet period so the next queued backup
			// doesn't start back-to-back with this one.
			tokio::time::sleep(registry.quiet_period).await;
		});

		subscription(None, receiver)
	}

	/// Drain the runner's events into the broadcast (updating the handle's
	/// latest/run id), emitting a heartbeat between events, until the runner
	/// finishes; then deregister the type.
	async fn pump(
		self: Arc<Self>,
		backup_type: String,
		handle: Arc<RunHandle>,
		mut run_rx: mpsc::UnboundedReceiver<Value>,
		events: broadcast::Sender<Value>,
	) {
		let mut heartbeat = tokio::time::interval(HEARTBEAT);
		heartbeat.tick().await; // the first tick is immediate; skip it
		loop {
			tokio::select! {
				message = run_rx.recv() => match message {
					Some(event) => {
						if let Some(id) = event.get("runId").and_then(Value::as_str) {
							*handle.run_id.lock().await = Some(id.to_owned());
						}
						*handle.latest.lock().await = event.clone();
						let _ = events.send(event);
					}
					None => break, // runner finished
				},
				_ = heartbeat.tick() => {
					let _ = events.send(json!({ "event": "heartbeat" }));
				}
			}
		}
		self.running.lock().await.remove(&backup_type);
		// `events` drops here, closing subscribers after the terminal event.
	}

	/// Tell the in-flight run of `backup_type` to keep its capture, reporting
	/// whether there was one to tell.
	///
	/// This is what keeping the switch on the handle buys: a transfer that has
	/// already run for hours is not interrupted or restarted, and the run reads
	/// the switch when it finishes with its capture.
	pub async fn request_hold(&self, backup_type: &str) -> bool {
		match self.running.lock().await.get(backup_type) {
			Some(handle) => {
				handle.hold.store(true, Ordering::Relaxed);
				true
			}
			None => false,
		}
	}

	/// In-flight runs, for the status endpoint.
	pub async fn running(&self) -> Vec<RunningBackup> {
		let map = self.running.lock().await;
		let mut out = Vec::with_capacity(map.len());
		for (backup_type, handle) in map.iter() {
			out.push(RunningBackup {
				r#type: backup_type.clone(),
				run_id: handle.run_id.lock().await.clone(),
				started_at: handle.started_at.to_string(),
				latest: handle.latest.lock().await.clone(),
			});
		}
		out
	}
}

fn subscription(
	replay: Option<Value>,
	receiver: broadcast::Receiver<Value>,
) -> BoxStream<'static, Value> {
	// Drop lagged markers; end the stream when the broadcast closes.
	let live = BroadcastStream::new(receiver).filter_map(|item| async move { item.ok() });
	match replay {
		Some(value) => futures::stream::once(async move { value })
			.chain(live)
			.boxed(),
		None => live.boxed(),
	}
}

/// Exposes `GET /tasks/backup/run?type=X[&hold=true]` (start-or-attach,
/// streaming status), `GET /tasks/backup/running` (in-flight runs), and `GET
/// /tasks/backup/hold?type=X` (tell the in-flight run to keep its capture).
/// Holds no periodic work; the registry drives runs on demand.
pub struct BackupTask {
	registry: Arc<BackupRegistry>,
}

impl BackupTask {
	pub fn new(registry: Arc<BackupRegistry>) -> Self {
		Self { registry }
	}
}

impl BackgroundTask for BackupTask {
	fn name(&self) -> &'static str {
		"backup"
	}

	fn interval(&self) -> Duration {
		// On-demand only; no periodic work. A long interval keeps the watchdog
		// from treating idleness as a hang.
		Duration::from_secs(3600)
	}

	fn run<'a>(&'a self, _ctx: &'a TaskContext) -> BoxFuture<'a, Result<()>> {
		Box::pin(async { Ok(()) })
	}

	fn http_endpoints(&self) -> Vec<TaskEndpoint> {
		let run_handler: TaskEndpointHandler = {
			let registry = self.registry.clone();
			Arc::new(move |ctx| {
				let registry = registry.clone();
				Box::pin(async move {
					let Some(backup_type) = ctx.query.get("type").cloned() else {
						return TaskEndpointResponse::Error {
							status: 400,
							message: "missing ?type= query parameter".into(),
						};
					};
					let hold = ctx
						.query
						.get("hold")
						.is_some_and(|v| v == "true" || v == "1");
					TaskEndpointResponse::JsonLines(registry.ensure_run(backup_type, hold).await)
				})
			})
		};

		let hold_handler: TaskEndpointHandler = {
			let registry = self.registry.clone();
			Arc::new(move |ctx| {
				let registry = registry.clone();
				Box::pin(async move {
					let Some(backup_type) = ctx.query.get("type").cloned() else {
						return TaskEndpointResponse::Error {
							status: 400,
							message: "missing ?type= query parameter".into(),
						};
					};
					let held = registry.request_hold(&backup_type).await;
					TaskEndpointResponse::Json(json!({
						"type": backup_type,
						"held": held,
					}))
				})
			})
		};

		let running_handler: TaskEndpointHandler = {
			let registry = self.registry.clone();
			Arc::new(move |_ctx| {
				let registry = registry.clone();
				Box::pin(async move {
					TaskEndpointResponse::Json(json!({ "running": registry.running().await }))
				})
			})
		};

		vec![
			TaskEndpoint {
				name: "run",
				handler: run_handler,
			},
			TaskEndpoint {
				name: "running",
				handler: running_handler,
			},
			TaskEndpoint {
				name: "hold",
				handler: hold_handler,
			},
		]
	}
}

#[cfg(test)]
mod tests {
	use tokio::sync::Notify;

	use super::*;

	fn event(value: &Value) -> String {
		value
			.get("event")
			.and_then(Value::as_str)
			.unwrap_or_default()
			.to_owned()
	}

	#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
	async fn start_then_attach_then_finish() {
		// A runner that announces, waits for the gate, then finishes — so a
		// second request can attach while it's in flight.
		let gate = Arc::new(Notify::new());
		let runner: BackupRunner = {
			let gate = gate.clone();
			Arc::new(move |_type, sink: mpsc::UnboundedSender<Value>, _hold| {
				let gate = gate.clone();
				Box::pin(async move {
					let _ = sink.send(json!({ "event": "started", "runId": "r1" }));
					gate.notified().await;
					let _ = sink.send(json!({ "event": "done", "success": true }));
				})
			})
		};
		let registry = BackupRegistry::new(runner);

		let mut starter = registry.ensure_run("pg".into(), false).await;
		// The run queues for the daemon-wide slot (free here, so instantly) before
		// the runner emits its first event.
		assert_eq!(event(&starter.next().await.unwrap()), "queued");
		assert_eq!(event(&starter.next().await.unwrap()), "started");

		// A second request for the same type attaches rather than starting a
		// second run.
		let mut attacher = registry.ensure_run("pg".into(), false).await;
		assert_eq!(event(&attacher.next().await.unwrap()), "attached");
		assert_eq!(registry.running().await.len(), 1);

		// Release the run; both subscribers see the terminal event and end.
		gate.notify_one();
		let starter_events: Vec<String> = starter.map(|v| event(&v)).collect().await;
		assert!(starter_events.contains(&"done".to_owned()));
		let attacher_events: Vec<String> = attacher.map(|v| event(&v)).collect().await;
		assert!(attacher_events.contains(&"done".to_owned()));
	}

	#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
	async fn distinct_types_run_sequentially_not_concurrently() {
		use std::sync::atomic::{AtomicUsize, Ordering};

		// Each run records how many are executing at once; the slot must keep that
		// at 1 even though three types are triggered together.
		let concurrent = Arc::new(AtomicUsize::new(0));
		let max_seen = Arc::new(AtomicUsize::new(0));
		let runner: BackupRunner = {
			let concurrent = concurrent.clone();
			let max_seen = max_seen.clone();
			Arc::new(move |_type, sink: mpsc::UnboundedSender<Value>, _hold| {
				let concurrent = concurrent.clone();
				let max_seen = max_seen.clone();
				Box::pin(async move {
					let now = concurrent.fetch_add(1, Ordering::SeqCst) + 1;
					max_seen.fetch_max(now, Ordering::SeqCst);
					let _ = sink.send(json!({ "event": "started" }));
					tokio::time::sleep(Duration::from_millis(50)).await;
					concurrent.fetch_sub(1, Ordering::SeqCst);
					let _ = sink.send(json!({ "event": "done", "success": true }));
				})
			})
		};
		// No quiet period so the test stays fast; serialisation is the slot, not it.
		let registry = BackupRegistry::with_quiet_period(runner, Duration::ZERO);

		let streams = futures::future::join_all(
			["a", "b", "c"].map(|t| registry.ensure_run(t.to_owned(), false)),
		)
		.await;
		for mut stream in streams {
			while stream.next().await.is_some() {}
		}

		assert_eq!(
			max_seen.load(Ordering::SeqCst),
			1,
			"backups of distinct types must not overlap"
		);
	}

	/// The point of the feature: an instruction arriving after the run started is
	/// seen by that run, without interrupting what it is doing.
	#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
	async fn holding_reaches_a_run_already_in_flight() {
		let gate = Arc::new(Notify::new());
		// What the run observed when it finished with its capture.
		let observed = Arc::new(Mutex::new(None::<bool>));
		let runner: BackupRunner = {
			let gate = gate.clone();
			let observed = observed.clone();
			Arc::new(move |_type, sink: mpsc::UnboundedSender<Value>, hold| {
				let gate = gate.clone();
				let observed = observed.clone();
				Box::pin(async move {
					let _ = sink.send(json!({ "event": "started", "runId": "r1" }));
					gate.notified().await;
					// Read at the end, as the driver reads it at cleanup.
					*observed.lock().await = Some(hold.load(Ordering::Relaxed));
					let _ = sink.send(json!({ "event": "done", "success": true }));
				})
			})
		};
		let registry = BackupRegistry::with_quiet_period(runner, Duration::ZERO);

		// The run starts without a hold…
		let mut stream = registry.ensure_run("pg".into(), false).await;
		while let Some(value) = stream.next().await {
			if event(&value) == "started" {
				break;
			}
		}

		// …and is told to keep its capture while it is still going.
		assert!(
			registry.request_hold("pg").await,
			"the run should be reachable"
		);

		gate.notify_waiters();
		while stream.next().await.is_some() {}
		assert_eq!(
			*observed.lock().await,
			Some(true),
			"the run should have seen the hold requested mid-flight"
		);
	}

	/// Nothing in flight is a plain "no run to tell", not an error: the caller
	/// reports it to the operator, who has to reach for another option.
	#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
	async fn holding_an_idle_type_reports_no_run() {
		let runner: BackupRunner =
			Arc::new(move |_type, _sink: mpsc::UnboundedSender<Value>, _hold| Box::pin(async {}));
		let registry = BackupRegistry::with_quiet_period(runner, Duration::ZERO);
		assert!(!registry.request_hold("pg").await);
	}
}