use std::{
process::Command,
time::{Duration, Instant},
};
use miette::{IntoDiagnostic, Result, bail};
use tracing::{debug, info, warn};
use bestool_tamanu::{
ApiServerKind,
config::load_config,
pm2,
server_info::query_patient_portal_enabled,
services::{self, Expectation, Supervisor, parse_systemd_unit, systemd_patient_portal_instanced},
systemd,
};
use super::{TamanuArgs, find_tamanu};
#[derive(Copy, Clone, Debug, PartialEq, Eq)]
pub enum WaitForDb {
No,
Yes,
}
const DB_WAIT_TIMEOUT: Duration = Duration::from_secs(120);
const DB_WAIT_INTERVAL: Duration = Duration::from_secs(2);
pub async fn config_and_expectations(
tamanu: &TamanuArgs,
wait_for_db: WaitForDb,
) -> Result<(Supervisor, Vec<Expectation>)> {
let supervisor = if cfg!(target_os = "linux") {
Supervisor::Systemd
} else if cfg!(target_os = "windows") {
Supervisor::Pm2
} else {
bail!("tamanu lifecycle commands are only supported on Linux (systemd) and Windows (pm2)");
};
let (_, root) = find_tamanu(tamanu)?;
let config = load_config(&root, None)?;
let kind = if config.is_facility() {
ApiServerKind::Facility
} else {
ApiServerKind::Central
};
let patient_portal_enabled =
if matches!(supervisor, Supervisor::Systemd) && matches!(kind, ApiServerKind::Central) {
query_patient_portal_enabled_with_wait(&config.database_url(), wait_for_db).await
} else {
Some(false)
};
let patient_portal_instanced = matches!(supervisor, Supervisor::Systemd)
&& matches!(kind, ApiServerKind::Central)
&& systemd_patient_portal_instanced().await;
let expectations = services::expected(
supervisor,
kind,
&config,
patient_portal_enabled,
patient_portal_instanced,
);
Ok((supervisor, expectations))
}
async fn query_patient_portal_enabled_with_wait(
database_url: &str,
wait_for_db: WaitForDb,
) -> Option<bool> {
query_patient_portal_enabled_with_wait_inner(
database_url,
wait_for_db,
DB_WAIT_TIMEOUT,
DB_WAIT_INTERVAL,
)
.await
}
async fn query_patient_portal_enabled_with_wait_inner(
database_url: &str,
wait_for_db: WaitForDb,
timeout: Duration,
interval: Duration,
) -> Option<bool> {
let deadline = match wait_for_db {
WaitForDb::No => None,
WaitForDb::Yes => Some(Instant::now() + timeout),
};
loop {
match bestool_postgres::pool::connect_one(database_url, "bestool-tamanu-lifecycle").await {
Ok(client) => return query_patient_portal_enabled(&client).await,
Err(err) => match deadline {
None => {
warn!(%err, "could not query features.patientPortal; expectation will be Unknown");
return None;
}
Some(deadline) if Instant::now() >= deadline => {
warn!(
%err,
"timed out after {}s waiting for the database; features.patientPortal expectation will be Unknown",
timeout.as_secs(),
);
return None;
}
Some(_) => {
warn!(
%err,
"database not ready; retrying in {}s (will give up after {}s total)",
interval.as_secs(),
timeout.as_secs(),
);
tokio::time::sleep(interval).await;
}
},
}
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Instance {
pub name: String,
pub instance: Option<String>,
pub pm_id: Option<i64>,
pub running: bool,
}
impl Instance {
pub fn unit(&self) -> String {
match &self.instance {
Some(i) => format!("{}@{}.service", self.name, i),
None => format!("{}.service", self.name),
}
}
pub fn display(&self) -> String {
match (&self.instance, self.pm_id) {
(Some(i), _) => format!("{}@{}", self.name, i),
(None, Some(id)) => format!("{}#{id}", self.name),
(None, None) => self.name.clone(),
}
}
}
pub async fn discover(supervisor: Supervisor) -> Result<Vec<Instance>> {
match supervisor {
Supervisor::Systemd => discover_systemd().await,
Supervisor::Pm2 => discover_pm2().map(|(v, _)| v),
}
}
async fn discover_systemd() -> Result<Vec<Instance>> {
let units = systemd::list_units(&["tamanu-*.service"]).await?;
let mut out = Vec::new();
for u in units {
let Some((base, instance)) = parse_systemd_unit(&u.name) else {
continue;
};
out.push(Instance {
name: base.to_string(),
instance: instance.map(str::to_string),
pm_id: None,
running: u.running(),
});
}
Ok(out)
}
fn discover_pm2() -> Result<(Vec<Instance>, pm2::Source)> {
let (procs, source) = pm2::list().map_err(|e| miette::miette!("pm2: {e}"))?;
let mut out = Vec::new();
for p in procs {
if !p.name.starts_with("tamanu-") {
continue;
}
out.push(Instance {
name: p.name,
instance: None,
pm_id: p.pm_id,
running: p.running,
});
}
Ok((out, source))
}
pub fn warn_unknown_expectations(expectations: &[&Expectation]) {
for exp in expectations {
if matches!(exp.state, services::ExpectedState::Unknown) {
warn!(
name = exp.name,
reason = %exp.reason,
"leaving service alone: expected state is Unknown"
);
}
}
}
pub fn group_by_expectation<'a>(
expectations: &'a [&'a Expectation],
instances: &[Instance],
) -> Vec<(&'a Expectation, Vec<Instance>)> {
expectations
.iter()
.map(|exp| {
let matches: Vec<Instance> = instances
.iter()
.filter(|d| {
d.name == exp.name && exp.instances.admits_instance(d.instance.as_deref())
})
.cloned()
.collect();
(*exp, matches)
})
.collect()
}
pub fn ensure_root_or_reexec(supervisor: Supervisor) -> Result<()> {
if !matches!(supervisor, Supervisor::Systemd) {
return Ok(());
}
if privilege::user::privileged() {
return Ok(());
}
info!("not running as root; re-execing under sudo");
let args: Vec<String> = std::env::args().collect();
let status = Command::new("sudo").args(args).status().into_diagnostic()?;
std::process::exit(status.code().unwrap_or(1));
}
pub async fn wait_running(supervisor: Supervisor, targets: &[String]) -> Result<()> {
wait_for(supervisor, targets, true, "active").await
}
pub async fn wait_stopped(supervisor: Supervisor, targets: &[String]) -> Result<()> {
wait_for(supervisor, targets, false, "inactive").await
}
pub async fn stop_targets(supervisor: Supervisor, targets: &[String]) -> Result<()> {
if targets.is_empty() {
return Ok(());
}
match supervisor {
Supervisor::Systemd => systemd::stop(targets).await,
Supervisor::Pm2 => {
let status = Command::new("pm2")
.arg("stop")
.args(targets)
.status()
.into_diagnostic()?;
if !status.success() {
bail!("pm2 stop failed: exit {status}");
}
Ok(())
}
}
}
pub fn pm2_restart_targets(targets: &[String]) -> Result<()> {
if targets.is_empty() {
return Ok(());
}
let stop = Command::new("pm2")
.arg("stop")
.args(targets)
.status()
.into_diagnostic()?;
if !stop.success() {
bail!("pm2 stop failed: exit {stop}");
}
std::thread::sleep(Duration::from_secs(1));
let start = Command::new("pm2")
.arg("start")
.args(targets)
.status()
.into_diagnostic()?;
if !start.success() {
bail!("pm2 start failed: exit {start}");
}
Ok(())
}
pub fn delete_pm2(names: &[String]) -> Result<()> {
if names.is_empty() {
return Ok(());
}
let status = Command::new("pm2")
.arg("delete")
.args(names)
.status()
.into_diagnostic()?;
if !status.success() {
bail!("pm2 delete failed: exit {status}");
}
Ok(())
}
pub async fn disable_systemd_units(units: &[String]) -> Result<()> {
systemd::disable(units).await
}
async fn wait_for(
supervisor: Supervisor,
targets: &[String],
want_running: bool,
state_label: &str,
) -> Result<()> {
let deadline = Instant::now() + Duration::from_secs(60);
let interval = Duration::from_millis(500);
loop {
let mut all_match = true;
for t in targets {
if is_running(supervisor, t).await != want_running {
all_match = false;
break;
}
}
if all_match {
return Ok(());
}
if Instant::now() >= deadline {
let mut still_wrong: Vec<&str> = Vec::new();
for t in targets {
if is_running(supervisor, t).await != want_running {
still_wrong.push(t.as_str());
}
}
bail!(
"timed out after 60s waiting for {} to become {state_label}",
still_wrong.join(", ")
);
}
tokio::time::sleep(interval).await;
}
}
pub async fn restart_one(supervisor: Supervisor, instance: &Instance) -> Result<()> {
match supervisor {
Supervisor::Systemd => systemd::restart(&instance.unit()).await,
Supervisor::Pm2 => {
let id = instance
.pm_id
.ok_or_else(|| miette::miette!("pm2 instance {} has no pm_id", instance.name))?;
pm2_restart_targets(&[id.to_string()])
}
}
}
pub async fn wait_running_one(
supervisor: Supervisor,
instance: &Instance,
timeout: Duration,
) -> Result<()> {
let deadline = Instant::now() + timeout;
let interval = Duration::from_millis(500);
loop {
let up = match supervisor {
Supervisor::Systemd => is_running(supervisor, &instance.unit()).await,
Supervisor::Pm2 => is_pm2_pm_id_online(instance.pm_id),
};
if up {
return Ok(());
}
if Instant::now() >= deadline {
bail!(
"timed out after {}s waiting for {} to become active",
timeout.as_secs(),
instance.display(),
);
}
tokio::time::sleep(interval).await;
}
}
fn is_pm2_pm_id_online(pm_id: Option<i64>) -> bool {
let Some(id) = pm_id else { return false };
match pm2::list() {
Ok((procs, _)) => procs.iter().any(|p| p.pm_id == Some(id) && p.running),
Err(_) => false,
}
}
#[cfg(target_os = "linux")]
pub async fn reload_caddy() {
match systemd::reload("caddy.service").await {
Ok(()) => debug!("caddy reloaded"),
Err(e) => warn!("could not reload caddy: {e}"),
}
let status = Command::new("resolvectl").arg("flush-caches").status();
match status {
Ok(s) if s.success() => debug!("resolvectl flush-caches OK"),
Ok(s) => warn!("resolvectl flush-caches exited with {s}"),
Err(e) => debug!("resolvectl not available: {e}"),
}
}
#[cfg(target_os = "windows")]
pub async fn reload_caddy() {
let path = r"C:\Caddy\Caddyfile";
match reload_caddy_via_admin_api(path).await {
Ok(()) => {
debug!("caddy reloaded via admin API");
return;
}
Err(err) => debug!(%err, "caddy admin API unreachable, falling back to CLI"),
}
let status = Command::new(bestool_tamanu::caddy::program())
.args(["reload", "--config", path, "--adapter", "caddyfile"])
.status();
match status {
Ok(s) if s.success() => debug!("caddy reloaded via CLI"),
Ok(s) => warn!("caddy reload exited with {s}"),
Err(e) => warn!("could not run caddy reload: {e}"),
}
}
#[cfg(not(any(target_os = "linux", target_os = "windows")))]
pub async fn reload_caddy() {
debug!("caddy reload is unsupported on this OS");
}
#[cfg(target_os = "windows")]
async fn reload_caddy_via_admin_api(path: &str) -> std::result::Result<(), String> {
let content = std::fs::read(path).map_err(|e| format!("read {path}: {e}"))?;
let client = crate::http::client_builder()
.timeout(Duration::from_secs(5))
.build()
.map_err(|e| format!("build client: {e}"))?;
let resp = client
.post("http://localhost:2019/load")
.header("Content-Type", "text/caddyfile")
.body(content)
.send()
.await
.map_err(|e| format!("POST localhost:2019/load: {e}"))?;
if !resp.status().is_success() {
let status = resp.status();
let body = resp.text().await.unwrap_or_default();
return Err(format!("admin API returned {status}: {body}"));
}
Ok(())
}
pub fn container_ip_for_unit(unit: &str) -> Result<Option<std::net::IpAddr>> {
let ps = Command::new("podman")
.args([
"ps",
"--filter",
&format!("label=PODMAN_SYSTEMD_UNIT={unit}"),
"--format",
"json",
])
.output();
let ps = match ps {
Ok(o) if o.status.success() => o,
Ok(o) => {
warn!(
"podman ps failed: {}",
String::from_utf8_lossy(&o.stderr).trim()
);
return Ok(None);
}
Err(e) => {
debug!("podman not available: {e}");
return Ok(None);
}
};
let entries: Vec<serde_json::Value> = serde_json::from_slice(&ps.stdout).into_diagnostic()?;
let Some(id) = entries.first().and_then(|c| c["Id"].as_str()) else {
return Ok(None);
};
let inspect = Command::new("podman")
.args(["inspect", id])
.output()
.into_diagnostic()?;
if !inspect.status.success() {
bail!(
"podman inspect failed: {}",
String::from_utf8_lossy(&inspect.stderr).trim()
);
}
let inspects: Vec<serde_json::Value> =
serde_json::from_slice(&inspect.stdout).into_diagnostic()?;
let ip = inspects
.first()
.and_then(|c| c["NetworkSettings"]["Networks"].as_object())
.and_then(|nets| nets.values().find_map(|n| n["IPAddress"].as_str()))
.filter(|s| !s.is_empty())
.map(|s| s.parse::<std::net::IpAddr>())
.transpose()
.into_diagnostic()?;
Ok(ip)
}
pub fn pm2_port_for(pm_id: i64) -> Result<Option<u16>> {
let output = Command::new("pm2").arg("jlist").output().into_diagnostic()?;
if !output.status.success() {
bail!(
"pm2 jlist failed: {}",
String::from_utf8_lossy(&output.stderr).trim()
);
}
let entries: Vec<serde_json::Value> =
serde_json::from_slice(&output.stdout).into_diagnostic()?;
let port = entries
.iter()
.find(|p| p["pm_id"].as_i64() == Some(pm_id))
.and_then(|p| p["pm2_env"].get("env"))
.and_then(|env| env.get("PORT"))
.and_then(|v| {
v.as_str()
.and_then(|s| s.parse().ok())
.or_else(|| v.as_u64().and_then(|n| u16::try_from(n).ok()))
});
Ok(port)
}
async fn is_running(supervisor: Supervisor, target: &str) -> bool {
match supervisor {
Supervisor::Systemd => systemd::is_active(target).await.unwrap_or(false),
Supervisor::Pm2 => match pm2::list() {
Ok((procs, _)) => procs.iter().any(|p| p.name == target && p.running),
Err(_) => false,
},
}
}
#[cfg(test)]
mod tests {
use super::*;
use bestool_tamanu::services::{ExpectedState, Instances};
fn exp(name: &'static str) -> Expectation {
Expectation {
name,
instances: Instances::Single,
state: ExpectedState::Up,
reason: "test".into(),
legacy: false,
behind_caddy: false,
}
}
fn templated_exp(name: &'static str) -> Expectation {
Expectation {
name,
instances: Instances::NumericAtLeast(2),
state: ExpectedState::Up,
reason: "test".into(),
legacy: false,
behind_caddy: false,
}
}
fn inst(name: &str, instance: Option<&str>, running: bool) -> Instance {
Instance {
name: name.to_string(),
instance: instance.map(str::to_string),
pm_id: None,
running,
}
}
#[test]
fn group_by_expectation_collects_matching_instances() {
let api = templated_exp("tamanu-central-api");
let tasks = exp("tamanu-central-tasks");
let expectations = [&api, &tasks];
let instances = vec![
inst("tamanu-central-api", Some("1"), true),
inst("tamanu-central-api", Some("2"), true),
inst("tamanu-central-tasks", None, true),
inst("tamanu-orphan", None, true), ];
let groups = group_by_expectation(&expectations, &instances);
assert_eq!(groups.len(), 2);
assert_eq!(groups[0].0.name, "tamanu-central-api");
assert_eq!(groups[0].1.len(), 2);
assert_eq!(groups[1].0.name, "tamanu-central-tasks");
assert_eq!(groups[1].1.len(), 1);
}
#[test]
fn instance_unit_and_display() {
let templated = inst("tamanu-frontend", Some("a"), true);
assert_eq!(templated.unit(), "tamanu-frontend@a.service");
assert_eq!(templated.display(), "tamanu-frontend@a");
let singleton = inst("tamanu-tasks", None, true);
assert_eq!(singleton.unit(), "tamanu-tasks.service");
assert_eq!(singleton.display(), "tamanu-tasks");
let pm2 = Instance {
name: "tamanu-api".into(),
instance: None,
pm_id: Some(3),
running: true,
};
assert_eq!(pm2.display(), "tamanu-api#3");
}
const UNREACHABLE_DB_URL: &str = "postgres://localhost:1/x";
#[tokio::test]
async fn wait_for_db_no_returns_none_after_single_attempt() {
let start = Instant::now();
let result = query_patient_portal_enabled_with_wait_inner(
UNREACHABLE_DB_URL,
WaitForDb::No,
Duration::from_secs(10),
Duration::from_millis(50),
)
.await;
assert_eq!(result, None, "DB unreachable must surface as Unknown, not a guess");
assert!(
start.elapsed() < Duration::from_secs(5),
"WaitForDb::No should not loop; took {:?}",
start.elapsed(),
);
}
#[tokio::test]
async fn wait_for_db_yes_retries_until_timeout() {
let start = Instant::now();
let result = query_patient_portal_enabled_with_wait_inner(
UNREACHABLE_DB_URL,
WaitForDb::Yes,
Duration::from_millis(300),
Duration::from_millis(50),
)
.await;
assert_eq!(result, None, "should surface as Unknown after timeout");
assert!(
start.elapsed() >= Duration::from_millis(300),
"should have waited at least the timeout; took {:?}",
start.elapsed(),
);
}
}