rrdnsd/

lib.rs

// rrdnsd - monitoring and failover for Round Robin DNS records
// Copyright 2024-2025 Federico Ceratto <federico@debian.org>
// Released under AGPLv3

use anyhow::{bail, Context, Result};
use axum::extract::{Json, State};
use futures::future::join_all;
// TODO: switch to tracing?
use log::{debug, error, info, warn, LevelFilter};
use reqwest::redirect::Policy;
use reqwest::Client as Reqw;
use serde::{Deserialize, Serialize};
use statsd::client::Client as Statsd;
use std::collections::BTreeMap;
use std::collections::HashMap;
use std::fs;
use std::net::{IpAddr, SocketAddr};
use std::process; // TODO: tokio exit?
use std::str;
use std::time::Instant;
use std::time::SystemTime;
use systemd_journal_logger::JournalLog;
use tokio::signal::unix::{signal, SignalKind};
use tokio::sync::mpsc;
use tokio::sync::mpsc::{Receiver, Sender};
use tokio::time;
use tokio::time::Duration;
use tokio_util::sync::CancellationToken;

mod core;
mod dash;
mod providers;

use crate::providers::SetProvider;

#[cfg(test)]
mod tests;

const VERSION: &str = env!("CARGO_PKG_VERSION");

#[derive(PartialEq, Eq, Debug, Serialize, Deserialize, Clone, Copy)]
pub enum Status {
    Unknown,
    Down,
    Up,
}

#[derive(Debug, Serialize, Deserialize, Clone, Copy)]
struct Enst {
    status: Status,
    last_update_time: SystemTime,
}

/// Used to exchange message with peers
// TODO: dudupe QuorumMsg vs Datapoint
#[derive(Debug, Serialize, Deserialize, Clone)]
struct QuorumMsg {
    format_version: u8,
    fqdn: String,
    ipaddr: IpAddr,
    status: Status,
    tstamp: SystemTime,
    node_addr: NodeAddr,
}

#[derive(Debug, Clone, PartialEq, Eq, Ord, PartialOrd, Hash, Copy, Serialize, Deserialize)]
struct NodeAddr {
    ipa: IpAddr,
    port: u16,
}

/*
Arrows represent hashmaps:

ServiceStatus
 computed_status: Status
 computed_status_time: SystemTime
 dns_ttl: u32
 dns_zone: String
 fqdn: String
 endpoint_statuses[IpAddr] -> EndpointStatus
                                computed_status: Status
                                computed_status_time: SystemTime
                                node_to_ens[node_name] -> Enst
                                                            status: Status
                                                            last_update_time: SystemTime
*/

type NodeToEnst = HashMap<NodeAddr, Enst>;

#[derive(Debug, Clone)]
struct EndpointStatus {
    node_to_ens: NodeToEnst,
    computed_status: Status,
    computed_status_time: SystemTime,
    probe_cancel: tokio_util::sync::CancellationToken,
}

type EndpointStatuses = HashMap<IpAddr, EndpointStatus>;

#[derive(Debug, Clone)]
struct ServiceStatus {
    dns_ttl: u32,
    dns_zone: String,
    endpoint_statuses: EndpointStatuses,
    probe_interval: Duration,
    probe_target_url: String,
}

type FqdnToStatuses = HashMap<String, ServiceStatus>;

/// Internal message to the core reactor
#[derive(Debug)]
enum CoreMsg {
    Datapoint {
        datapoint: Datapoint,
    },
    Fetch {
        chan_dash_tx: DashTx,
    },
    APIChangeRequestWrap {
        request: APIChangeRequest,
        resp_chan_tx: tokio::sync::oneshot::Sender<APIChangeResponse>,
    },
}

/// Monitoring datapoint received from or set to other nodes
#[derive(Debug)]
struct Datapoint {
    fqdn: String,
    ipaddr: IpAddr,
    status: Status,
    tstamp: SystemTime,
    node_addr: NodeAddr,
}

// Channels
type DashTx = tokio::sync::oneshot::Sender<FqdnToStatuses>; // chan_dash_tx
type CoreTx = Sender<CoreMsg>; // chan_core_tx
type CoreRx = Receiver<CoreMsg>; // chan_core_rx
type OutboxTx = Sender<QuorumMsg>; // chan_outbox_tx
type OutboxRx = Receiver<QuorumMsg>; // chan_outbox_rx

