neiam /apollos-dbus

//! One broker connection per source, for as long as the daemon runs.
//!
//! Nothing is polled. An Ankyra's Pythiae publishes on change, a Keryx on its
//! own tick; this subscribes, asks for a fresh copy once on every connect --
//! a desktop waking from suspend is looking at whatever it last heard -- and
//! otherwise listens.

use std::time::Duration;

use rumqttc::{
    AsyncClient, Event as MqttEvent, MqttOptions, Packet, QoS, TlsConfiguration, Transport,
};
use serde_json::{Value, json};
use tokio::sync::mpsc;
use tracing::{debug, info, warn};

use crate::config::{Kind, Source};
use crate::wire::{self, Reading};

#[derive(Debug)]
pub enum Event {
    Status {
        source: String,
        connected: bool,
        error: Option<String>,
    },
    /// Readings from a source. `whole` when they are everything it has --
    /// an Ankyra board is, so a query missing from it has gone.
    Readings {
        source: String,
        readings: Vec<Reading>,
        whole: bool,
    },
    Plus {
        entries: Vec<(String, Value)>,
    },
}

#[derive(Debug)]
pub enum Command {
    /// Ask for fresh data: one query by id, or everything.
    Refresh(Option<u32>),
}

/// Boards carry a whole vision, and the default limit is 10 KiB.
const MAX_PACKET: usize = 4 * 1024 * 1024;
const MAX_BACKOFF: Duration = Duration::from_secs(60);

pub async fn run(
    source: Source,
    client_id: String,
    events: mpsc::Sender<Event>,
    mut commands: mpsc::Receiver<Command>,
) {
    let label = source.label();
    let b = &source.broker;

    let mut opts = MqttOptions::new(client_id, &b.host, b.port);
    opts.set_keep_alive(Duration::from_secs(30));
    opts.set_max_packet_size(MAX_PACKET, MAX_PACKET);
    if !b.username.is_empty() {
        opts.set_credentials(&b.username, &b.password);
    }
    if b.tls {
        // The platform's TLS (OpenSSL on Linux): the system roots, plus
        // whatever SSL_CERT_FILE / SSL_CERT_DIR name, and the host name checked.
        opts.set_transport(Transport::tls_with_config(TlsConfiguration::Native));
    }

    let (client, mut eventloop) = AsyncClient::new(opts, 32);
    let mut backoff = Duration::from_secs(1);

    loop {
        tokio::select! {
            polled = eventloop.poll() => match polled {
                Ok(MqttEvent::Incoming(Packet::ConnAck(_))) => {
                    info!("{label}: connected to {}:{}", b.host, b.port);
                    backoff = Duration::from_secs(1);
                    subscribe(&client, &source.kind).await;
                    request(&client, &source.kind, None).await;
                    send(&events, Event::Status { source: label.clone(), connected: true, error: None }).await;
                }
                Ok(MqttEvent::Incoming(Packet::Publish(p))) => {
                    if let Some(event) = interpret(&source, &label, &p.topic, &p.payload) {
                        send(&events, event).await;
                    }
                }
                Ok(_) => {}
                Err(e) => {
                    warn!("{label}: {e}; retrying in {}s", backoff.as_secs());
                    send(&events, Event::Status { source: label.clone(), connected: false, error: Some(e.to_string()) }).await;
                    tokio::time::sleep(backoff).await;
                    backoff = (backoff * 2).min(MAX_BACKOFF);
                }
            },
            Some(Command::Refresh(query_id)) = commands.recv() => {
                request(&client, &source.kind, query_id).await;
            }
        }
    }
}

async fn send(events: &mpsc::Sender<Event>, event: Event) {
    // Only fails if the store has gone, and then so has everything else.
    let _ = events.send(event).await;
}

async fn subscribe(client: &AsyncClient, kind: &Kind) {
    let filters: Vec<String> = match kind {
        // The Plus topic under both spellings: which one the broker matches
        // a dotted MQTT filter against is its business, and only one of the
        // two will ever deliver.
        Kind::Ankyra { topic } => vec![
            topic.clone(),
            format!("{topic}.plus"),
            format!("{topic}/plus"),
        ],
        // Only the condensed reading. Everything else under a Keryx -- the
        // raw results, one message per item, per route -- is for consumers
        // that want one thing, and would be most of the traffic.
        Kind::Keryx { id } => vec![format!("apollos/keryx/{id}/+/+/meta")],
    };
    for f in filters {
        if let Err(e) = client.subscribe(&f, QoS::AtMostOnce).await {
            warn!("subscribe {f}: {e}");
        }
    }
}

/// Ask for a fresh copy. An Ankyra only has whole boards to give, so a
/// query's refresh there is the board's.
async fn request(client: &AsyncClient, kind: &Kind, query_id: Option<u32>) {
    let (topic, payload) = match kind {
        Kind::Ankyra { topic } => (format!("{topic}.publish.board"), json!({})),
        Kind::Keryx { id } => (
            format!("apollos/keryx/{id}/control"),
            match query_id {
                Some(q) => json!({"action": "refresh", "query_id": q}),
                None => json!({"action": "refresh"}),
            },
        ),
    };
    debug!("requesting on {topic}");
    if let Err(e) = client
        .publish(&topic, QoS::AtMostOnce, false, payload.to_string())
        .await
    {
        warn!("request on {topic}: {e}");
    }
}

fn interpret(source: &Source, label: &str, topic: &str, payload: &[u8]) -> Option<Event> {
    let topic_n = wire::normalise(topic);
    match &source.kind {
        Kind::Ankyra { topic: ours } => {
            let ours = wire::normalise(ours);
            if topic_n == format!("{ours}.plus") {
                match wire::plus(payload) {
                    Ok(entries) => Some(Event::Plus { entries }),
                    Err(e) => {
                        warn!("{label}: unreadable plus board: {e}");
                        None
                    }
                }
            } else if topic_n == ours {
                match wire::board(payload) {
                    Ok(readings) => Some(Event::Readings {
                        source: label.into(),
                        readings,
                        whole: true,
                    }),
                    Err(e) => {
                        warn!("{label}: unreadable board: {e}");
                        None
                    }
                }
            } else {
                None
            }
        }
        Kind::Keryx { id } => wire::keryx_meta(*id, topic, payload).map(|r| Event::Readings {
            source: label.into(),
            readings: vec![r],
            whole: false,
        }),
    }
}