bestool 1.61.2

//! Config-driven, canopy-managed backups.
//!
//! A backup def (`/etc/bestool/backups/*.toml`) names a `type` (the Canopy
//! label), optional `pre`/`post` hooks and `tags`, and exactly one method
//! ([`method::Method`]). The driver fetches creds + target from Canopy, runs the
//! `pre` hooks, [`method::Method::prepare`]s a source, kopia-snapshots it with
//! the canopy-* tags, cleans up (always), runs `post`, and reports the outcome.
//!
//! The run logic lives in [`run_backup`] so the standalone `bestool canopy
//! backup` subcommand and (later) the in-process alertd trigger drive the same
//! code.

pub mod config;
pub mod hold;
pub mod method;
pub mod postgresql;
pub(super) mod progress;
pub mod provider;
mod secret_key;
mod simple;

use std::{
	collections::BTreeMap,
	path::Path,
	sync::{
		Arc,
		atomic::{AtomicBool, Ordering},
	},
	time::{Duration, Instant},
};

use bestool_canopy::{
	CanopyClient, DEFAULT_CANOPY_URL, TargetOutcome, connect_to,
	registration::Registration,
	schema::{BackupPurpose, ReportArgs, RunOutcome},
};
use bestool_kopia::{
	CacheLimits, CacheProfile, RunAs, S3Connection, S3KopiaEnv, args_policy_set_ignores,
	args_repository_connect_s3, args_snapshot_create,
	build_kopia_command_with_s3, find_kopia_binary,
	proxy::{self, RunningProxy, S3ProxyConfig},
};
use clap::Parser;
use miette::{Context as _, IntoDiagnostic as _, Result, bail, miette};
use reqwest::Url;
use tracing::{debug, error, info, warn};
use uuid::Uuid;

use self::{
	config::{BackupDef, Hook},
	provider::CanopyCredentialProvider,
};
use crate::actions::Context;

/// Run a configured backup, driving kopia and reporting to Canopy.
#[derive(Debug, Clone, Parser)]
pub struct BackupArgs {
	/// The backup type to run.
	///
	/// Must have a definition in the backups directory (a `*.toml` whose `type`
	/// matches).
	#[arg(long = "type", value_name = "TYPE")]
	pub backup_type: String,

	/// Override the registration directory (matching `register`/`export`).
	#[arg(long, value_name = "DIR")]
	pub config: Option<std::path::PathBuf>,

	/// Override the backups definition directory.
	#[arg(long, value_name = "DIR")]
	pub backups_dir: Option<std::path::PathBuf>,

	/// Run the backup in this process instead of delegating to the alertd daemon.
	///
	/// By default, when the daemon is running, the backup is run by it and its
	/// progress is streamed here; this forces a local run.
	#[arg(long)]
	pub no_daemon: bool,

	/// Keep the capture on this device after the run, as a local rollback point.
	///
	/// The run is otherwise an ordinary backup. Restoring from the kept capture
	/// costs a local copy instead of a full download, which is what makes it
	/// useful across an upgrade window that outlasts the upload.
	///
	/// A kept capture is released only by `bestool canopy hold drop`.
	#[arg(long)]
	pub hold: bool,

	/// Take the capture and stop there, without a repository.
	///
	/// Fetches no credentials, contacts no repository, and reports no run; the
	/// definition's pre/post hooks and the capture itself are as they would be for
	/// an uploading run. Runs in this process, since it needs no daemon.
	#[arg(long, requires = "hold")]
	pub no_upload: bool,
}

pub async fn run(args: BackupArgs, _ctx: Context) -> Result<()> {
	let run_local = || {
		run_backup(
			&args.backup_type,
			args.config.as_deref(),
			args.backups_dir.as_deref(),
			None,
			Arc::new(AtomicBool::new(args.hold)),
		)
	};

	// A capture-only run has nothing for the daemon to host — no credentials, no
	// transfer, no reporting — so it runs here. It still takes the per-type lock,
	// so it can't capture alongside a run the daemon is already driving.
	if args.no_upload {
		return capture_only(
			&args.backup_type,
			args.backups_dir.as_deref(),
		)
		.await;
	}

	if args.no_daemon {
		return run_local().await;
	}

	match run_via_daemon(&args.backup_type, args.hold).await {
		Ok(()) => Ok(()),
		Err(DaemonError::Failed(message)) => bail!("backup failed: {message}"),
		Err(DaemonError::Unreachable(err)) => {
			info!(%err, "alertd daemon not reachable; running the backup locally");
			run_local().await
		}
	}
}

/// Why a delegated run didn't yield a clean success.
pub(super) enum DaemonError {
	/// The daemon couldn't be reached (or doesn't expose the endpoint); the
	/// caller should run locally.
	Unreachable(String),
	/// The daemon ran the backup and it failed, or the stream was lost mid-run.
	Failed(String),
}

/// Tell the daemon's in-flight run of `backup_type` to keep its capture.
///
/// Reports whether there was a run to tell — a type with nothing in flight is a
/// plain "no", not a failure, since the operator's next move depends on which it
/// is.
pub(super) async fn request_daemon_hold(backup_type: &str) -> std::result::Result<bool, DaemonError> {
	let response = crate::http::daemon_get(
		&crate::http::client(),
		&format!("/tasks/backup/hold?type={backup_type}"),
		|r| r,
	)
	.await
	.map_err(|err| DaemonError::Unreachable(err.to_string()))?;
	if !response.status().is_success() {
		return Err(DaemonError::Unreachable(format!(
			"alertd returned {}",
			response.status()
		)));
	}
	let body: serde_json::Value = response
		.json()
		.await
		.map_err(|err| DaemonError::Failed(err.to_string()))?;
	Ok(body.get("held").and_then(serde_json::Value::as_bool).unwrap_or(false))
}