#[derive(Debug, Clone)]
struct Shared {
    chan_core_tx: CoreTx,
}
type SharedState = axum::extract::State<Shared>;

macro_rules! unwrap_or_return {
    ($res:expr) => {
        match $res {
            Ok(val) => val,
            Err(e) => {
                error!("error: {}", e);
                return;
            }
        }
    };
}

// Like expect("..")
macro_rules! exp {
    ($res:expr) => {
        match $res {
            Ok(val) => val,
            Err(e) => {
                error!("{:?}", e);
                process::exit(1);
            }
        }
    };
}

// UNIX signal handlers

/// Shuts down the daemon when requested
async fn handle_sig(sig: SignalKind) {
    exp!(signal(sig)).recv().await;
    info!("[main] exiting");
    process::exit(0);
}

/// Statsd timer
fn timer(statsd: &Statsd, name: &str, t0: Instant) {
    let delta = t0.elapsed().as_millis() as f64;
    statsd.timer(name, delta);
}

// Configuration handling //

/// Service-specific configuration block
#[derive(Deserialize, Debug, Clone)]
struct ServiceConf {
    fqdn: String,
    healthcheck: String,
    ipaddrs: Vec<String>,
    ttl: u32,
    zone: String,
    probe_interval_ms: Option<u64>,
}

/// API configuration block

#[derive(Debug, Deserialize, Clone)]
struct APIConfig {
    enabled: bool,
}

/// How to send an update to the DNS resolver e.g. nsupdate
#[derive(Deserialize, Debug, Clone)]
#[serde(rename_all = "lowercase")]
enum ResolverUpdateMethod {
    Dynu,
    Dynv6,
    Nsupdate,
    Knsupdate,
    Script { path: String },
}

/// Configuration
#[derive(Deserialize, Debug, Clone)]
struct Conf {
    conf_version: u8,
    local_node: String,
    #[serde(skip_deserializing)]
    parsed_local_node: Option<NodeAddr>,
    nodes: Vec<String>,
    probe_interval_ms: Option<u64>,
    #[serde(skip_deserializing)]
    parsed_nodes: Vec<NodeAddr>,
    services: Vec<ServiceConf>,
    update_method: ResolverUpdateMethod,
    update_resolvers: Vec<String>,
    // TODO: parsed_resolvers
    update_credentials: String,
    enable_fail_open: bool,
    api: APIConfig,
}

fn trim_dots(s: &str) -> String {
    s.trim_start_matches('.').trim_end_matches('.').to_owned()
}

fn prepare_conf(jc: &str, local_node: Option<String>) -> Result<Conf> {
    let mut conf: Conf = serde_json::from_str(jc).context("Unable to parse configuration")?;
    debug!("[main] conf loaded");
    if conf.conf_version != 1 {
        bail!("Invalid configuration format version");
    }
    if let Some(ln) = local_node {
        conf.local_node = ln;
    }
    conf.parsed_local_node = Some(parse_ipaddr_port(&conf.local_node)?);
    conf.parsed_nodes = conf
        .nodes
        .iter()
        .map(|n| parse_ipaddr_port(n).expect("Unable to parse {n}"))
        .collect();
    for n in &conf.update_resolvers {
        parse_ipaddr_port(n)?;
    }
    for sv in &mut conf.services {
        sv.fqdn = trim_dots(&sv.fqdn);
        sv.zone = trim_dots(&sv.zone);
        if sv.fqdn.ends_with(&sv.zone) {
            continue;
        }
        bail!("Invalid fqdn '{}' not matching zone '{}'", sv.fqdn, sv.zone);
    }
    Ok(conf)
}

/// Loads configuration from file. Supports overriding the filename and local node name.
fn load_conf(conf_fn: &str, local_node: Option<String>) -> Result<Conf> {
    info!("[load_conf] reading {conf_fn}");
    let jc = fs::read_to_string(conf_fn).context("Unable to read config file")?;
    prepare_conf(&jc, local_node)
}

