neiam /apollos-dbus

//! What arrives, and what it means.
//!
//! Kept apart from the connection so it can be tested without one. Two
//! shapes reach here:
//!
//! * An Ankyra board: one JSON object per tick, keyed `"<type>-<id>"`,
//!   each value `{data, query: {name, meta}}` -- or, from a publisher that
//!   could not resolve the query, the bare data.
//! * A Keryx `.meta` message: one query's `{data, query}`, its type and id in
//!   the topic rather than a key.
//!
//! Both come to the same [`Reading`], so nothing downstream knows which.

use serde::{Deserialize, Serialize};
use serde_json::Value;

/// One query's latest, wherever it came from.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Reading {
    /// `gtfs-12`: the key the board uses, and what everything else keys on.
    pub key: String,
    pub kind: String,
    /// Everything after the type: a query's id, `12`, or for a Plani broken
    /// out a card per stop, the source and the stop, `4:place-sstat`.
    #[serde(default)]
    pub id: String,
    /// The id as a query id, where it is one; 0 where it is not.
    pub query_id: u32,
    pub name: Option<String>,
    pub meta: Value,
    pub data: Value,
}

impl Reading {
    fn new(kind: &str, id: &str, value: Value) -> Reading {
        let (data, name, meta) = unwrap(value);
        Reading {
            key: format!("{kind}-{id}"),
            kind: kind.to_string(),
            id: id.to_string(),
            query_id: id.parse().unwrap_or(0),
            name,
            meta,
            data,
        }
    }
}

/// `{data, query: {name, meta}}` apart, or the bare data as it is.
fn unwrap(value: Value) -> (Value, Option<String>, Value) {
    match value {
        Value::Object(mut map) if map.contains_key("data") && map.contains_key("query") => {
            let data = map.remove("data").unwrap_or(Value::Null);
            let query = map.remove("query").unwrap_or(Value::Null);
            let name = query
                .get("name")
                .and_then(Value::as_str)
                .map(str::to_string);
            let meta = query.get("meta").cloned().unwrap_or(Value::Null);
            (data, name, meta)
        }
        bare => (bare, None, Value::Null),
    }
}

/// `gtfs-12` into its type and id, on the first dash.
///
/// The id is not always a number: a Plani broken out into a card per stop
/// keys each as `gtfs-<source>:<stop>`, and a stop id is the feed's own --
/// `place-sstat` has a dash of its own. Types have none, so the first dash
/// is the one, as the Android client reads it too.
pub fn split_key(key: &str) -> Option<(&str, &str)> {
    let (kind, id) = key.split_once('-')?;
    (!kind.is_empty() && !id.is_empty()).then_some((kind, id))
}

/// A routing key as it reaches an MQTT client has its dots as slashes. Put
/// back, so a topic compares the same however the broker spelled it.
pub fn normalise(topic: &str) -> String {
    topic.replace('/', ".")
}

/// Every query on a board. Entries whose key is not `<type>-<id>` are
/// skipped rather than fatal: a newer publisher should not break this.
pub fn board(payload: &[u8]) -> Result<Vec<Reading>, serde_json::Error> {
    let value: Value = serde_json::from_slice(payload)?;
    let Value::Object(map) = value else {
        return Ok(vec![]);
    };
    Ok(map
        .into_iter()
        .filter_map(|(key, value)| {
            let (kind, id) = split_key(&key)?;
            Some(Reading::new(kind, id, value))
        })
        .collect())
}

/// The Plus board: the same queries, each the extended reading. Kept whole
/// -- what to make of it is a consumer's business.
///
/// A GTFS entry's Plus reading is a different shape and says so in its key,
/// `gtfs_plus-12`; it is filed under the board's `gtfs-12`, which is the
/// query it extends.
pub fn plus(payload: &[u8]) -> Result<Vec<(String, Value)>, serde_json::Error> {
    let value: Value = serde_json::from_slice(payload)?;
    let Value::Object(map) = value else {
        return Ok(vec![]);
    };
    Ok(map
        .into_iter()
        .filter_map(|(k, v)| {
            let (kind, id) = split_key(&k)?;
            let kind = if kind == "gtfs_plus" { "gtfs" } else { kind };
            Some((format!("{kind}-{id}"), v))
        })
        .collect())
}

