use super::permission::{Decision, Issuer};
use std::collections::{BTreeMap, BTreeSet};
use tokio::sync::{mpsc, watch};
#[derive(Clone)]
pub(super) struct Expected {
pub digest: String,
}
pub(super) struct Answer {
pub draft: String,
pub decision: Decision,
pub person: String,
}
pub(super) struct ApprovalWatch {
requests: watch::Sender<BTreeMap<String, Expected>>,
pub ready: watch::Receiver<bool>,
pub events: mpsc::Receiver<anyhow::Result<Answer>>,
task: tokio::task::JoinHandle<anyhow::Result<()>>,
}
impl ApprovalWatch {
pub fn start(issuer: Issuer, capacity: usize) -> Self {
let (requests, received) = watch::channel(BTreeMap::new());
let (ready, readiness) = watch::channel(false);
let (events, queued) = mpsc::channel(capacity);
let task = tokio::spawn(async move {
let result = run(&issuer, received, ready, &events).await;
if let Err(error) = &result {
events
.send(Err(anyhow::anyhow!("{error}")))
.await
.map_err(|_| anyhow::anyhow!("permission_watch_receiver_closed"))?;
}
result
});
Self {
requests,
ready: readiness,
events: queued,
task,
}
}
pub fn replace(&self, requests: BTreeMap<String, Expected>) {
self.requests.send_replace(requests);
}
pub async fn close(&mut self) -> anyhow::Result<()> {
super::closing::stop(&mut self.task, "permission_watch")
.await?
.unwrap_or(Ok(()))
}
}
async fn run(
issuer: &Issuer,
mut requests: watch::Receiver<BTreeMap<String, Expected>>,
ready: watch::Sender<bool>,
events: &mpsc::Sender<anyhow::Result<Answer>>,
) -> anyhow::Result<()> {
let mut generation = issuer.generation(None).await?;
ready.send_replace(true);
let mut inspected = BTreeSet::new();
let mut resolved = BTreeSet::new();
loop {
let current = requests.borrow_and_update().clone();
inspected.retain(|id| current.contains_key(id));
resolved.retain(|id| current.contains_key(id));
for (id, expected) in ¤t {
if inspected.contains(id) || resolved.contains(id) {
continue;
}
if let Some(answer) = issuer.draft_answer(id, expected).await? {
resolved.insert(id.clone());
events
.send(Ok(answer))
.await
.map_err(|_| anyhow::anyhow!("permission_watch_receiver_closed"))?;
}
inspected.insert(id.clone());
}
if current.is_empty() {
requests
.changed()
.await
.map_err(|_| anyhow::anyhow!("permission_watch_requests_closed"))?;
continue;
}
tokio::select! {
changed=requests.changed()=>changed.map_err(|_|anyhow::anyhow!("permission_watch_requests_closed"))?,
next=issuer.generation(Some(&generation))=>{
let next=next?;
anyhow::ensure!(next!=generation,"permission_change_generation_unchanged");
generation=next;
inspected.clear();
}
}
}
}
impl Issuer {
async fn generation(&self, after: Option<&str>) -> anyhow::Result<String> {
let client = if after.is_some() {
&self.changes
} else {
&self.http
};
let mut request = client
.get(format!("{}/api/changes", self.origin))
.bearer_auth(self.watch_pass.as_str());
if let Some(after) = after {
request = request.query(&[("after", after)]);
}
let response = request
.send()
.await
.map_err(|_| anyhow::anyhow!("permission_changes_unreachable"))?;
let value = self.json(response).await?;
value["generation"]
.as_str()
.filter(|generation| !generation.is_empty() && generation.len() <= 128)
.map(str::to_owned)
.ok_or_else(|| anyhow::anyhow!("permission_change_generation_invalid"))
}
async fn draft_answer(&self, id: &str, expected: &Expected) -> anyhow::Result<Option<Answer>> {
let response = self
.http
.get(format!("{}/api/product-drafts/{id}", self.origin))
.bearer_auth(self.watch_pass.as_str())
.send()
.await
.map_err(|_| anyhow::anyhow!("permission_draft_unreachable"))?;
let value = self.json(response).await?;
anyhow::ensure!(
value["id"] == id
&& value["holder"] == self.agent
&& value["request_digest"] == expected.digest
&& value["target"]["kind"] == "haematite.database"
&& value["target"]["id"] == self.store
&& value["target"]["action"] == "agent_permission",
"permission_draft_binding_differs"
);
let decision = match value["state"].as_str() {
Some("held") => return Ok(None),
Some("approved") => Decision::Approve,
Some("refused") => Decision::Refuse,
_ => anyhow::bail!("permission_draft_state_invalid"),
};
let person = value["decider"]
.as_str()
.filter(|person| !person.is_empty() && person.len() <= 128 && *person != self.agent)
.ok_or_else(|| anyhow::anyhow!("permission_draft_decider_invalid"))?;
anyhow::ensure!(
value["decided_at"].as_u64().is_some_and(|time| time > 0),
"permission_draft_decision_time_invalid"
);
Ok(Some(Answer {
draft: id.to_owned(),
decision,
person: person.to_owned(),
}))
}
}
impl Drop for ApprovalWatch {
fn drop(&mut self) {
self.task.abort();
}
}