hyperhive/swarm-controller/src/request_reviewer.rs

//! Requesting a review for the notification router, which holds no forge
//! token.
//!
//! The router sends a [`RequestReviewer`] on
//! `swarm_wire::router_sink::REQUEST_REVIEWER_SUBJECT` and [`spawn`]'s
//! responder answers with an [`Outcome`] on the reply subject, having acted
//! under the controller's token. A reviewer whose review of the PR's head
//! already stands is skipped: requesting it again makes the forge dismiss it,
//! approval included. What stands is [`hive_types::review`], the rule
//! `hive-forge pr assign-reviewer` applies too.
//!
//! The request may be repeated (the router's event comes back after a failed
//! step): a reviewer with a request pending is a no-op for the forge, and one
//! who has reviewed since stands and is skipped.

use std::future::Future;
use std::sync::Arc;
use std::time::Duration;

use futures_util::StreamExt as _;
use hive_types::review::{ReviewInfo, standing};
use swarm_wire::router_sink::{Outcome, REQUEST_REVIEWER_SUBJECT, RequestReviewer, WORK_DEADLINE};

use crate::forge;
use crate::jobq::Late;
use crate::review_classifier::split_repo;
use crate::routes::AppState;

/// How long to wait before subscribing again.
const RETRY: Duration = Duration::from_secs(5);

/// The forge as the responder uses it: [`forge::Client`] in production, a fake
/// in the tests.
pub(crate) trait ReviewForge {
    fn reviews(
        &self,
        owner: &str,
        repo: &str,
        number: i64,
    ) -> impl Future<Output = anyhow::Result<Vec<ReviewInfo>>> + Send;

    fn request(
        &self,
        owner: &str,
        repo: &str,
        number: i64,
        reviewer: &str,
    ) -> impl Future<Output = anyhow::Result<()>> + Send;
}

impl ReviewForge for forge::Client {
    async fn reviews(
        &self,
        owner: &str,
        repo: &str,
        number: i64,
    ) -> anyhow::Result<Vec<ReviewInfo>> {
        self.pull_reviews(owner, repo, number).await
    }

    async fn request(
        &self,
        owner: &str,
        repo: &str,
        number: i64,
        reviewer: &str,
    ) -> anyhow::Result<()> {
        self.request_review(owner, repo, number, reviewer).await
    }
}

/// Answer the router's review requests for as long as the controller runs.
pub(crate) fn spawn(state: &AppState) {
    let Some(client) = state.status.as_ref().map(|s| s.queue_client()) else {
        tracing::warn!("no swarm queue is configured on this host; review requests go unanswered");
        return;
    };
    let forge = Arc::clone(&state.forge);
    tokio::spawn(async move { follow(&client, &forge).await });
}

/// Subscribe to the requests and answer each, and subscribe again if the
/// subscription ends.
async fn follow<F: ReviewForge>(client: &async_nats::Client, forge: &Late<Arc<F>>) {
    loop {
        match client.subscribe(REQUEST_REVIEWER_SUBJECT).await {
            Ok(mut requests) => {
                tracing::info!(
                    subject = REQUEST_REVIEWER_SUBJECT,
                    "request reviewer: answering requests"
                );
                while let Some(msg) = requests.next().await {
                    let Some(reply) = msg.reply.clone() else {
                        continue;
                    };
                    let outcome = answer(forge.get().map(Arc::as_ref), &msg.payload).await;
                    match &outcome {
                        Outcome::Error(why) => {
                            tracing::warn!(%why, "request reviewer: not requested");
                        }
                        other => {
                            tracing::info!(outcome = %other.encode(), "request reviewer: answered");
                        }
                    }
                    if let Err(e) = client.publish(reply, outcome.encode().into()).await {
                        tracing::warn!(error = %e, "request reviewer: answering failed");
                    }
                }
                tracing::warn!("request reviewer: subscription ended");
            }
            Err(e) => tracing::warn!(error = %e, "request reviewer: subscribe failed"),
        }
        tokio::time::sleep(RETRY).await;
    }
}