/// A Keryx `.meta` message, if this topic is one: `apollos.keryx.<id>.<type>.<qid>.meta`.
pub fn keryx_meta(keryx_id: u32, topic: &str, payload: &[u8]) -> Option<Reading> {
    let topic = normalise(topic);
    let prefix = format!("apollos.keryx.{keryx_id}.");
    let rest = topic.strip_prefix(&prefix)?;
    let mut parts = rest.split('.');
    let (kind, qid, meta) = (parts.next()?, parts.next()?, parts.next()?);
    if meta != "meta" || parts.next().is_some() {
        return None;
    }
    let value: Value = serde_json::from_slice(payload).ok()?;
    qid.parse::<u32>().ok()?;
    Some(Reading::new(kind, qid, value))
}

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

    #[test]
    fn a_board_unwraps_each_query() {
        let payload = br##"{
          "gtfs-12": {"data": [{"route": "14"}], "query": {"name": "Forest Hills", "meta": {"x": 1}}},
          "weather-3": [{"name": "Home"}],
          "nonsense": {}
        }"##;
        let mut r = board(payload).unwrap();
        r.sort_by(|a, b| a.key.cmp(&b.key));

        assert_eq!(r.len(), 2, "a key that is not <type>-<id> is skipped");
        assert_eq!(r[0].key, "gtfs-12");
        assert_eq!(r[0].name.as_deref(), Some("Forest Hills"));
        assert_eq!(r[0].meta, json!({"x": 1}));
        assert_eq!(r[0].data, json!([{"route": "14"}]));
        assert_eq!(r[1].name, None, "the bare list has no name to give");
        assert_eq!(r[1].data, json!([{"name": "Home"}]));
    }

    #[test]
    fn keryx_meta_is_read_from_either_spelling_of_the_topic() {
        let payload = br#"{"data": [], "query": {"name": "Bikes", "meta": {}}}"#;
        let slashed = keryx_meta(7, "apollos/keryx/7/gbfs/4/meta", payload).unwrap();
        let dotted = keryx_meta(7, "apollos.keryx.7.gbfs.4.meta", payload).unwrap();
        assert_eq!(slashed, dotted);
        assert_eq!(slashed.key, "gbfs-4");
        assert_eq!(slashed.name.as_deref(), Some("Bikes"));
    }

    #[test]
    fn keryx_ignores_what_is_not_meta() {
        let p = b"[]";
        assert!(keryx_meta(7, "apollos/keryx/7/gbfs/4", p).is_none());
        assert!(keryx_meta(7, "apollos/keryx/7/gbfs/4/some-dock", p).is_none());
        assert!(keryx_meta(7, "apollos/keryx/7/control", p).is_none());
        assert!(keryx_meta(7, "apollos/keryx/71/gbfs/4/meta", p).is_none());
    }

    #[test]
    fn a_key_splits_on_its_first_dash() {
        assert_eq!(split_key("gtfs-12"), Some(("gtfs", "12")));
        assert_eq!(
            split_key("gtfs-4:place-sstat"),
            Some(("gtfs", "4:place-sstat"))
        );
        assert_eq!(split_key("gtfs"), None);
        assert_eq!(split_key("gtfs-"), None);
    }

    #[test]
    fn a_plani_broken_out_by_stop_is_read_whole() {
        // What a Plani with a card per stop publishes: source and stop in
        // the key, and the feed's stop id with a dash of its own.
        let payload = br##"{
          "gtfs-4:place-sstat": {"data": [], "query": {"name": "South Station", "meta": {}}},
          "gbfs-9:a3b1": {"data": [], "query": {"name": "Dock", "meta": {}}}
        }"##;
        let mut r = board(payload).unwrap();
        r.sort_by(|a, b| a.key.cmp(&b.key));

        assert_eq!(r.len(), 2);
        assert_eq!(r[1].key, "gtfs-4:place-sstat");
        assert_eq!(r[1].kind, "gtfs");
        assert_eq!(r[1].id, "4:place-sstat");
        assert_eq!(r[1].query_id, 0, "not a query id, so none");
        assert_eq!(r[1].name.as_deref(), Some("South Station"));
    }

    #[test]
    fn gtfs_plus_is_filed_under_the_query_it_extends() {
        let entries = plus(br#"{"gtfs_plus-12": [], "gbfs-3": []}"#).unwrap();
        let mut keys: Vec<&str> = entries.iter().map(|(k, _)| k.as_str()).collect();
        keys.sort();
        assert_eq!(keys, vec!["gbfs-3", "gtfs-12"]);
    }
}