/// Outbox: receives from `chan_outbox_rx` and send updates to peers
async fn outbox_sender(mut chan_outbox_rx: OutboxRx, nodes: Vec<String>) {
    // let statsd = unwrap_or_return!(Statsd::new("127.0.0.1:8125", "rrdnsd"));
    // TODO reuse connections
    let client_ = Reqw::builder()
        .connect_timeout(Duration::new(5, 0))
        .pool_max_idle_per_host(9)
        .tcp_keepalive(Duration::new(15, 0))
        .timeout(Duration::new(5, 0))
        .user_agent("rrdnsd")
        .build();
    let client = unwrap_or_return!(client_);

    while let Some(msg) = chan_outbox_rx.recv().await {
        for node in &nodes {
            let client = client.clone();
            let msg = msg.clone();
            let url = format!("http://{node}/qmsg");
            let node = node.clone();
            tokio::spawn(async move {
                if let Err(e) = client.post(&url).json(&msg).send().await {
                    info!("Error sending msg to {node}: {e:?}");
                }
                // TODO generate metrics // statsd.incr("outbox_sent");
            });
        }
    }
    info!("[outbox] exiting");
    drop(chan_outbox_rx);
}

// Probe

/// Runs an HTTP[S] probe. Sends datapoints into `chan_outbox_tx`
async fn probe(
    interval: Duration,
    local_node: NodeAddr,
    endpoint_ipaddr: IpAddr,
    target_uri: String,
    fqdn: String,
    chan_outbox_tx: OutboxTx,
    cancel_probe: CancellationToken,
) {
    // FIXME stagger runs!
    // TODO make statsd ipaddr/port configurable
    // TODO export service status in statsd?
    let statsd = unwrap_or_return!(Statsd::new("127.0.0.1:8125", "rrdnsd"));

    // Port number in endpoint_saddr is ignored. Conventionally set to 0.
    // See https://docs.rs/reqwest/latest/reqwest/struct.ClientBuilder.html#method.resolve
    let endpoint_saddr = SocketAddr::new(endpoint_ipaddr, 0);

    // TODO configure timeouts
    info!("[probe] creating probe for uri {target_uri} at ipaddr {endpoint_saddr}");
    let client_ = Reqw::builder()
        .connect_timeout(Duration::new(5, 0))
        .pool_max_idle_per_host(2)
        .redirect(Policy::none())
        .resolve(&fqdn, endpoint_saddr)
        .tcp_keepalive(Duration::new(15, 0))
        .timeout(Duration::new(5, 0))
        .user_agent("rrdnsd")
        .build();
    let client = unwrap_or_return!(client_);

    debug!("[probe] starting");
    let mut ticker = time::interval(interval);
    loop {
        if cancel_probe.is_cancelled() {
            info!("[probe] exited probe for uri {target_uri} at ipaddr {endpoint_saddr}");
            return;
        }
        let t0 = Instant::now();
        ticker.tick().await;
        let t1 = Instant::now();

        debug!("[probe] probing {target_uri} at endpoint {endpoint_ipaddr}");
        let resp = client.get(&target_uri).send().await;
        let success = resp.is_ok();
        // statsd metric `probe.duration` - internal
        timer(&statsd, "probe.duration", t0);
        if success {
            // statsd metric `probe.success.cnt` - Successful service probes count
            statsd.incr("probe.success.cnt");
        } else {
            // statsd metric `probe.failure.cnt` - Failed service probes count
            statsd.incr("probe.failure.cnt");
        }
        let msg = QuorumMsg {
            format_version: 1,
            fqdn: fqdn.clone(),
            ipaddr: endpoint_ipaddr,
            status: if success { Status::Up } else { Status::Down },
            tstamp: SystemTime::now(),
            node_addr: local_node,
        };
        // TODO send datapoint to myself using channel instead of HTTP?
        if chan_outbox_tx.send(msg).await.is_err() {
            info!("[probe] send error");
            drop(chan_outbox_tx);
            return;
        }

        let loop_elapsed = t0.elapsed();
        let probing = t1.elapsed();
        // statsd metric `probe_time_msec` - Service probing elapsed time in milliseconds
        statsd.timer("probe_time_msec", probing.as_millis() as f64);
        let lf = probing.as_secs_f64() / loop_elapsed.as_secs_f64();
        // statsd metric `probe.load_factor` - debugging
        statsd.histogram("probe.load_factor", lf);
    }
}

