79 lines
3.1 KiB
Rust
79 lines
3.1 KiB
Rust
//! Reads an agent's icon out of the swarm's agent-icon KV bucket.
|
|
//!
|
|
//! Sibling of [`crate::agent_status`] and built the same way — the bucket
|
|
//! is resolved on first use and cached, a resolution failure is not — but
|
|
//! a much smaller read: one key, no roster to be complete against and no
|
|
//! freshness to derive. An icon is not a report about the agent, it is a
|
|
//! property of it, so there is nothing for a timestamp to mean here.
|
|
//!
|
|
//! **No hive is named on this path**, because the key does not carry one —
|
|
//! see `swarm_wire::agent_icon` for why, and for who writes it.
|
|
//!
|
|
//! This reader is also what creates the bucket: the agents that write it are
|
|
//! granted their own key and nothing that could create a stream.
|
|
|
|
use anyhow::{Context, Result};
|
|
|
|
/// Reads the agent-icon bucket. Holds a NATS client rather than a bucket
|
|
/// handle, so a controller that starts before the bucket exists picks it
|
|
/// up without a restart.
|
|
pub struct AgentIconReader {
|
|
client: async_nats::Client,
|
|
store: tokio::sync::OnceCell<async_nats::jetstream::kv::Store>,
|
|
}
|
|
|
|
impl AgentIconReader {
|
|
#[must_use]
|
|
pub fn new(client: async_nats::Client) -> Self {
|
|
Self {
|
|
client,
|
|
store: tokio::sync::OnceCell::new(),
|
|
}
|
|
}
|
|
|
|
async fn store(
|
|
&self,
|
|
) -> std::result::Result<&async_nats::jetstream::kv::Store, swarm_queue_client::Error> {
|
|
self.store
|
|
.get_or_try_init(|| swarm_queue_client::agent_icon::open_or_create(&self.client))
|
|
.await
|
|
}
|
|
|
|
/// Open the bucket, creating it, as soon as the queue is connected.
|
|
///
|
|
/// Agents write their keys without opening the bucket, so until this or
|
|
/// a first [`Self::get`] has run, every write is refused and retried.
|
|
pub async fn create_when_connected(&self) {
|
|
const POLL: std::time::Duration = std::time::Duration::from_secs(5);
|
|
loop {
|
|
if swarm_queue_client::ensure_connected(&self.client).is_ok() {
|
|
match self.store().await {
|
|
Ok(_) => return,
|
|
Err(e) => tracing::warn!(
|
|
error = swarm_queue_client::chain(&e),
|
|
"creating the agent-icon bucket failed; retrying"
|
|
),
|
|
}
|
|
}
|
|
tokio::time::sleep(POLL).await;
|
|
}
|
|
}
|
|
|
|
/// `agent`'s icon, or `None` when it has published none.
|
|
///
|
|
/// The bytes are returned unopened: this daemon does not parse, rewrite
|
|
/// or validate the SVG, and a consumer that renders it must treat it as
|
|
/// the untrusted document it is — see `get_agent_icon`'s response
|
|
/// headers.
|
|
pub async fn get(&self, agent: &str) -> Result<Option<Vec<u8>>> {
|
|
// An unconnected client does not fail a JetStream request, it hangs
|
|
// on it — see `swarm_queue_client::ensure_connected`.
|
|
swarm_queue_client::ensure_connected(&self.client)?;
|
|
let store = self.store().await?;
|
|
let icon = store
|
|
.get(agent)
|
|
.await
|
|
.with_context(|| format!("reading the agent-icon entry for {agent}"))?;
|
|
Ok(icon.map(Into::into))
|
|
}
|
|
}
|