use std::collections::{BTreeMap, BTreeSet};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use super::framing::INVALID_PARAMS;
use super::negotiate::Refusal;
use super::tools::ToolResult;
use crate::network::{Rail, TapCloser, TapReader};
use serde_json::{Map, Value, json};
pub(super) fn tool_entries() -> Vec<Value> {
["subscribe", "unsubscribe"].into_iter().map(|name| json!({
"name":name,
"description": if name == "subscribe" {"Receive a service's tapped messages as logging notifications, on the seat's grant."} else {"Close this ticket's subscriptions to a service."},
"inputSchema":{"type":"object","required":["service"],"properties":{"service":{"type":"string", "minLength":1}},"additionalProperties":false}
})).collect()
}
pub(super) fn service_argument(arguments: &Map<String, Value>) -> Result<&str, Refusal> {
arguments
.get("service")
.and_then(Value::as_str)
.filter(|service| !service.is_empty() && arguments.len() == 1)
.ok_or_else(|| Refusal {
code: INVALID_PARAMS,
message: "the tool takes only a nonempty service string".to_owned(),
data: None,
})
}
#[derive(Debug)]
pub(super) struct Service {
pub name: String,
pub device: String,
pub subjects: Vec<String>,
station: String,
pub active: AtomicBool,
closers: Vec<Arc<dyn TapCloser>>,
}
impl Service {
fn answer(&self) -> ToolResult {
ToolResult::ok(
json!({"service":self.name, "station":self.station, "subjects":self.subjects}),
)
}
pub fn close(&self) -> Result<(), String> {
self.active.store(false, Ordering::SeqCst);
close_all(&self.closers)
}
}
fn close_all(closers: &[Arc<dyn TapCloser>]) -> Result<(), String> {
let errors: Vec<_> = closers
.iter()
.filter_map(|closer| closer.close().err().map(|error| error.to_string()))
.collect();
if errors.is_empty() {
Ok(())
} else {
Err(errors.join("; "))
}
}
#[derive(Debug)]
pub(super) struct Pending {
pub reader: Box<dyn TapReader>,
pub service: Arc<Service>,
pub subject: String,
}
#[derive(Debug, Default)]
struct State {
services: BTreeMap<String, Arc<Service>>,
pending: Vec<Pending>,
}
#[derive(Debug, Default)]
pub(super) struct Subscriptions(Mutex<State>);
fn failed(service: &str, words: &str) -> ToolResult {
ToolResult::failed(words, json!({"service":service}))
}
impl Subscriptions {
pub fn tool<R: Rail>(&self, rail: &R, tool: &str, name: &str) -> ToolResult {
let Ok(mut state) = self.0.lock() else {
return failed(name, "the ticket's subscription lock is poisoned");
};
if tool == "unsubscribe" {
return unsubscribe(&mut state, name);
}
if let Some(service) = state
.services
.get(name)
.filter(|service| service.active.load(Ordering::SeqCst))
{
return service.answer();
}
state
.services
.retain(|_, service| service.active.load(Ordering::SeqCst));
if state.services.len() >= 16 {
return failed(name, "the ticket already subscribes to 16 services");
}
let record = match resolve(rail, name) {
Ok(record) => record,
Err(words) => return failed(name, &words),
};
let mut streams = Vec::new();
for subject in &record.subjects {
match rail.tap(&format!("console.tap.{subject}")) {
Ok(stream) => streams.push(stream),
Err(error) => {
let closers: Vec<_> = streams
.iter()
.map(|stream| Arc::clone(&stream.closer))
.collect();
let rollback = close_all(&closers).err();
let mut data = json!({"service":name,"subject":subject});
if let Some(words) = rollback {
data["close_error"] = json!(words);
}
return ToolResult::failed(&error.to_string(), data);
}
}
}
let service = Arc::new(Service {
name: name.to_owned(),
device: rail.server_address().to_owned(),
station: record.station,
subjects: record.subjects,
active: AtomicBool::new(true),
closers: streams
.iter()
.map(|stream| Arc::clone(&stream.closer))
.collect(),
});
for (stream, subject) in streams.into_iter().zip(&service.subjects) {
state.pending.push(Pending {
reader: stream.reader,
service: Arc::clone(&service),
subject: subject.clone(),
});
}
state.services.insert(name.to_owned(), Arc::clone(&service));
service.answer()
}
pub fn pending(&self) -> Result<Vec<Pending>, String> {
let mut state = self
.0
.lock()
.map_err(|_| "the ticket's subscription lock is poisoned")?;
Ok(std::mem::take(&mut state.pending))
}
pub fn close_all(&self) -> Result<(), String> {
let mut state = self
.0
.lock()
.map_err(|_| "the ticket's subscription lock is poisoned")?;
let services = std::mem::take(&mut state.services);
let errors: Vec<_> = services
.values()
.filter_map(|service| service.close().err())
.collect();
state.pending.clear();
drop(state);
if errors.is_empty() {
Ok(())
} else {
Err(errors.join("; "))
}
}
}
impl Drop for Subscriptions {
fn drop(&mut self) {
if let Err(words) = self.close_all() {
eprintln!("liminal subscription teardown failed: {words}");
}
}
}
fn unsubscribe(state: &mut State, name: &str) -> ToolResult {
let Some(service) = state.services.remove(name) else {
return failed(name, &format!("no subscription to service '{name}'"));
};
let was_active = service.active.load(Ordering::SeqCst);
let closed = service.close();
state
.pending
.retain(|pending| !Arc::ptr_eq(&pending.service, &service));
if let Err(words) = closed {
return failed(name, &words);
}
if !was_active {
return failed(name, &format!("no subscription to service '{name}'"));
}
ToolResult::ok(json!({"service":name,"subjects":service.subjects}))
}
struct Resolved {
station: String,
subjects: Vec<String>,
}
fn resolve<R: Rail>(rail: &R, name: &str) -> Result<Resolved, String> {
let listing = crate::network::lifecycle::tool(rail, "services", &Map::new())
.ok_or("the services tool is unavailable")?;
if listing.is_error {
return Err(listing.structured["error"]
.as_str()
.ok_or("the services refusal has no words")?
.to_owned());
}
let services = listing.structured["services"]
.as_array()
.ok_or("the services listing has no services array")?;
let mut matching = services.iter().filter(|record| record["name"] == name);
let record = matching
.next()
.ok_or_else(|| format!("no service named '{name}'"))?;
if matching.next().is_some() {
return Err(format!("more than one service named '{name}'"));
}
let station = match record.get("station") {
Some(station) => station
.as_str()
.filter(|station| !station.is_empty())
.ok_or("the service's station is not a nonempty string")?,
None => record["id"]
.as_str()
.filter(|id| !id.is_empty())
.ok_or("the service has no id for its station")?,
};
let subjects = record["subjects"]
.as_array()
.filter(|subjects| !subjects.is_empty())
.ok_or("the service has no nonempty subjects array")?;
let subjects: Vec<String> = subjects
.iter()
.map(|subject| {
subject
.as_str()
.filter(|subject| !subject.is_empty())
.map(str::to_owned)
.ok_or_else(|| "the service's subjects contain an invalid subject".to_owned())
})
.collect::<Result<_, _>>()?;
if subjects.iter().collect::<BTreeSet<_>>().len() != subjects.len() {
return Err("the service's subjects contain a duplicate".to_owned());
}
Ok(Resolved {
station: station.to_owned(),
subjects,
})
}