/// Parses an `<ipaddr>:<port>` string
fn parse_ipaddr_port(n: &str) -> Result<NodeAddr> {
    let (ipa_str, port_str) = n.rsplit_once(':').context("Invalid ipaddr:port format")?;
    let ipa = ipa_str.parse::<IpAddr>().context("Invalid IP address")?;
    let port = port_str.parse::<u16>().context("Invalid port number")?;
    Ok(NodeAddr { ipa, port })
}

/// Generates a config/script file for nsupdate to add or delete a record
fn create_nsupdate_blob(
    resolver_addr: NodeAddr,
    fqdn: &str,
    zone: &str,
    ttl: u32,
    endpoint_ipaddr: &IpAddr,
    status: Status,
) -> String {
    let record_type = if endpoint_ipaddr.is_ipv4() {
        "A"
    } else {
        "AAAA"
    };

    let operation = match status {
        Status::Down => format!("delete {fqdn}. {ttl} {record_type} {endpoint_ipaddr}"),
        Status::Up => format!("add {fqdn}. {ttl} {record_type} {endpoint_ipaddr}"),
        Status::Unknown => "exit".to_string(),
    };
    format!(
        "server {ipa} {port}\nzone {zone}\nttl {ttl}\n{op}\nsend\nexit\n",
        ipa = resolver_addr.ipa,
        port = resolver_addr.port,
        zone = zone,
        ttl = ttl,
        op = operation,
    )
}

/// Updates resolver[s] using nsupdate or knsupdate
async fn run_nsupdates(
    update_method: &ResolverUpdateMethod,
    resolvers: &Vec<String>,
    srv: &ServiceStatus,
    fqdn: &str,
    endpoint_ipaddr: &IpAddr,
) -> Result<()> {
    // TODO pass status from caller
    let endpoint_that_changed = srv
        .endpoint_statuses
        .get(endpoint_ipaddr)
        .context("Endpoint not found")?;
    let status = endpoint_that_changed.computed_status;

    let cmd = match update_method {
        ResolverUpdateMethod::Nsupdate => "/usr/bin/nsupdate",
        ResolverUpdateMethod::Knsupdate => "/usr/bin/knsupdate",
        _ => bail!("Incorrect function call"),
    };

    for resolver in resolvers {
        let nsupdate_blob = create_nsupdate_blob(
            parse_ipaddr_port(resolver)?,
            fqdn,
            &srv.dns_zone,
            srv.dns_ttl,
            endpoint_ipaddr,
            status,
        );
        run_nsupdate_once(cmd, nsupdate_blob).await?;
    }
    Ok(())
}

/// Write configuration file and execute [k]nsupdate
#[cfg(not(test))]
async fn run_nsupdate_once(cmd: &str, conf: String) -> Result<()> {
    use tokio::process::Command;
    let tmpfile = tempfile::NamedTempFile::new().context("Failed to create tmpfile")?;
    let conf_fn = tmpfile.path();
    fs::write(conf_fn, &conf).context(format!("Failed to write {}", conf_fn.display()))?;
    info!(
        "[dns] calling {} {} with commands:\n{}",
        cmd,
        conf_fn.display(),
        conf
    );
    let out = Command::new(cmd).arg(conf_fn).output().await;
    match out {
        Ok(_) => info!("[dns] success"),
        Err(_) => warn!("[dns] failure"),
    }
    Ok(())
}

#[cfg(test)]
#[allow(clippy::unused_async)]
async fn run_nsupdate_once(cmd: &str, conf: String) -> Result<()> {
    info!("--cmd--{cmd}--conf--\n{conf}--");
    Ok(())
}

// Logic

/// Calculates quorum value e.g. 5 -> 3, 3 -> 1
#[must_use]
const fn quorum(n: u16) -> u16 {
    n.div_ceil(2)
}