/// What to answer to one request `payload`, `forge` being `None` while the
/// controller has no forge access.
async fn answer<F: ReviewForge>(forge: Option<&F>, payload: &[u8]) -> Outcome {
    let request: RequestReviewer = match serde_json::from_slice(payload) {
        Ok(request) => request,
        Err(e) => return Outcome::Error(format!("the request does not decode: {e}")),
    };
    let Some(forge) = forge else {
        return Outcome::Error("this controller has no forge access yet".to_owned());
    };
    match tokio::time::timeout(WORK_DEADLINE, act(forge, &request)).await {
        Ok(outcome) => outcome,
        Err(_) => Outcome::Error(format!(
            "the forge did not answer within {}s",
            WORK_DEADLINE.as_secs()
        )),
    }
}

async fn act<F: ReviewForge>(forge: &F, request: &RequestReviewer) -> Outcome {
    let RequestReviewer {
        repo,
        number,
        reviewer,
    } = request;
    let (owner, name) = match split_repo(repo) {
        Ok(split) => split,
        Err(why) => return Outcome::Error(why),
    };
    let Ok(number_i64) = i64::try_from(*number) else {
        return Outcome::Error(format!("PR number {number} is out of range"));
    };
    if reviewer.trim().is_empty() {
        return Outcome::Error("no reviewer named".to_owned());
    }
    let reviews = match forge.reviews(owner, name, number_i64).await {
        Ok(reviews) => reviews,
        Err(e) => return Outcome::Error(format!("{e:#}")),
    };
    if let Some(review) = standing(&reviews, reviewer) {
        return Outcome::Skipped(format!(
            "{reviewer} already has a standing {} review on {repo}#{number}",
            review.state
        ));
    }
    match forge.request(owner, name, number_i64, reviewer).await {
        Ok(()) => Outcome::Requested,
        Err(e) => Outcome::Error(format!("{e:#}")),
    }
}

#[cfg(test)]
mod tests {
    use std::sync::Mutex;

    use super::*;

    #[derive(Default)]
    struct Fake {
        reviews: Vec<ReviewInfo>,
        reviews_fail: bool,
        request_fail: bool,
        asked: Mutex<Vec<(String, i64, String)>>,
    }

    impl ReviewForge for Fake {
        async fn reviews(&self, _: &str, _: &str, _: i64) -> anyhow::Result<Vec<ReviewInfo>> {
            if self.reviews_fail {
                anyhow::bail!("the forge is down");
            }
            Ok(self.reviews.clone())
        }

        async fn request(
            &self,
            owner: &str,
            repo: &str,
            number: i64,
            reviewer: &str,
        ) -> anyhow::Result<()> {
            if self.request_fail {
                anyhow::bail!("422: the reviewer is the PR's author");
            }
            self.asked.lock().unwrap().push((
                format!("{owner}/{repo}"),
                number,
                reviewer.to_owned(),
            ));
            Ok(())
        }
    }

    fn review(login: &str, verdict: &str, stale: bool, dismissed: bool) -> ReviewInfo {
        ReviewInfo {
            login: login.to_owned(),
            state: verdict.to_owned(),
            stale,
            dismissed,
        }
    }

    fn payload(repo: &str, number: u64, reviewer: &str) -> Vec<u8> {
        serde_json::to_vec(&RequestReviewer {
            repo: repo.to_owned(),
            number,
            reviewer: reviewer.to_owned(),
        })
        .unwrap()
    }

    fn asked(fake: &Fake) -> Vec<(String, i64, String)> {
        fake.asked.lock().unwrap().clone()
    }

    #[tokio::test]
    async fn a_reviewer_with_no_review_is_requested() {
        let fake = Fake::default();
        let got = answer(Some(&fake), &payload("o/r", 7, "review-classifier")).await;
        assert_eq!(got, Outcome::Requested);
        assert_eq!(
            asked(&fake),
            [("o/r".to_owned(), 7, "review-classifier".to_owned())]
        );
    }

    /// A review that still stands is skipped, because requesting from its
    /// reviewer would make the forge dismiss it.
    #[tokio::test]
    async fn a_standing_review_is_skipped_and_a_stale_or_dismissed_one_is_requested_again() {
        let reviewer = "review-classifier";
        let standing = Fake {
            reviews: vec![review(reviewer, "APPROVED", false, false)],
            ..Fake::default()
        };
        let got = answer(Some(&standing), &payload("o/r", 7, reviewer)).await;
        assert!(
            matches!(&got, Outcome::Skipped(why) if why.contains("APPROVED") && why.contains("o/r#7")),
            "{got:?}"
        );
        assert!(asked(&standing).is_empty(), "nothing was requested");

        for (stale, dismissed) in [(true, false), (false, true)] {
            let fake = Fake {
                reviews: vec![review(reviewer, "APPROVED", stale, dismissed)],
                ..Fake::default()
            };
            let got = answer(Some(&fake), &payload("o/r", 7, reviewer)).await;
            assert_eq!(
                got,
                Outcome::Requested,
                "stale {stale}, dismissed {dismissed}"
            );
            assert_eq!(asked(&fake).len(), 1);
        }

        // Another reviewer's standing review is no reason to skip this one.
        let other = Fake {
            reviews: vec![review("argus", "APPROVED", false, false)],
            ..Fake::default()
        };
        let got = answer(Some(&other), &payload("o/r", 7, reviewer)).await;
        assert_eq!(got, Outcome::Requested);
    }

