haematite-cli 0.21.3

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 failure: Option<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 &current {
            if inspected.contains(id) || resolved.contains(id) {
                continue;
            }
            let answer = match issuer.draft_answer(id, expected).await {
                Ok(answer) => answer,
                Err(error) => Some(Answer {
                    draft: id.clone(),
                    decision: Decision::Cancelled,
                    person: String::new(),
                    failure: Some(error.to_string()),
                }),
            };
            if let Some(answer) = answer {
                resolved.insert(id.clone());
                events
                    .send(Ok(answer))
                    .await
                    .map_err(|_| anyhow::anyhow!("permission_watch_receiver_closed"))?;
            }
            inspected.insert(id.clone());
        }
        let deadline = issuer.watch_deadline().await;
        tokio::select! {
            changed=requests.changed()=>changed.map_err(|_|anyhow::anyhow!("permission_watch_requests_closed"))?,
            ()=tokio::time::sleep_until(deadline)=>{issuer.watch_credentials().await?;},
            next=issuer.generation(Some(&generation)), if !current.is_empty()=>{
                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_credentials().await?);
        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_credentials().await?)
            .send()
            .await
            .map_err(|_| anyhow::anyhow!("permission_draft_unreachable"))?;
        let value = self.json(response).await?;
        super::approval_decode::draft(&value, id, expected, &self.agent, &self.store)
    }
}

impl Drop for ApprovalWatch {
    fn drop(&mut self) {
        self.task.abort();
    }
}

#[cfg(test)]
#[path = "approval_watch_tests.rs"]
mod tests;