/// Computes endpoint status using quorum (majority)
#[must_use]
#[allow(clippy::cast_possible_truncation)]
#[allow(clippy::match_same_arms)]
fn compute_status(endp_statuses: &NodeToEnst) -> Status {
    if endp_statuses.is_empty() {
        return Status::Unknown;
    }
    let q = quorum(endp_statuses.len() as u16);
    let up_cnt = endp_statuses
        .values()
        .filter(|ens| ens.status == Status::Up)
        .count() as u16;
    let down_cnt = endp_statuses
        .values()
        .filter(|ens| ens.status == Status::Down)
        .count() as u16;

    match (up_cnt >= q, down_cnt >= q) {
        (true, false) => Status::Up,
        (false, true) => Status::Down,
        (true, true) => Status::Up, // possible when the number of nodes is even
        (false, false) => Status::Unknown, // not enough datapoints yet
    }
}

/// Stores a probe datapoint. Returns `true` if a service endpoint change is detected.
fn ingest_datapoint(end: &mut EndpointStatus, msg: &Datapoint) -> bool {
    let Some(s) = end.node_to_ens.get_mut(&msg.node_addr) else {
        warn!("Received datapoint from unknown node {:?}", &msg.node_addr);
        return false;
    };

    s.status = msg.status;
    s.last_update_time = msg.tstamp;

    // Computes updated status
    let prev_computed_status = end.computed_status;
    end.computed_status = compute_status(&end.node_to_ens);
    end.computed_status_time = SystemTime::now();

    if prev_computed_status == Status::Unknown {
        return false;
    }
    if prev_computed_status == end.computed_status {
        return false;
    }
    true
}

/// This is called when an enpdoint is going down. Decide if we want to
/// delete its record from DNS or fail open (aka do nothing and keep it around).
#[must_use]
fn fail_open(estatuses: &HashMap<IpAddr, EndpointStatus>) -> bool {
    let up = estatuses
        .values()
        .filter(|s| s.computed_status == Status::Up)
        .count();
    let down = estatuses
        .values()
        .filter(|s| s.computed_status == Status::Down)
        .count();
    let unknown = estatuses
        .values()
        .filter(|s| s.computed_status == Status::Unknown)
        .count();

    // Threshold for majority
    let th = estatuses.len().div_ceil(2);
    let msg = format!("fail-open for: up {up} down {down} unknown {unknown} q {th}");
    {
        if unknown >= th {
            info!("{msg}: too many endpoints in unknown state");
            true
        } else if down >= th {
            info!("{msg}: too many endpoints in down state");
            true
        } else if up <= th {
            info!("{msg}: not enough endpoints in up state");
            true
        } else {
            debug!("No {msg}");
            false
        }
    }
}

/// Runs user-supplied command or script
async fn run_update_command(
    script_path: &str,
    endpoint_ipaddr: IpAddr,
    fqdn: &str,
    dns_zone: &str,
    status: Status,
) {
    let args = [
        &endpoint_ipaddr.to_string(),
        fqdn,
        dns_zone,
        &format!("{status:?}"),
    ];
    info!("[script] running with {:?}", &args);

    let mut cmd = tokio::process::Command::new(script_path);
    cmd.args(args);
    let Ok(output) = cmd.output().await else {
        error!("[script] failed to execute '{script_path}'");
        return;
    };
    if !output.status.success() {
        error!("[script] failed with {}", output.status);
    }
    let stderr = String::from_utf8_lossy(&output.stderr);
    let stdout = String::from_utf8_lossy(&output.stdout);
    if !stderr.is_empty() {
        error!("[script] stderr: {stderr}");
    }
    if !stdout.is_empty() {
        info!("[script] output: {stdout}");
    }
}

