liminal-cli 0.2.0

//! Tap wire records mapped to human-readable logging notifications.

use super::framing::INVALID_PARAMS;
use super::negotiate::Refusal;
use super::subscriptions::Service;
use serde::Deserialize;
use serde_json::{Map, Value, json};
use std::sync::atomic::{AtomicBool, Ordering};

pub(super) fn set_level(
    params: &Map<String, Value>,
    warning_only: &AtomicBool,
) -> Result<(), Refusal> {
    let level = match params.get("level").and_then(Value::as_str) {
        Some("debug" | "info" | "notice") => false,
        Some("warning" | "error" | "critical" | "alert" | "emergency") => true,
        _ => {
            return Err(Refusal {
                code: INVALID_PARAMS,
                message: "logging/setLevel needs a standard logging level".to_owned(),
                data: None,
            });
        }
    };
    warning_only.store(level, Ordering::SeqCst);
    Ok(())
}

#[derive(Deserialize)]
pub(super) struct Batch {
    #[serde(rename = "type")]
    kind: String,
    target: String,
    pub dropped: u64,
    pub messages: Vec<Message>,
}

#[derive(Deserialize)]
pub(super) struct Message {
    id: String,
    subject: String,
    size: u64,
    outcome: String,
    t: u64,
    from: String,
    from_station: Option<String>,
}

impl Batch {
    pub fn read(record: Value, subject: &str) -> Result<Self, String> {
        let batch: Self = serde_json::from_value(record)
            .map_err(|error| format!("the tap record could not be read: {error}"))?;
        if batch.kind != "tap.batch" || batch.target != subject {
            return Err(format!("the tap record is not a tap.batch for {subject}"));
        }
        for message in &batch.messages {
            if message.subject != subject
                || message.id.is_empty()
                || message.from.is_empty()
                || message.from_station.as_ref().is_some_and(String::is_empty)
                || !matches!(
                    message.outcome.as_str(),
                    "answered"
                        | "unanswered"
                        | "delivered"
                        | "reply"
                        | "nobody"
                        | "died"
                        | "deadline"
                        | "deferred"
                        | "refused"
                        | "dropped"
                        | "redelivered"
                        | "unresolved"
                        | "unaccepted"
                )
            {
                return Err(format!(
                    "the tap record carries invalid message fields for {subject}"
                ));
            }
        }
        Ok(batch)
    }
}

fn line(service: &Service, level: &str, data: &Value) -> Value {
    json!({"jsonrpc":"2.0","method":"notifications/message","params":{"level":level,"logger":format!("liminal.{}",service.name),"data":data}})
}

pub(super) fn message(service: &Service, message: &Message) -> Value {
    let from = message.from_station.clone().unwrap_or_else(|| {
        if message.from.bytes().all(|byte| byte.is_ascii_digit()) {
            format!("connection {}", message.from)
        } else {
            message.from.clone()
        }
    });
    line(
        service,
        "info",
        &json!({"device":service.device,"service":service.name,"subject":message.subject,"from":from,"outcome":message.outcome,"at_ms":message.t,"tap":message.id,"words":format!("{}: {}, {} B from {from}",message.subject,message.outcome,message.size).replace(['\r', '\n'], " ")}),
    )
}

pub(super) fn warning(service: &Service, subject: &str, outcome: &str, words: &str) -> Value {
    line(
        service,
        "warning",
        &json!({"device":service.device,"service":service.name,"subject":subject,"from":null,"outcome":outcome,"at_ms":null,"tap":null,"words":words.replace(['\r', '\n'], " ")}),
    )
}