/// Ask the running daemon to run the backup (starting a run or attaching to one
/// already in flight) and render its streamed status. Returns `Unreachable` if
/// the daemon isn't there, so the caller can fall back to a local run.
async fn run_via_daemon(backup_type: &str, hold: bool) -> std::result::Result<(), DaemonError> {
	use futures::StreamExt as _;

	let response = crate::http::daemon_get(
		&crate::http::client(),
		&format!("/tasks/backup/run?type={backup_type}&hold={hold}"),
		|r| r,
	)
	.await
	.map_err(|err| DaemonError::Unreachable(err.to_string()))?;
	if !response.status().is_success() {
		return Err(DaemonError::Unreachable(format!(
			"alertd returned {}",
			response.status()
		)));
	}

	let mut stream = response.bytes_stream();
	let mut buffer = Vec::<u8>::new();
	let mut terminal: Option<std::result::Result<(), String>> = None;
	let mut on_progress_line = false;
	while let Some(chunk) = stream.next().await {
		let chunk = chunk.map_err(|err| DaemonError::Failed(format!("daemon stream: {err}")))?;
		buffer.extend_from_slice(&chunk);
		while let Some(nl) = buffer.iter().position(|&b| b == b'\n') {
			let line: Vec<u8> = buffer.drain(..=nl).collect();
			let Ok(event) = serde_json::from_slice::<serde_json::Value>(&line) else {
				continue;
			};
			if event.get("event").and_then(serde_json::Value::as_str) == Some("progress") {
				if let Some(status) = event.get("status").and_then(serde_json::Value::as_str) {
					render_progress(status, &mut on_progress_line);
				}
				continue;
			}
			// Finish the in-place progress line before logging the next event.
			if std::mem::take(&mut on_progress_line) {
				eprintln!();
			}
			// A chained run streams a terminal event per backup; a later
			// follower's success must not mask an earlier failure.
			if let Some(outcome) = render_daemon_event(backup_type, &event)
				&& !matches!(terminal, Some(Err(_)))
			{
				terminal = Some(outcome);
			}
		}
	}
	if on_progress_line {
		eprintln!();
	}

	match terminal {
		Some(Ok(())) => Ok(()),
		Some(Err(message)) => Err(DaemonError::Failed(message)),
		None => Err(DaemonError::Failed(
			"lost the connection to the daemon; the backup may still be running".into(),
		)),
	}
}

/// Render a live progress line: in place (overwriting) on a terminal, or as a
/// log line otherwise.
fn render_progress(status: &str, on_progress_line: &mut bool) {
	use std::io::{IsTerminal as _, Write as _};

	let mut stderr = std::io::stderr();
	if stderr.is_terminal() {
		// Carriage return + clear-to-end-of-line: overwrite the previous update.
		let _ = write!(stderr, "\r{status}\x1b[K");
		let _ = stderr.flush();
		*on_progress_line = true;
	} else {
		info!(%status, "backup progress");
	}
}

/// Render one streamed status event; returns `Some` for a terminal event.
fn render_daemon_event(
	backup_type: &str,
	event: &serde_json::Value,
) -> Option<std::result::Result<(), String>> {
	let field = |key| event.get(key).and_then(serde_json::Value::as_str);
	match field("event") {
		Some("started") => {
			info!(backup_type, run_id = field("runId"), "daemon started the backup");
			None
		}
		Some("attached") => {
			info!(
				backup_type,
				run_id = field("runId"),
				started_at = field("startedAt"),
				"attached to a backup already running on the daemon"
			);
			None
		}
		Some("phase") => {
			info!(backup_type, phase = field("phase"), "backup phase");
			None
		}
		Some("follower") => {
			info!(follower_type = field("type"), "continuing into follower backup");
			None
		}
		Some("heartbeat") => None,
		Some("done") => {
			info!(
				backup_type,
				snapshot_id = field("snapshotId"),
				"daemon finished the backup"
			);
			Some(Ok(()))
		}
		Some("error") => Some(Err(field("message").unwrap_or("unknown error").to_owned())),
		_ => None,
	}
}

/// Parsed bits of a finished kopia `snapshot create --json` we report to Canopy.
#[derive(Debug, Default, PartialEq, Eq)]
struct SnapshotResult {
	id: Option<String>,
	bytes_uploaded: Option<i64>,
}