/// Endpoint status change detected: Update DNS
async fn handle_endpoint_change(
    conf: &Conf,
    srv: &ServiceStatus,
    endpoint_ipaddr: IpAddr,
    fqdn: &str,
) -> Result<()> {
    let Some(endpoint_that_changed) = srv.endpoint_statuses.get(&endpoint_ipaddr) else {
        return Ok(());
    };
    let status = endpoint_that_changed.computed_status;
    if status == Status::Unknown {
        return Ok(());
    }
    if (status == Status::Down) & conf.enable_fail_open & fail_open(&srv.endpoint_statuses) {
        // Do not delete the endpoint record
        // TODO: log
        return Ok(());
    }

    let zone = &srv.dns_zone;
    match &conf.update_method {
        // HTTP DNS APIs
        ResolverUpdateMethod::Dynu => match status {
            Status::Up => {
                providers::Dynu::add_record(&conf.update_credentials, fqdn, zone, endpoint_ipaddr)
                    .await?;
            }
            Status::Down => {
                providers::Dynu::delete_record(
                    &conf.update_credentials,
                    fqdn,
                    zone,
                    endpoint_ipaddr,
                )
                .await?;
            }
            Status::Unknown => {}
        },
        ResolverUpdateMethod::Dynv6 => match status {
            Status::Up => {
                providers::Dynv6::add_record(&conf.update_credentials, fqdn, zone, endpoint_ipaddr)
                    .await?;
            }
            Status::Down => {
                providers::Dynv6::delete_record(
                    &conf.update_credentials,
                    fqdn,
                    zone,
                    endpoint_ipaddr,
                )
                .await?;
            }
            Status::Unknown => {}
        },

        ResolverUpdateMethod::Nsupdate | ResolverUpdateMethod::Knsupdate => {
            run_nsupdates(
                &conf.update_method,
                &conf.update_resolvers,
                srv,
                fqdn,
                &endpoint_ipaddr,
            )
            .await?;
        }
        ResolverUpdateMethod::Script { path } => {
            run_update_command(path, endpoint_ipaddr, fqdn, &srv.dns_zone, status).await;
        }
    }
    Ok(())
}

// HTTP routes //

async fn fetch_status(chan_core_tx: CoreTx) -> Result<FqdnToStatuses> {
    // send a request to core_task and wait
    let (chan_dash_tx, chan_dash_rx) = tokio::sync::oneshot::channel();
    let wu = CoreMsg::Fetch { chan_dash_tx };
    chan_core_tx.send(wu).await?;
    Ok(chan_dash_rx.await?)
}

async fn handle_http_get_dashboard(State(shared): SharedState) -> axum::response::Html<String> {
    let resp = fetch_status(shared.chan_core_tx).await;
    match resp {
        Err(e) => {
            error!("dashboard: failed to fetch: {e}");
            axum::response::Html("Error".into())
        }
        Ok(r) => {
            let dash = dash::render_html_dashboard(&r);
            axum::response::Html(dash)
        }
    }
}

async fn handle_http_get_health() -> &'static str {
    "ok"
}

async fn handle_http_post_qmsg(State(shared): SharedState, Json(msg): Json<QuorumMsg>) {
    // send to core
    let wu = CoreMsg::Datapoint {
        datapoint: Datapoint {
            fqdn: msg.fqdn.to_string(),
            ipaddr: msg.ipaddr,
            status: msg.status,
            tstamp: msg.tstamp,
            node_addr: msg.node_addr,
        },
    };
    if shared.chan_core_tx.send(wu).await.is_err() {
        info!("[http] exiting");
        // TODO handle error
    };
}

// HTTP routes: API //

// API - List endpoints

/// Represents a service under monitoring including the state of each endpoint
#[derive(Serialize, Deserialize, Debug, Eq, PartialEq)]
pub struct APIListItem {
    pub ipaddrs: HashMap<IpAddr, Status>,
    dns_ttl: u32,
    dns_zone: String,
    probe_interval_ms: u128,
    probe_target_url: String,
}

pub type APIListOut = BTreeMap<String, APIListItem>;

/// Lists monitored services. Includes the current state of each endpoint
async fn handle_http_api_list(State(shared): SharedState) -> Json<APIListOut> {
    let resp = fetch_status(shared.chan_core_tx).await.unwrap();

    let out = resp
        .into_iter()
        .map(|(fqdn, srvs)| {
            // collect ipaddr -> status hashmap
            let ipaddrs = srvs
                .endpoint_statuses
                .into_iter()
                .map(|(srv_ipaddr, ep)| (srv_ipaddr, ep.computed_status))
                .collect();

            (
                fqdn,
                APIListItem {
                    ipaddrs,
                    dns_ttl: srvs.dns_ttl,
                    dns_zone: srvs.dns_zone,
                    probe_interval_ms: srvs.probe_interval.as_millis(),
                    probe_target_url: srvs.probe_target_url,
                },
            )
        })
        .collect();

    Json(out)
}

