use super::listen_config::ListenConfig;
use serde_json::{Value, json};
pub(super) struct Batch {
pub dropped: u64,
pub messages: Vec<Message>,
}
pub(super) struct Message {
pub id: String,
pub subject: String,
pub from: String,
pub at_ms: u64,
pub size: u64,
pub delivered: bool,
pub payload: Option<String>,
pub declared_type: String,
}
impl Batch {
pub fn read(
bytes: &[u8],
subject: &str,
config: &ListenConfig,
record_bytes: usize,
) -> Result<Self, String> {
if bytes.len() > record_bytes {
return Err(format!("a tap record exceeds record_bytes {record_bytes}"));
}
let value: Value =
serde_json::from_slice(bytes).map_err(|_| "the channel tap record is not JSON")?;
if value["type"] != "tap.batch" || value["target"] != subject {
return Err(format!("the tap record is not a tap.batch for {subject}"));
}
let dropped = value["dropped"]
.as_u64()
.ok_or("the tap dropped count is invalid")?;
let records = value["messages"]
.as_array()
.ok_or("the tap messages are not an array")?;
if records.len() > config.batch_messages {
return Err(format!(
"the tap batch exceeds batch_messages {}",
config.batch_messages
));
}
let messages = records
.iter()
.map(|record| Message::read(record, subject))
.collect::<Result<_, _>>()?;
Ok(Self { dropped, messages })
}
}
fn string<'a>(record: &'a Value, name: &str) -> Result<&'a str, String> {
record[name]
.as_str()
.filter(|value| !value.is_empty())
.ok_or_else(|| format!("the tap {name} is invalid"))
}
impl Message {
fn read(record: &Value, subject: &str) -> Result<Self, String> {
if string(record, "subject")? != subject {
return Err("the tap message names a different subject".to_owned());
}
let source = string(record, "from")?;
let station = match record.get("from_station") {
None | Some(Value::Null) => None,
Some(Value::String(name)) if !name.is_empty() => Some(name.as_str()),
Some(_) => return Err("the tap from_station is invalid".to_owned()),
};
let from = super::notifications::sender(source, station);
let payload = match record.get("payload") {
None | Some(Value::Null) => None,
Some(Value::String(text)) => Some(text.clone()),
Some(_) => return Err("the tap payload is not opaque UTF-8 text".to_owned()),
};
let declared_type = match record.get("content_type") {
None => "unknown".to_owned(),
Some(Value::String(kind)) if !kind.is_empty() => kind.clone(),
Some(_) => return Err("the tap declared type is invalid".to_owned()),
};
let outcome = string(record, "outcome")?;
if !matches!(
outcome,
"answered"
| "unanswered"
| "delivered"
| "reply"
| "nobody"
| "died"
| "deadline"
| "deferred"
| "refused"
| "dropped"
| "redelivered"
| "unresolved"
| "unaccepted"
) {
return Err("the tap outcome is invalid".to_owned());
}
Ok(Self {
id: string(record, "id")?.to_owned(),
subject: subject.to_owned(),
from,
at_ms: record["t"].as_u64().ok_or("the tap time is invalid")?,
size: record["size"].as_u64().ok_or("the tap size is invalid")?,
delivered: matches!(outcome, "delivered" | "reply" | "redelivered"),
payload,
declared_type,
})
}
pub fn notification(&self, device: &str, content: &str) -> Value {
json!({"jsonrpc":"2.0","method":"notifications/claude/channel","params":{"content":content,
"meta":{"subject":self.subject,"from":self.from,"at_ms":self.at_ms.to_string(),"id":self.id,"device":device}}})
}
}
pub(super) fn outcome(subject: &str, outcome: &str, content: &str) -> Value {
json!({"jsonrpc":"2.0","method":"notifications/claude/channel","params":{"content":content,"meta":{"subject":subject,"outcome":outcome}}})
}
pub(super) fn body(bytes: &[u8], message: &Message, bound: usize) -> String {
if bytes.len() <= bound
&& let Ok(text) = std::str::from_utf8(bytes)
{
return text.to_owned();
}
format!(
"{} bytes of {} on {}",
bytes.len(),
message.declared_type,
message.subject
)
}