/// A status event emitted by [`run_backup`] when it's given a progress sink, so
/// the daemon's backup task can stream a run's progress to an attached client.
#[derive(Debug, Clone)]
pub enum BackupEvent {
	/// The run acquired its per-type lock and started.
	Started { run_id: String },
	/// Entered a named phase: `prepare`, `snapshot`, or `report`.
	Phase(&'static str),
	/// A live progress line from kopia during the snapshot upload (its own
	/// human-readable status, e.g. "8 hashed (800 MB), uploaded 800 MB, 100%").
	Progress(String),
	/// Continuing into a follower backup (a def with `after` naming the type
	/// that just completed).
	Follower { backup_type: String },
	/// The run finished successfully.
	Done {
		snapshot_id: Option<String>,
		bytes_uploaded: Option<i64>,
	},
	/// The run failed.
	Failed { error: String },
}

/// Sink for [`BackupEvent`]s. Unbounded so emitting never blocks the run on a
/// slow consumer; events are best-effort and dropped once the receiver is gone.
pub type BackupProgress = tokio::sync::mpsc::UnboundedSender<BackupEvent>;

fn emit(sink: &Option<BackupProgress>, event: BackupEvent) {
	if let Some(tx) = sink {
		let _ = tx.send(event);
	}
}

/// Connection details for a kopia run: the loopback proxy endpoint and the repo
/// passphrase, owned so they outlive the borrows in [`S3KopiaEnv`].
pub(super) struct RepoConn {
	pub endpoint: String,
	pub password: String,
}

/// Spawn the loopback re-signing proxy for `(backup_type, purpose)`, drawing
/// live credentials from Canopy. The returned proxy serves until dropped.
pub(super) async fn spawn_proxy(
	client: Arc<CanopyClient>,
	backup_type: String,
	purpose: BackupPurpose,
	region: &str,
	run_id: Uuid,
) -> Result<RunningProxy> {
	let provider = Arc::new(CanopyCredentialProvider::new(
		client,
		backup_type,
		purpose,
		run_id,
	));
	let upstream_host = upstream_host_for(region).await;
	let config = S3ProxyConfig {
		upstream: format!("https://{upstream_host}"),
		upstream_host,
		region: region.to_owned(),
	};
	proxy::spawn(config, provider)
		.await
		.into_diagnostic()
		.wrap_err("starting the S3 re-signing proxy")
}

fn dualstack_host(region: &str) -> String {
	format!("s3.dualstack.{region}.amazonaws.com")
}

fn plain_host(region: &str) -> String {
	format!("s3.{region}.amazonaws.com")
}

/// The S3 host the proxy connects to for `region`.
///
/// Prefer the dualstack endpoint: it carries AAAA records, so the proxy reaches
/// S3 over IPv6 where the host has it (and A records too, so IPv4-only hosts are
/// unaffected). Not every partition/region has a dualstack alias, so fall back
/// to the plain (IPv4-only) endpoint when the dualstack name doesn't resolve.
async fn upstream_host_for(region: &str) -> String {
	let dualstack = dualstack_host(region);
	let resolves = match tokio::net::lookup_host((dualstack.as_str(), 443)).await {
		Ok(mut addrs) => addrs.next().is_some(),
		Err(_) => false,
	};
	if resolves {
		debug!(host = %dualstack, "using dualstack S3 endpoint");
		dualstack
	} else {
		let plain = plain_host(region);
		debug!(host = %plain, "dualstack S3 endpoint unavailable; using plain endpoint");
		plain
	}
}

/// Connect kopia to the canopy-managed repo through the proxy (source host =
/// server id).
pub(super) async fn connect_repo(
	kopia: &Path,
	s3env: &S3KopiaEnv<'_>,
	target: &bestool_canopy::schema::BackupTarget,
	endpoint: &str,
	server_id: &str,
	run_as: RunAs,
	cache: CacheProfile,
) -> Result<()> {
	let mut connect = build_kopia_command_with_s3(kopia, s3env, run_as).map_err(|e| miette!("{e}"))?;
	args_repository_connect_s3(
		&mut connect,
		&S3Connection {
			bucket: &target.bucket,
			prefix: &target.prefix,
			region: &target.region,
			endpoint,
			username: "canopy",
			hostname: server_id,
			cache: CacheLimits::resolve(cache, config::cache_budget_override()),
		},
	);
	run_kopia(connect, "repository connect").await?;
	Ok(())
}

/// How a run ended, for the chaining decision: only a completed backup pulls
/// its followers along.
enum RunEnd {
	/// A backup ran and reported.
	Completed,
	/// Nothing ran: the device isn't authorised yet, or another run of this
	/// type already holds the lock.
	Idle,
}

/// Drive a backup end-to-end, then any followers (defs whose `after` names the
/// type that just completed), each a full run of its own.
///
/// On a dormant target (the device isn't authorised for backups yet) this logs
/// and returns `Ok(())` without reporting. Otherwise each run in the chain
/// reports its outcome to Canopy once kopia has started. An idle or failed run
/// chains nothing, but doesn't stop runs already queued behind it; the first
/// failure is returned once the chain is drained.
///
/// spec: BAK#follower-backups
pub async fn run_backup(
	backup_type: &str,
	registration_dir: Option<&Path>,
	backups_dir: Option<&Path>,
	progress: Option<BackupProgress>,
	hold: Arc<AtomicBool>,
) -> Result<()> {
	let dir = backups_dir
		.map(|d| d.to_path_buf())
		.unwrap_or_else(config::backups_dir);

	let mut visited = std::collections::BTreeSet::from([backup_type.to_owned()]);
	let mut queue = std::collections::VecDeque::from([backup_type.to_owned()]);
	let mut lead_type = true;
	let mut first_error = None;
	let mut shared: Option<SharedCapture> = None;
	while let Some(current) = queue.pop_front() {
		if !lead_type {
			info!(backup_type = %current, "continuing into follower backup");
			emit(&progress, BackupEvent::Follower { backup_type: current.clone() });
		}
		lead_type = false;

		let end = match run_one(
			&current,
			registration_dir,
			backups_dir,
			&progress,
			&hold,
			&mut shared,
		)
		.await
		{
			Ok(end) => end,
			Err(err) => {
				first_error.get_or_insert(err);
				continue;
			}
		};
		if !matches!(end, RunEnd::Completed) {
			continue;
		}
		let followers = match config::followers_of(&dir, &current).await {
			Ok(followers) => followers,
			Err(err) => {
				first_error.get_or_insert(err);
				break;
			}
		};
		for follower in followers {
			if visited.insert(follower.r#type.clone()) {
				queue.push_back(follower.r#type);
			}
		}
	}
	release_shared(shared.take()).await;
	match first_error {
		Some(err) => Err(err),
		None => Ok(()),
	}
}

/// Drive one backup run end-to-end.
async fn run_one(
	backup_type: &str,
	registration_dir: Option<&Path>,
	backups_dir: Option<&Path>,
	progress: &Option<BackupProgress>,
	hold: &Arc<AtomicBool>,
	shared: &mut Option<SharedCapture>,
) -> Result<RunEnd> {
	let run_id = Uuid::new_v4().to_string();

	// Resolve the def first: fail fast (and without touching the network) if this
	// host has no definition for the requested type.
	let dir = backups_dir
		.map(|d| d.to_path_buf())
		.unwrap_or_else(config::backups_dir);
	let def = config::find_def(&dir, backup_type)
		.await?
		.ok_or_else(|| miette!("no backup def for type '{backup_type}' in {}", dir.display()))?;

	// Cross-process guard: a run holds an exclusive lock for its type for its
	// whole duration, so a re-emitted "back up now" or a manual run racing the
	// daemon doesn't start a second concurrent kopia. Held until the function
	// returns (the OS releases it if we crash).
	let Some(_lock) = try_acquire_lock(&lock_path(backup_type)).await? else {
		info!(backup_type, "a backup of this type is already running; skipping");
		return Ok(RunEnd::Idle);
	};
	// A capture that can't be held is refused before anything is captured, so the
	// operator finds out from the flag rather than from a run that took a snapshot
	// and then couldn't keep it.
	if hold.load(Ordering::Relaxed) && !def.method.supports_hold() {
		bail!(
			"the {} method cannot hold a capture (backup type '{backup_type}')",
			def.method.name()
		);
	}
	info!(backup_type, method = def.method.name(), %run_id, "starting backup");
	emit(progress, BackupEvent::Started { run_id: run_id.clone() });

	let result =
		backup_after_start(&def, backup_type, &run_id, registration_dir, progress, hold, shared).await;
	if let Err(err) = &result {
		error!(backup_type, %run_id, "backup failed: {}", trim_error(err));
	}
	result
}

/// The networked part of a run: resolve the canopy target, snapshot through the
/// proxy, and report the outcome. Separated so [`run_backup`] can log one
/// success/failure line covering every exit path — otherwise an outcome reaches
/// only canopy, never the daemon's own journal.
async fn backup_after_start(
	def: &BackupDef,
	backup_type: &str,
	run_id: &str,
	registration_dir: Option<&Path>,
	progress: &Option<BackupProgress>,
	hold: &Arc<AtomicBool>,
	shared: &mut Option<SharedCapture>,
) -> Result<RunEnd> {
	let reg = load_registration(registration_dir)
		.await?
		.ok_or_else(|| miette!("not registered with canopy; run `bestool canopy register` first"))?;
	let server_id = reg
		.server_id
		.clone()
		.ok_or_else(|| miette!("registration has no server id"))?;
	// device_id is only a snapshot tag: canopy authenticates by the device cert,
	// the report doesn't carry it, and restore selects by snapshot id + type, not
	// device. Legacy-migrated registrations have no device_id (it's only set by
	// `canopy register` enrolment), so tag the snapshot when we know it and just
	// omit the tag otherwise.
	let device_id = reg.device_id.clone();

	let client = build_client(base_url_of(&reg)?, reg.device_key.as_deref()).await?;

	let target = match TargetOutcome::from_result(client.backup_target().await)? {
		TargetOutcome::Dormant => {
			info!(
				backup_type,
				"nothing to do: device not yet authorised for backups"
			);
			return Ok(RunEnd::Idle);
		}
		TargetOutcome::Ready(target) => target,
	};

	let run_uuid: Uuid = run_id
		.parse()
		.into_diagnostic()
		.wrap_err("backup run_id is not a valid uuid")?;

	// The proxy serves for the whole run; held in scope until reporting is done.
	// The run id is carried on every credential issuance so Canopy can correlate
	// the whole session.
	let proxy = spawn_proxy(
		client.clone(),
		backup_type.to_owned(),
		BackupPurpose::Backup,
		&target.region,
		run_uuid,
	)
	.await?;
	let conn = RepoConn {
		endpoint: proxy.endpoint(),
		password: target.repo_password.0.clone(),
	};

	// Sample this run's progress to Canopy while it's in flight: the kopia engine
	// counters (via the cell the snapshot phase keeps current) and the proxy's S3
	// tallies. Best-effort — the reporter never fails the run.
	let cell = Arc::new(progress::ProgressCell::default());
	let reporter = progress::ProgressReporter::spawn(
		client.clone(),
		run_uuid,
		backup_type.to_owned(),
		BackupPurpose::Backup,
		proxy.traffic_handle(),
		Some(cell.clone()),
	);

	let outcome =
		run_kopia_backup(
			def,
			&target,
			&conn,
			&server_id,
			device_id.as_deref(),
			run_id,
			progress,
			&cell,
			hold,
			shared,
		)
		.await;

	// Stop sampling before the final report, so nothing posts after teardown.
	reporter.stop().await;
	emit(progress, BackupEvent::Phase("report"));

	// The proxy saw every S3 request this run made (success or failure), so its
	// tallies are a rough measure of the network/S3 traffic this run accounts for.
	let traffic = proxy.traffic();
	info!(
		backup_type,
		run_id,
		sent_raw = traffic.sent_raw,
		sent_payload = traffic.sent_payload,
		received_raw = traffic.received_raw,
		received_payload = traffic.received_payload,
		"s3 traffic for this run"
	);

	// Report whatever happened, then surface the original error (if any). The
	// traffic counts go on both outcomes: bytes flow whether or not it succeeds.
	let to_i64 = |n: u64| i64::try_from(n).unwrap_or(i64::MAX);
	let s3_sent_raw_bytes = Some(to_i64(traffic.sent_raw));
	let s3_sent_payload_bytes = Some(to_i64(traffic.sent_payload));
	let s3_received_raw_bytes = Some(to_i64(traffic.received_raw));
	let s3_received_payload_bytes = Some(to_i64(traffic.received_payload));
	// The freeze instant, recorded by the method during prepare. Reported on both
	// outcomes (prepare may succeed and the snapshot then fail); Canopy records it
	// write-once, so a value already sent on a progress sample still stands.
	let snapshot_taken_at = cell.taken_at();
	let report = match &outcome {
		Ok(snapshot) => ReportArgs::builder()
			.run_id(run_uuid)
			.type_(backup_type.to_owned())
			.purpose(BackupPurpose::Backup)
			.outcome(RunOutcome::Success)
			.maybe_bytes_uploaded(snapshot.bytes_uploaded)
			.maybe_snapshot_id(snapshot.id.clone())
			.maybe_snapshot_taken_at(snapshot_taken_at)
			.maybe_s3_sent_raw_bytes(s3_sent_raw_bytes)
			.maybe_s3_sent_payload_bytes(s3_sent_payload_bytes)
			.maybe_s3_received_raw_bytes(s3_received_raw_bytes)
			.maybe_s3_received_payload_bytes(s3_received_payload_bytes)
			.build(),
		Err(err) => ReportArgs::builder()
			.run_id(run_uuid)
			.type_(backup_type.to_owned())
			.purpose(BackupPurpose::Backup)
			.outcome(RunOutcome::Failure)
			.error(trim_error(err))
			.maybe_snapshot_taken_at(snapshot_taken_at)
			.maybe_s3_sent_raw_bytes(s3_sent_raw_bytes)
			.maybe_s3_sent_payload_bytes(s3_sent_payload_bytes)
			.maybe_s3_received_raw_bytes(s3_received_raw_bytes)
			.maybe_s3_received_payload_bytes(s3_received_payload_bytes)
			.build(),
	};
	client
		.backup_report(&report)
		.await
		.into_diagnostic()
		.wrap_err("reporting backup outcome to canopy")?;

	match &outcome {
		Ok(snapshot) => {
			info!(
				backup_type,
				run_id,
				snapshot_id = ?snapshot.id,
				bytes_uploaded = ?snapshot.bytes_uploaded,
				"backup completed"
			);
			emit(
				progress,
				BackupEvent::Done {
					snapshot_id: snapshot.id.clone(),
					bytes_uploaded: snapshot.bytes_uploaded,
				},
			)
		}
		Err(err) => emit(progress, BackupEvent::Failed { error: trim_error(err) }),
	}

	outcome.map(|_| RunEnd::Completed)
}

/// Connect kopia to the repo, snapshot the prepared source, parse the result.
///
/// Wraps the method's `prepare`/`cleanup` in the def's `pre`/`post` hooks, and
/// always runs cleanup + post even when the snapshot fails.
#[expect(
	clippy::too_many_arguments,
	reason = "one run's inputs plus its two progress channels; bundling them buys nothing"
)]
async fn run_kopia_backup(
	def: &BackupDef,
	target: &bestool_canopy::schema::BackupTarget,
	conn: &RepoConn,
	server_id: &str,
	device_id: Option<&str>,
	run_id: &str,
	progress: &Option<BackupProgress>,
	cell: &Arc<progress::ProgressCell>,
	hold: &Arc<AtomicBool>,
	shared: &mut Option<SharedCapture>,
) -> Result<SnapshotResult> {
	run_hooks(&def.pre, true).await?;

	emit(progress, BackupEvent::Phase("prepare"));
	let within = shared.as_ref().map(|kept| kept.volume.clone());
	let prepared = def.method.prepare(&def.r#type, within.as_ref()).await?;
	// Record the freeze instant (where the method has one) so the reporter and the
	// final report can carry it.
	if let Some(taken_at) = prepared.taken_at {
		cell.set_taken_at(taken_at);
	}
	let source_path = prepared.path.clone();
	let tags = assemble_tags(&def.tags, &prepared.extra_tags, device_id, run_id, &def.r#type);

	emit(progress, BackupEvent::Phase("snapshot"));
	info!(
		backup_type = %def.r#type,
		source = %source_path.display(),
		"uploading snapshot to kopia repository"
	);
	let result = snapshot(
		target,
		conn,
		&source_path,
		server_id,
		&def.r#type,
		&tags,
		&prepared.ignore,
		progress,
		cell,
	)
	.await;

	// Cleanup and post-hooks run regardless of the snapshot outcome — except that
	// a run told to hold keeps its capture instead of releasing it. The
	// instruction is read here rather than at the start because it can arrive at
	// any point up to now, from an operator who decided mid-transfer that they
	// want a local rollback point.
	let cleanup = if hold.load(Ordering::Relaxed) {
		hold_capture(def, prepared, result.is_ok()).await.map(|_| ())
	} else if let Some(volume) = prepared.volume.clone()
		&& shared.is_none()
	{
		*shared = Some(SharedCapture {
			def: def.clone(),
			prepared,
			volume,
		});
		Ok(())
	} else {
		def.method.cleanup(prepared).await
	};
	run_hooks(&def.post, false).await.ok();

	let snapshot = result?;
	cleanup?;
	Ok(snapshot)
}

/// A leader's whole-volume capture, kept past its own run so followers whose
/// sources sit on that volume read from it instead of taking a second snapshot
/// of data it already froze. Only the first capture in a chain is kept; a
/// follower's own is released with its run.
///
/// spec: BAK#sharing-a-capture
struct SharedCapture {
	def: BackupDef,
	prepared: method::Prepared,
	volume: method::VolumeCapture,
}

/// Release a kept capture. Best-effort: the runs that used it have finished and
/// reported, so failing to tear it down is a leak to warn about, not an error to
/// fail anything with.
async fn release_shared(kept: Option<SharedCapture>) {
	let Some(SharedCapture { def, prepared, .. }) = kept else {
		return;
	};
	if let Err(err) = def.method.cleanup(prepared).await {
		warn!(backup_type = %def.r#type, "releasing the shared capture failed: {}", trim_error(&err));
	}
}

/// Retain this run's capture and record it, so it outlives the process as a
/// local rollback point.
async fn hold_capture(def: &BackupDef, prepared: method::Prepared, uploaded: bool) -> Result<hold::HoldRecord> {
	let taken_at = prepared.taken_at;
	let held_at = jiff::Timestamp::now();
	// A capture with no freeze instant (a streamed base backup represents an
	// interval, not a point) is named for when it was kept instead, and records
	// no freeze rather than passing that off as one.
	let id = hold::mint_id(&def.r#type, taken_at.unwrap_or(held_at));

	let (source, capture) = def
		.method
		.hold(prepared, &id)
		.await
		.wrap_err("holding the capture")?;

	let record = hold::HoldRecord {
		id,
		backup_type: def.r#type.clone(),
		taken_at,
		held_at,
		source,
		uploaded,
		capture,
	};
	hold::save(&record).await?;
	info!(
		hold = %record.id,
		source = %record.source.display(),
		backend = record.capture.backend(),
		"capture held; release it with `bestool canopy hold drop`"
	);
	Ok(record)
}

/// Take a capture and keep it, without a repository: no credentials, no
/// transfer, no report. The definition's hooks and the capture itself are as
/// they are for an uploading run, so the result is the same artefact.
async fn capture_only(backup_type: &str, backups_dir: Option<&Path>) -> Result<()> {
	let dir = backups_dir
		.map(|d| d.to_path_buf())
		.unwrap_or_else(config::backups_dir);
	let def = config::find_def(&dir, backup_type)
		.await?
		.ok_or_else(|| miette!("no backup def for type '{backup_type}' in {}", dir.display()))?;

	// The same per-type lock an uploading run takes, so a capture-only run can't
	// start alongside one the daemon is already driving.
	let Some(_lock) = try_acquire_lock(&lock_path(backup_type)).await? else {
		info!(backup_type, "a backup of this type is already running; skipping");
		return Ok(());
	};
	if !def.method.supports_hold() {
		bail!(
			"the {} method cannot hold a capture (backup type '{backup_type}')",
			def.method.name()
		);
	}

	info!(backup_type, method = def.method.name(), "capturing without uploading");
	run_hooks(&def.pre, true).await?;
	let held = match def.method.prepare(backup_type, None).await {
		Ok(prepared) => hold_capture(&def, prepared, false).await.map(|_| ()),
		Err(err) => Err(err),
	};
	run_hooks(&def.post, false).await.ok();
	held
}

/// Connect to the repo and create the snapshot.
#[expect(
	clippy::too_many_arguments,
	reason = "the snapshot inputs plus its two progress channels; bundling them buys nothing"
)]
async fn snapshot(
	target: &bestool_canopy::schema::BackupTarget,
	conn: &RepoConn,
	source_path: &Path,
	server_id: &str,
	backup_type: &str,
	tags: &BTreeMap<String, String>,
	ignore: &[String],
	progress: &Option<BackupProgress>,
	cell: &Arc<progress::ProgressCell>,
) -> Result<SnapshotResult> {
	let kopia = find_kopia_binary(None).ok_or_else(|| miette!("could not find the kopia binary"))?;

	// A transient kopia config so the bucket/password never persist on the device.
	let config_dir = transient_config_dir()?;
	let config_path = config_dir.path().join("repository.config");

	// When kopia runs as the kopia user (not us), it writes and reads this config
	// itself, so hand the dir to that user. Root-owned 0700 tempdir would deny it.
	#[cfg(target_os = "linux")]
	if let Some(user) = bestool_kopia::kopia_run_as_user() {
		let status = tokio::process::Command::new("chown")
			.arg("-R")
			.arg(format!("{user}:{user}"))
			.arg(config_dir.path())
			.status()
			.await
			.into_diagnostic()
			.wrap_err("handing the transient kopia config dir to the kopia user")?;
		if !status.success() {
			bail!("chown of transient kopia config dir to {user} failed ({status})");
		}
	}
	let s3env = S3KopiaEnv {
		password: &conn.password,
		config_path: &config_path,
	};

	connect_repo(
		&kopia,
		&s3env,
		target,
		&conn.endpoint,
		server_id,
		RunAs::KopiaUser,
		CacheProfile::Push,
	)
	.await?;

	if !ignore.is_empty() {
		let mut policy =
			build_kopia_command_with_s3(&kopia, &s3env, RunAs::KopiaUser).map_err(|e| miette!("{e}"))?;
		args_policy_set_ignores(&mut policy, source_path, ignore);
		run_kopia(policy, "policy set").await?;
	}

	let mut create =
		build_kopia_command_with_s3(&kopia, &s3env, RunAs::KopiaUser).map_err(|e| miette!("{e}"))?;
	// Force kopia's progress output (it stays silent on a non-TTY otherwise) and
	// stream it: the parsed counters feed the Canopy progress reporter, and the
	// raw lines the local CLI display, on every run.
	create.arg("--progress");
	args_snapshot_create(&mut create, source_path, tags, backup_type);
	let stdout = run_kopia_streaming(create, "snapshot create", progress, cell).await?;
	Ok(parse_snapshot_output(&stdout))
}

/// Create a transient directory for a throwaway kopia config (so the bucket and
/// password never persist on the device), ensuring the base temp dir exists
/// first. On Windows `%TEMP%` is often a per-session subdirectory
/// (`…\Temp\<session>`) that may not exist yet, and `tempfile` doesn't create the
/// parent — so create it here, or the tempdir fails with "path not found".
pub(super) fn transient_config_dir() -> Result<tempfile::TempDir> {
	let base = std::env::temp_dir();
	std::fs::create_dir_all(&base)
		.into_diagnostic()
		.wrap_err_with(|| format!("creating temp dir {}", base.display()))?;
	tempfile::tempdir_in(&base)
		.into_diagnostic()
		.wrap_err("creating transient kopia config dir")
}

/// Run a kopia command with its stdout/stderr inherited, so kopia's own
/// progress display reaches the terminal (it stays silent when its output is a
/// pipe, as [`run_kopia`] leaves it). stdin is detached so kopia can't consume
/// the caller's prompts. Errors on a non-zero exit; the output the user already
/// saw stands in for a captured error message.
pub(super) async fn run_kopia_visible(cmd: std::process::Command, what: &str) -> Result<()> {
	let status = tokio::process::Command::from(cmd)
		.stdin(std::process::Stdio::null())
		.status()
		.await
		.into_diagnostic()
		.wrap_err_with(|| format!("spawning kopia {what}"))?;
	if !status.success() {
		bail!("kopia {what} failed ({status})");
	}
	Ok(())
}

/// Run a kopia command, returning its stdout on success.
pub(super) async fn run_kopia(cmd: std::process::Command, what: &str) -> Result<String> {
	let output = tokio::process::Command::from(cmd)
		.output()
		.await
		.into_diagnostic()
		.wrap_err_with(|| format!("spawning kopia {what}"))?;
	if !output.status.success() {
		let stderr = String::from_utf8_lossy(&output.stderr);
		bail!("kopia {what} failed: {}", stderr.trim());
	}
	Ok(String::from_utf8_lossy(&output.stdout).into_owned())
}

/// Run a kopia command, streaming its progress lines to the sink as they arrive
/// (kopia writes progress to stderr, `\r`-separated), and returning its stdout.
async fn run_kopia_streaming(
	cmd: std::process::Command,
	what: &str,
	progress: &Option<BackupProgress>,
	cell: &Arc<progress::ProgressCell>,
) -> Result<String> {
	use std::process::Stdio;

	use tokio::io::AsyncReadExt as _;

	let mut child = tokio::process::Command::from(cmd)
		.stdout(Stdio::piped())
		.stderr(Stdio::piped())
		.spawn()
		.into_diagnostic()
		.wrap_err_with(|| format!("spawning kopia {what}"))?;
	let mut stdout = child.stdout.take().expect("piped stdout");
	let mut stderr = child.stderr.take().expect("piped stderr");

	// Read stderr concurrently, splitting on `\r`/`\n` (kopia updates its progress
	// line with carriage returns), forwarding progress lines and keeping the lot
	// for an error message.
	let progress = progress.clone();
	let cell = cell.clone();
	let stderr_task = tokio::spawn(async move {
		let mut captured = String::new();
		let mut buf = [0u8; 4096];
		let mut segment = Vec::new();
		// Throttle journal logging: kopia rewrites its progress line many times a
		// second, but the journal only wants an occasional heartbeat.
		let mut last_log: Option<Instant> = None;
		while let Ok(n) = stderr.read(&mut buf).await {
			if n == 0 {
				break;
			}
			for &byte in &buf[..n] {
				if byte == b'\r' || byte == b'\n' {
					flush_progress_segment(&mut segment, &mut captured, &progress, &cell, &mut last_log);
				} else {
					segment.push(byte);
				}
			}
		}
		flush_progress_segment(&mut segment, &mut captured, &progress, &cell, &mut last_log);
		captured
	});

	let mut out = String::new();
	stdout
		.read_to_string(&mut out)
		.await
		.into_diagnostic()
		.wrap_err_with(|| format!("reading kopia {what} stdout"))?;
	let status = child
		.wait()
		.await
		.into_diagnostic()
		.wrap_err_with(|| format!("waiting for kopia {what}"))?;
	let captured = stderr_task.await.unwrap_or_default();

	if !status.success() {
		bail!("kopia {what} failed: {}", captured.trim());
	}
	Ok(out)
}

/// How often a kopia progress line is logged to the journal. Progress also goes
/// to the status sink unthrottled; this only paces the journal heartbeat.
const PROGRESS_LOG_INTERVAL: Duration = Duration::from_secs(30);

fn flush_progress_segment(
	segment: &mut Vec<u8>,
	captured: &mut String,
	progress: &Option<BackupProgress>,
	cell: &progress::ProgressCell,
	last_log: &mut Option<Instant>,
) {
	if segment.is_empty() {
		return;
	}
	let line = String::from_utf8_lossy(segment).trim().to_string();
	segment.clear();
	if line.is_empty() {
		return;
	}
	captured.push_str(&line);
	captured.push('\n');
	// kopia's progress line carries these counters; other stderr lines (e.g.
	// maintenance) aren't progress and aren't forwarded. A parseable line also
	// updates the cell the Canopy reporter samples — gated separately so a format
	// drift that defeats the parser still leaves the local display working.
	if line.contains("hashing") || line.contains("hashed") || line.contains("uploaded") {
		if let Some(parsed) = bestool_kopia::progress::parse_progress(&line) {
			cell.set_kopia(line.clone(), parsed);
		}
		let now = Instant::now();
		if last_log.is_none_or(|prev| now.duration_since(prev) >= PROGRESS_LOG_INTERVAL) {
			info!(progress = %line, "kopia upload progress");
			*last_log = Some(now);
		}
		emit(progress, BackupEvent::Progress(line));
	}
}

/// Run a sequence of hooks. `fail_fast` aborts on the first failure (pre-hooks);
/// otherwise failures are logged and the rest still run (post-hooks).
pub(super) async fn run_hooks(hooks: &[Hook], fail_fast: bool) -> Result<()> {
	for hook in hooks {
		if let Err(err) = run_hook(hook).await {
			if fail_fast {
				return Err(err);
			}
			warn!("post-hook failed (continuing): {err}");
		}
	}
	Ok(())
}

async fn run_hook(hook: &Hook) -> Result<()> {
	let Some((program, args)) = hook.command.split_first() else {
		bail!("hook has an empty command");
	};
	let status = tokio::process::Command::new(program)
		.args(args)
		.status()
		.await
		.into_diagnostic()
		.wrap_err_with(|| format!("running hook {program}"))?;
	if !status.success() {
		bail!("hook {program} exited with {status}");
	}
	Ok(())
}

/// Merge the def's tags, the method's extra tags, and the canopy-* tags.
///
/// The canopy-* tags take precedence so a def can't accidentally override them.
fn assemble_tags(
	def_tags: &BTreeMap<String, String>,
	extra_tags: &BTreeMap<String, String>,
	device_id: Option<&str>,
	run_id: &str,
	backup_type: &str,
) -> BTreeMap<String, String> {
	let mut tags = def_tags.clone();
	tags.extend(extra_tags.iter().map(|(k, v)| (k.clone(), v.clone())));
	if let Some(device_id) = device_id {
		tags.insert("canopy-device".to_owned(), device_id.to_owned());
	}
	tags.insert("canopy-run".to_owned(), run_id.to_owned());
	tags.insert("canopy-type".to_owned(), backup_type.to_owned());
	tags
}

/// Best-effort extraction of the snapshot id and uploaded bytes from
/// `kopia snapshot create --json` output.
fn parse_snapshot_output(stdout: &str) -> SnapshotResult {
	let Ok(value) = serde_json::from_str::<serde_json::Value>(stdout.trim()) else {
		return SnapshotResult::default();
	};
	let id = value
		.get("id")
		.and_then(|v| v.as_str())
		.map(|s| s.to_owned());
	// `snapshot create --json` strips the whole `stats` object (kopia only emits it
	// under `--json-verbose`), so the size lives in the root directory summary
	// (`rootEntry.summ.size`), which is always present. It equals `stats.totalSize`
	// for a complete snapshot (both sum every file once); fall back to it for the
	// verbose form.
	let bytes_uploaded = value
		.get("rootEntry")
		.and_then(|r| r.get("summ"))
		.and_then(|s| s.get("size"))
		.and_then(serde_json::Value::as_i64)
		.or_else(|| {
			value
				.get("stats")
				.and_then(|s| s.get("totalSize"))
				.and_then(serde_json::Value::as_i64)
		});
	SnapshotResult { id, bytes_uploaded }
}

/// Resolve the canopy base URL from the registration (or the default).
pub(super) fn base_url_of(reg: &Registration) -> Result<Url> {
	reg.api_url
		.as_deref()
		.unwrap_or(DEFAULT_CANOPY_URL)
		.parse()
		.into_diagnostic()
		.wrap_err("parsing canopy api_url")
}

/// Build a canopy client for an already-enrolled host (tailscale, then mTLS).
///
/// The device key is optional: a host on the tailnet authenticates by its
/// tailscale identity, and only needs the key when the tailnet is out of reach.
pub(super) async fn build_client(
	base_url: Url,
	device_key: Option<&str>,
) -> Result<Arc<CanopyClient>> {
	let tailscale_url = bestool_canopy::TAILSCALE_URL
		.parse()
		.into_diagnostic()
		.wrap_err("parsing default canopy tailscale URL")?;
	let client =
		connect_to(base_url, tailscale_url, device_key, crate::http::client_builder)
			.await?
			.ok_or_else(|| {
				miette!("could not build a canopy client: the tailnet is unreachable and the registration has no device key")
			})?;
	Ok(Arc::new(client))
}

/// Load the registration, honouring an explicit `--config` dir.
pub(super) async fn load_registration(config: Option<&Path>) -> Result<Option<Registration>> {
	match config {
		Some(dir) => bestool_canopy::registration::load_from(dir).await,
		None => bestool_canopy::registration::load().await,
	}
}

/// Cap a kopia error to a length that fits a log line and a canopy report field.
const MAX_ERROR_LEN: usize = 1500;

pub(super) fn trim_error(err: &miette::Report) -> String {
	// Render the whole cause chain: `Display` on a report shows only the outermost
	// context ("spawning pg_basebackup"), and the cause is the part worth reporting.
	let msg = err
		.chain()
		.map(|e| e.to_string())
		.collect::<Vec<_>>()
		.join(": ");
	let len = msg.chars().count();
	if len <= MAX_ERROR_LEN {
		return msg;
	}
	// kopia leads with warnings ("too many index blobs …") and progress lines and
	// prints the operative error last, so keep the tail rather than the head.
	let tail: String = msg.chars().skip(len - MAX_ERROR_LEN).collect();
	format!("…{tail}")
}

/// Per-type lockfile path, in a machine-global runtime dir. On Linux that's
/// `/run/bestool` (tmpfs, so it's cleared on reboot and never stale across
/// crashes); on Windows it's under `%ProgramData%` — see [`windows_lock_dir`] for
/// why there rather than the per-user `%TEMP%`.
fn lock_path(backup_type: &str) -> std::path::PathBuf {
	let name = format!("backup-{}.lock", backup_type.replace(['/', '\\'], "_"));
	#[cfg(unix)]
	{
		std::path::PathBuf::from("/run/bestool").join(name)
	}
	#[cfg(not(unix))]
	{
		windows_lock_dir(std::env::var_os("ProgramData").as_deref()).join(name)
	}
}

/// The machine-global directory holding backup lockfiles on Windows:
/// `%ProgramData%\bestool\run`, shared by every account. The daemon runs as
/// LocalSystem and a manual `bestool canopy backup` as an interactive
/// administrator; the per-user `%TEMP%` would give each its own lockfile, so the
/// cross-process guard wouldn't serialise them and two runs could stream a
/// `pg_basebackup` into the one staging dir at once. The advisory lock is released
/// by the OS when the holder exits, so a lockfile persisting between runs is never
/// stale.
#[cfg(any(not(unix), test))]
fn windows_lock_dir(program_data: Option<&std::ffi::OsStr>) -> std::path::PathBuf {
	program_data
		.map(std::path::PathBuf::from)
		.unwrap_or_else(|| std::path::PathBuf::from(r"C:\ProgramData"))
		.join("bestool")
		.join("run")
}

/// Try to take the exclusive per-run lock. `Ok(Some(file))` holds the lock for
/// as long as the returned handle lives; `Ok(None)` means another run holds it.
async fn try_acquire_lock(path: &Path) -> Result<Option<tokio::fs::File>> {
	use fs4::tokio::AsyncFileExt as _;

	if let Some(parent) = path.parent() {
		tokio::fs::create_dir_all(parent).await.ok();
	}
	let file = tokio::fs::OpenOptions::new()
		.create(true)
		.write(true)
		.truncate(false)
		.open(path)
		.await
		.into_diagnostic()
		.wrap_err_with(|| format!("opening backup lockfile {}", path.display()))?;
	match file.try_lock() {
		Ok(()) => Ok(Some(file)),
		Err(fs4::TryLockError::WouldBlock) => Ok(None),
		Err(fs4::TryLockError::Error(err)) => Err(err)
			.into_diagnostic()
			.wrap_err_with(|| format!("locking backup lockfile {}", path.display())),
	}
}

#[cfg(test)]
mod tests {
	use super::*;

	#[test]
	fn trim_error_renders_the_cause_chain() {
		let err = Err::<(), _>(std::io::Error::new(
			std::io::ErrorKind::NotFound,
			"No such file or directory",
		))
		.into_diagnostic()
		.wrap_err("spawning pg_basebackup")
		.unwrap_err();
		assert_eq!(
			trim_error(&err),
			"spawning pg_basebackup: No such file or directory"
		);
	}

	#[test]
	fn trim_error_keeps_the_operative_tail() {
		let head = "Found too many index blobs (1256), …\n".repeat(100);
		let msg = format!("{head}not authorized to perform: s3:PutObjectRetention");
		let trimmed = trim_error(&miette::miette!("{msg}"));
		assert!(trimmed.chars().count() <= MAX_ERROR_LEN + 1);
		assert!(trimmed.contains("s3:PutObjectRetention"));
		assert!(trimmed.starts_with('…'));
	}

	#[test]
	fn upstream_host_formats() {
		assert_eq!(
			dualstack_host("ap-southeast-2"),
			"s3.dualstack.ap-southeast-2.amazonaws.com"
		);
		assert_eq!(plain_host("ap-southeast-2"), "s3.ap-southeast-2.amazonaws.com");
	}

	#[test]
	fn flush_forwards_only_progress_lines() {
		let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
		let progress = Some(tx);
		let cell = progress::ProgressCell::default();
		let mut captured = String::new();
		let mut last_log = None;

		// A kopia progress update is forwarded and updates the cell…
		let mut seg =
			b" * 8 hashing, 8 hashed (800 MB), 0 cached (0 B), uploaded 800 MB, estimating...".to_vec();
		flush_progress_segment(&mut seg, &mut captured, &progress, &cell, &mut last_log);
		// …an ordinary stderr line (e.g. maintenance) is not.
		let mut seg = b"Finished full maintenance.".to_vec();
		flush_progress_segment(&mut seg, &mut captured, &progress, &cell, &mut last_log);

		drop(progress);
		let mut events = Vec::new();
		while let Ok(event) = rx.try_recv() {
			events.push(event);
		}
		assert_eq!(events.len(), 1);
		assert!(matches!(&events[0], BackupEvent::Progress(line) if line.contains("uploaded")));
		// Both lines are retained for error context.
		assert!(captured.contains("Finished full maintenance."));
		// The progress line's counters reached the cell.
		assert_eq!(cell.taken_at(), None);
	}

	#[test]
	fn assemble_tags_merges_and_canopy_tags_win() {
		let mut def_tags = BTreeMap::new();
		def_tags.insert("app".to_owned(), "tamanu".to_owned());
		// A def must not be able to override the canopy-* tags.
		def_tags.insert("canopy-type".to_owned(), "spoofed".to_owned());
		let mut extra = BTreeMap::new();
		extra.insert("pg-version".to_owned(), "16".to_owned());

		let tags = assemble_tags(
			&def_tags,
			&extra,
			Some("device-uuid"),
			"run-uuid",
			"tamanu-postgres",
		);

		assert_eq!(tags.get("app").map(String::as_str), Some("tamanu"));
		assert_eq!(tags.get("pg-version").map(String::as_str), Some("16"));
		assert_eq!(tags.get("canopy-device").map(String::as_str), Some("device-uuid"));
		assert_eq!(tags.get("canopy-run").map(String::as_str), Some("run-uuid"));
		assert_eq!(
			tags.get("canopy-type").map(String::as_str),
			Some("tamanu-postgres")
		);
	}

	#[test]
	fn assemble_tags_omits_device_when_absent() {
		// A legacy-migrated host has no device id; the snapshot is still tagged
		// with run and type, just without canopy-device.
		let tags = assemble_tags(
			&BTreeMap::new(),
			&BTreeMap::new(),
			None,
			"run-uuid",
			"tamanu-postgres",
		);

		assert!(!tags.contains_key("canopy-device"));
		assert_eq!(tags.get("canopy-run").map(String::as_str), Some("run-uuid"));
		assert_eq!(
			tags.get("canopy-type").map(String::as_str),
			Some("tamanu-postgres")
		);
	}

	#[test]
	fn lock_path_names_per_type_and_sanitises_separators() {
		assert!(
			lock_path("tamanu-postgres")
				.to_string_lossy()
				.ends_with("backup-tamanu-postgres.lock")
		);
		assert!(
			lock_path("a/b")
				.to_string_lossy()
				.ends_with("backup-a_b.lock")
		);
	}

	#[test]
	fn windows_lock_dir_is_machine_global_under_programdata() {
		use std::ffi::OsStr;

		// Rooted at %ProgramData% (shared by every account: the LocalSystem daemon and
		// a manual admin run must contend on one file), never a per-user temp dir.
		let base = OsStr::new(r"D:\ProgramData");
		let dir = windows_lock_dir(Some(base));
		assert!(dir.starts_with(std::path::Path::new(base)));
		let last_two: Vec<String> = dir
			.components()
			.rev()
			.take(2)
			.map(|c| c.as_os_str().to_string_lossy().into_owned())
			.collect();
		assert_eq!(last_two, vec!["run".to_owned(), "bestool".to_owned()]);

		// Falls back to the default ProgramData when the env var is unset.
		assert!(
			windows_lock_dir(None).starts_with(std::path::Path::new(r"C:\ProgramData"))
		);
	}

	#[tokio::test]
	async fn lock_is_exclusive_and_releases_on_drop() {
		let tmp = tempfile::tempdir().unwrap();
		let path = tmp.path().join("backup-test.lock");
		let held = try_acquire_lock(&path).await.unwrap();
		assert!(held.is_some(), "first acquire takes the lock");
		// A second attempt while the first is held is refused.
		assert!(
			try_acquire_lock(&path).await.unwrap().is_none(),
			"a concurrent run is locked out"
		);
		drop(held);
		// Released → acquirable again.
		assert!(try_acquire_lock(&path).await.unwrap().is_some());
	}

	#[test]
	fn parse_snapshot_output_extracts_id_and_bytes() {
		let out = r#"{"id":"abc123","stats":{"totalSize":987654},"rootEntry":{}}"#;
		assert_eq!(
			parse_snapshot_output(out),
			SnapshotResult {
				id: Some("abc123".to_owned()),
				bytes_uploaded: Some(987654),
			}
		);
	}

	#[test]
	fn parse_snapshot_output_prefers_root_summary_size() {
		// The real `--json` output has no `stats` (kopia strips it); the size comes
		// from the root summary. When both are present, the root summary wins.
		let out = r#"{"id":"x","stats":{"totalSize":10},"rootEntry":{"summ":{"size":987654}}}"#;
		assert_eq!(
			parse_snapshot_output(out),
			SnapshotResult {
				id: Some("x".to_owned()),
				bytes_uploaded: Some(987654),
			}
		);
	}

	#[test]
	fn parse_snapshot_output_tolerates_missing_fields_and_garbage() {
		assert_eq!(parse_snapshot_output("not json"), SnapshotResult::default());
		assert_eq!(
			parse_snapshot_output(r#"{"id":"x"}"#),
			SnapshotResult {
				id: Some("x".to_owned()),
				bytes_uploaded: None,
			}
		);
	}
}