// API - Update endpoints

/// Struct sent to the API to modify the running configuration: add or remove services or endpoints
#[derive(Debug, Serialize, Deserialize, Clone)]
#[serde(tag = "method", rename_all = "snake_case")]
pub enum APIRPCUpdateAction {
    CreateService {
        fqdn: String,
        healthcheck: String,
        ttl: u32,
        zone: String,
        probe_interval_ms: u64,
    },
    DeleteService {
        fqdn: String,
    },
    AddIpAddr {
        fqdn: String,
        ipaddr: IpAddr,
    },
    DeleteIpAddr {
        fqdn: String,
        ipaddr: IpAddr,
    },
}

#[derive(Deserialize, Serialize, Debug, Clone)]
pub struct APIChangeRequest {
    pub actions: Vec<APIRPCUpdateAction>,
}

#[derive(Debug, Serialize, Deserialize, Eq, PartialEq)]
#[serde(tag = "status", rename_all = "lowercase")]
pub enum APIChangeResult {
    Ok,
    Error { error: String },
}

#[derive(Debug, Serialize, Deserialize, Eq, PartialEq)]
pub struct APIChangeResponse {
    results: Vec<APIChangeResult>,
}

impl APIChangeResponse {
    pub fn all_ok(&self) -> bool {
        self.results
            .iter()
            .all(|r| matches!(r, APIChangeResult::Ok))
    }
}

// HTTP API receives an update request and sends it into chan_core_tx
async fn handle_http_api_update(
    State(shared): SharedState,
    axum::Json(req): Json<APIChangeRequest>,
) -> axum::Json<APIChangeResponse> {
    let (resp_chan_tx, resp_chan_rx) = tokio::sync::oneshot::channel();
    let wu = CoreMsg::APIChangeRequestWrap {
        request: req,
        resp_chan_tx,
    };
    debug!("sending API change request into chan_core_tx");
    let _ = shared.chan_core_tx.send(wu).await;
    debug!("waiting from resp_chan_rx");
    let resp = resp_chan_rx
        .await
        .unwrap_or_else(|_| APIChangeResponse { results: vec![] });

    debug!("received a reply from resp_chan_rx");
    Json(resp)
}

// end of API routes

fn create_router(chan_core_tx: CoreTx, api_enabled: bool) -> axum::Router {
    let r = axum::Router::new()
        .route("/dash", axum::routing::get(handle_http_get_dashboard))
        .route("/health", axum::routing::get(handle_http_get_health))
        .route("/qmsg", axum::routing::post(handle_http_post_qmsg));

    if api_enabled {
        r.route(
            "/api/v0/list_endpoints",
            axum::routing::get(handle_http_api_list),
        )
        .route(
            "/api/v0/update_endpoints",
            axum::routing::post(handle_http_api_update),
        )
    } else {
        r
    }
    .with_state(Shared { chan_core_tx })
}

async fn wait_for_shutdown(token: CancellationToken) {
    token.cancelled().await;
    debug!("Shutdown token triggered. Axum server exiting.");
}

async fn run_axum_server(
    http_listener: tokio::net::TcpListener,
    http_router: axum::Router,
    shutdown_token: CancellationToken,
) {
    if let Err(e) = axum::serve(http_listener, http_router)
        .with_graceful_shutdown(wait_for_shutdown(shutdown_token))
        .await
    {
        error!("Axum server failed: {e}");
        process::exit(1);
    }
}

