bestool 1.27.0

//! 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 creds;
pub mod method;
pub mod postgresql;

use std::{collections::BTreeMap, path::Path, sync::Arc};

use bestool_canopy::{
	BackupReport, CanopyClient, DEFAULT_CANOPY_URL, Outcome, Purpose, TargetOutcome,
	registration::Registration,
};
use bestool_kopia::{
	S3KopiaEnv, args_policy_set_ignores, args_repository_connect_s3, args_snapshot_create,
	build_kopia_command_with_s3, find_kopia_binary,
};
use clap::Parser;
use miette::{Context as _, IntoDiagnostic as _, Result, bail, miette};
use reqwest::Url;
use tracing::{info, warn};
use uuid::Uuid;

use self::{
	config::{BackupDef, Hook},
	creds::CredsServer,
};
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>,
}

pub async fn run(args: BackupArgs, _ctx: Context) -> Result<()> {
	run_backup(
		&args.backup_type,
		args.config.as_deref(),
		args.backups_dir.as_deref(),
	)
	.await
}

/// 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>,
}

/// The per-run kopia env values (loopback creds endpoint + repo password),
/// owned so they outlive the borrow of the lease.
pub(super) struct LeaseEnv {
	pub uri: String,
	pub token: String,
	pub password: String,
}

/// Lease a creds token bound to a `(type, purpose)`, refreshed from Canopy.
pub(super) fn make_lease(
	server: &CredsServer,
	client: Arc<CanopyClient>,
	base_url: Url,
	backup_type: String,
	purpose: Purpose,
) -> creds::CredsLease {
	server.lease(Arc::new(move || {
		let client = client.clone();
		let base_url = base_url.clone();
		let backup_type = backup_type.clone();
		Box::pin(async move {
			client
				.backup_credentials(&base_url, &backup_type, purpose)
				.await
				.map_err(|err| format!("{err}"))
		})
	}))
}

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

/// Drive one backup run end-to-end.
///
/// On a dormant target (the device isn't authorised for backups yet) this logs
/// and returns `Ok(())` without reporting. Otherwise it always reports the
/// outcome to Canopy once kopia has started.
pub async fn run_backup(
	backup_type: &str,
	registration_dir: Option<&Path>,
	backups_dir: Option<&Path>,
) -> Result<()> {
	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(());
	};
	info!(backup_type, method = def.method.name(), %run_id, "starting backup");

	let reg = load_registration(registration_dir)
		.await?
		.ok_or_else(|| miette!("not registered with canopy; run `bestool canopy register` first"))?;
	let device_key = reg
		.device_key
		.clone()
		.ok_or_else(|| miette!("registration has no device key"))?;
	let server_id = reg
		.server_id
		.clone()
		.ok_or_else(|| miette!("registration has no server id"))?;
	let device_id = reg
		.device_id
		.clone()
		.ok_or_else(|| miette!("registration has no device id"))?;
	let base_url = base_url_of(&reg)?;

	let client = build_client(&device_key).await?;

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

	let creds_server = CredsServer::start().await?;
	let lease = make_lease(
		&creds_server,
		client.clone(),
		base_url.clone(),
		backup_type.to_owned(),
		Purpose::Backup,
	);

	let env = LeaseEnv {
		uri: lease.uri().to_owned(),
		token: lease.token().to_owned(),
		password: target.repo_password.0.clone(),
	};
	let outcome = run_kopia_backup(&def, &target, &env, &server_id, &device_id, &run_id).await;

	// Report whatever happened, then surface the original error (if any).
	let report = match &outcome {
		Ok(snapshot) => BackupReport {
			run_id: &run_id,
			r#type: backup_type,
			purpose: Purpose::Backup,
			outcome: Outcome::Success,
			error: None,
			bytes_uploaded: snapshot.bytes_uploaded,
			snapshot_id: snapshot.id.as_deref(),
		},
		Err(err) => BackupReport {
			run_id: &run_id,
			r#type: backup_type,
			purpose: Purpose::Backup,
			outcome: Outcome::Failure,
			error: Some(&trim_error(err)),
			bytes_uploaded: None,
			snapshot_id: None,
		},
	};
	client
		.backup_report(&base_url, &report)
		.await
		.wrap_err("reporting backup outcome to canopy")?;

	outcome.map(|_| ())
}

/// 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.
async fn run_kopia_backup(
	def: &BackupDef,
	target: &bestool_canopy::BackupTarget,
	env: &LeaseEnv,
	server_id: &str,
	device_id: &str,
	run_id: &str,
) -> Result<SnapshotResult> {
	run_hooks(&def.pre, true).await?;

	let prepared = def.method.prepare(&def.r#type).await?;
	let source_path = prepared.path.clone();
	let tags = assemble_tags(&def.tags, &prepared.extra_tags, device_id, run_id, &def.r#type);

	let result = snapshot(target, env, &source_path, server_id, &tags, &prepared.ignore).await;

	// Cleanup and post-hooks run regardless of the snapshot outcome.
	let cleanup = def.method.cleanup(prepared).await;
	run_hooks(&def.post, false).await.ok();

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

/// Connect to the repo and create the snapshot.
async fn snapshot(
	target: &bestool_canopy::BackupTarget,
	env: &LeaseEnv,
	source_path: &Path,
	server_id: &str,
	tags: &BTreeMap<String, String>,
	ignore: &[String],
) -> 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 = tempfile::tempdir()
		.into_diagnostic()
		.wrap_err("creating transient kopia config dir")?;
	let config_path = config_dir.path().join("repository.config");
	let s3env = S3KopiaEnv {
		full_uri: &env.uri,
		token: &env.token,
		password: &env.password,
		config_path: &config_path,
	};

	connect_repo(&kopia, &s3env, target, server_id).await?;

	if !ignore.is_empty() {
		let mut policy = build_kopia_command_with_s3(&kopia, &s3env).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).map_err(|e| miette!("{e}"))?;
	args_snapshot_create(&mut create, source_path, tags);
	let stdout = run_kopia(create, "snapshot create").await?;
	Ok(parse_snapshot_output(&stdout))
}

/// 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 sequence of hooks. `fail_fast` aborts on the first failure (pre-hooks);
/// otherwise failures are logged and the rest still run (post-hooks).
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: &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())));
	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());
	let bytes_uploaded = value
		.get("stats")
		.and_then(|s| s.get("totalSize"))
		.and_then(|v| v.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).
pub(super) async fn build_client(device_key: &str) -> Result<Arc<CanopyClient>> {
	let version = env!("CARGO_PKG_VERSION");
	let client = CanopyClient::new(version, Some(device_key), move || {
		bestool_canopy::client_builder(version)
	})
	.await?
	.ok_or_else(|| miette!("could not build a canopy client (no auth path available)"))?;
	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,
	}
}

fn trim_error(err: &miette::Report) -> String {
	let msg = format!("{err}");
	msg.chars().take(500).collect()
}

/// Per-type lockfile path, in a runtime dir (tmpfs on Linux, so it's cleared on
/// reboot and never stale across crashes).
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))]
	{
		std::env::temp_dir().join(format!("bestool-{name}"))
	}
}

/// 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 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, "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 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")
		);
	}

	#[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_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,
			}
		);
	}
}