    #[tokio::test]
    async fn a_forge_failure_is_an_error_and_a_failed_read_requests_nothing() {
        let down = Fake {
            reviews_fail: true,
            ..Fake::default()
        };
        let got = answer(Some(&down), &payload("o/r", 7, "review-classifier")).await;
        assert!(
            matches!(&got, Outcome::Error(e) if e.contains("the forge is down")),
            "{got:?}"
        );
        assert!(asked(&down).is_empty());

        let refused = Fake {
            request_fail: true,
            ..Fake::default()
        };
        let got = answer(Some(&refused), &payload("o/r", 7, "review-classifier")).await;
        assert!(
            matches!(&got, Outcome::Error(e) if e.contains("422")),
            "{got:?}"
        );
    }

    #[tokio::test]
    async fn a_request_that_is_malformed_or_has_no_forge_is_an_error() {
        let fake = Fake::default();
        for bad in [
            b"not json".to_vec(),
            payload("no-slash", 7, "review-classifier"),
            payload("o/r", 7, " "),
            payload("o/r", u64::MAX, "review-classifier"),
        ] {
            let got = answer(Some(&fake), &bad).await;
            assert!(
                matches!(got, Outcome::Error(_)),
                "{:?}",
                String::from_utf8_lossy(&bad)
            );
        }
        assert!(asked(&fake).is_empty());
        let none: Option<&Fake> = None;
        let got = answer(none, &payload("o/r", 7, "review-classifier")).await;
        assert!(
            matches!(&got, Outcome::Error(e) if e.contains("no forge")),
            "{got:?}"
        );
    }

    /// The built-in flow's reviewer is the account the controller makes.
    #[test]
    fn the_built_in_flow_requests_the_classifiers_account() {
        assert_eq!(
            swarm_wire::notify_flows::CLASSIFIER_ACCOUNT,
            forge::classifier_account::LOGIN
        );
    }

    /// Against a real server: a request sent the way the router sends it comes
    /// back with the outcome, and a subscriber before the controller is up
    /// leaves the sender with no responders.
    #[tokio::test]
    async fn a_request_over_the_queue_is_answered_with_its_outcome() {
        let nats = crate::test_nats::TestNats::start();
        let router = nats.client().await;
        let ask = |body: Vec<u8>| {
            let router = router.clone();
            async move {
                let sent = async_nats::Request::new()
                    .payload(body.into())
                    .timeout(Some(Duration::from_secs(5)));
                router.send_request(REQUEST_REVIEWER_SUBJECT, sent).await
            }
        };
        assert!(
            ask(payload("o/r", 7, "review-classifier")).await.is_err(),
            "no controller listening"
        );

        let fake = Arc::new(Fake {
            reviews: vec![review("argus", "APPROVED", false, false)],
            ..Fake::default()
        });
        let late: Late<Arc<Fake>> = Arc::default();
        late.set(Arc::clone(&fake)).ok().unwrap();
        let controller = nats.client().await;
        tokio::spawn(async move { follow(&controller, &late).await });
        // The subscription is processed once a request gets through.
        let mut reply = None;
        for _ in 0..50 {
            if let Ok(msg) = ask(payload("o/r", 7, "review-classifier")).await {
                reply = Some(msg);
                break;
            }
            tokio::time::sleep(Duration::from_millis(100)).await;
        }
        let reply = reply.expect("the controller answers");
        assert_eq!(
            Outcome::decode(&String::from_utf8_lossy(&reply.payload)),
            Outcome::Requested
        );
        let skipped = ask(payload("o/r", 7, "argus")).await.unwrap();
        assert!(matches!(
            Outcome::decode(&String::from_utf8_lossy(&skipped.payload)),
            Outcome::Skipped(_)
        ));
    }
}