// init section //
pub async fn init(
    conf_fn: &str,
    local_node: Option<String>,
    tasks: &mut Vec<tokio::task::JoinHandle<()>>,
) -> Result<CancellationToken> {
    // Load and parse conf
    let conf = load_conf(conf_fn, local_node).unwrap_or_else(|err| {
        error!("Failed to load config: {err}");
        process::exit(1)
    });

    // Init
    let statsd = Statsd::new("127.0.0.1:8125", "rrdnsd")?;
    let service_statuses = HashMap::new();

    // Setup channels
    let (chan_core_tx, chan_core_rx): (CoreTx, CoreRx) = mpsc::channel(10);
    let (chan_outbox_tx, chan_outbox_rx): (OutboxTx, OutboxRx) = mpsc::channel(100);

    // Outbox sender
    let sender = outbox_sender(chan_outbox_rx, conf.nodes.clone());
    tasks.push(tokio::spawn(sender));

    // Core reactor
    let core_t = core::core(
        statsd,
        conf.clone(),
        service_statuses,
        chan_core_rx,
        chan_outbox_tx,
    );
    tasks.push(tokio::spawn(core_t));

    time::sleep(Duration::from_millis(100)).await;

    // Create cancellation token
    let axum_shutdown = CancellationToken::new();
    let shutdown_token_child = axum_shutdown.child_token();

    // HTTP receiver setup
    debug!("creating listener");
    let http_router = create_router(chan_core_tx.clone(), conf.api.enabled);
    let http_listener = tokio::net::TcpListener::bind(conf.local_node.clone()).await?;

    // Spawn Axum server
    debug!("starting axum");
    let fut = tokio::spawn(run_axum_server(
        http_listener,
        http_router,
        shutdown_token_child,
    ));

    tasks.push(fut);

    // Scan config to setup services and spawn probes
    debug!("starting probes");
    setup_and_spawn_probes(conf, chan_core_tx).await?;
    debug!("starting probes: done");

    Ok(axum_shutdown)
}

pub async fn run() -> Result<()> {
    // Setup logging and metrics
    JournalLog::new()
        .context("Failed to create JournalLog")?
        .install()
        .context("Failed to setup journald logging")?;

    // TODO: configurable loglevel
    log::set_max_level(LevelFilter::Info);
    info!("[main] starting");

    // Start signal handlers
    let mut tasks = Vec::new();
    for s in [SignalKind::interrupt(), SignalKind::hangup()] {
        tasks.push(tokio::spawn(handle_sig(s)));
    }

    let default_conf_fn = "/etc/rrdnsd.json";
    let conf_fn = std::env::var("CONF").ok();
    let conf_fn = conf_fn.as_deref().unwrap_or(default_conf_fn);

    let local_node = std::env::var("LOCAL_NODE").ok();

    init(conf_fn, local_node, &mut tasks).await?;

    join_all(tasks).await;
    Ok(())
}

/// Setup services from conf file and spawn related probes
async fn setup_and_spawn_probes(conf: Conf, chan_core_tx: Sender<CoreMsg>) -> Result<()> {
    for service in &conf.services {
        let probe_interval_ms = service
            .probe_interval_ms
            .or(conf.probe_interval_ms)
            .unwrap_or(1000);

        let request = APIChangeRequest {
            actions: vec![APIRPCUpdateAction::CreateService {
                fqdn: service.fqdn.clone(),
                healthcheck: service.healthcheck.clone(),
                ttl: service.ttl,
                zone: service.zone.clone(),
                probe_interval_ms,
            }],
        };
        send_api_change_request_then_check_resp(&chan_core_tx, request).await?;

        for ipaddr_str in &service.ipaddrs {
            let request = APIChangeRequest {
                actions: vec![APIRPCUpdateAction::AddIpAddr {
                    fqdn: service.fqdn.clone(),
                    ipaddr: ipaddr_str.parse()?,
                }],
            };
            send_api_change_request_then_check_resp(&chan_core_tx, request).await?;
        }
    }
    Ok(())
}

async fn send_api_change_request_then_check_resp(
    chan_core_tx: &Sender<CoreMsg>,
    request: APIChangeRequest,
) -> Result<(), anyhow::Error> {
    let (resp_chan_tx, resp_chan_rx) = tokio::sync::oneshot::channel();
    let msg = CoreMsg::APIChangeRequestWrap {
        request: request.clone(),
        resp_chan_tx,
    };
    chan_core_tx.send(msg).await?;
    let resp = resp_chan_rx.await?;

    for (action, result) in request.actions.iter().zip(resp.results.iter()) {
        if let APIChangeResult::Error { error } = result {
            error!("Action {action:?} failed: {error}");
        }
    }

    Ok(())
}