johnandersen777.bsky.social/knotless

Switch repository

Loading…

//! What the pages follow as it happens, as server-sent events: a
//! repository's records as the index takes them in, so its pages draw
//! again, and a running check's output as its runner sends it
//! ([`crate::live`]). Each is followed only by whoever may read it, as its
//! page is, and for ten minutes at most: the browser then asks again, and is
//! proved again.

use super::{CheckParams, Result, indexed, named_check, not_found, repo};
use crate::App;
use crate::live::Run;
use axum::extract::{Query, State};
use axum::http::HeaderMap;
use axum::response::sse::{Event, KeepAlive, Sse};
use axum::response::{IntoResponse, Response};
use futures::{Stream, StreamExt, stream};
use serde::Deserialize;
use serde_json::Value;
use std::convert::Infallible;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::broadcast::Receiver;

/// How long one following lasts.
const FOLLOW: Duration = Duration::from_secs(600);

/// What `changed` and `ended` carry, which says nothing: a browser drops an
/// event with no data, unheard, and axum sends an empty one with none.
const NOTHING: &str = "null";

/// How long after a write others are waited for, so that a burst of them,
/// as a push or a run makes, draws a page again once.
const SETTLE: Duration = Duration::from_millis(250);

#[derive(Deserialize)]
pub struct RepoParams {
    repo: String,
}

/// `/_live/repo`: `changed`, each time the index takes in a record about
/// the repository `repo`: a push, a pull, a comment, a check.
pub async fn follow_repo(
    State(app): State<Arc<App>>,
    Query(p): Query<RepoParams>,
) -> Result<Response> {
    let repo = repo::found(&app, &p.repo).await?;
    let none = not_found(anyhow::anyhow!("no index of {} here", repo.name));
    let index = indexed(&app, &repo).ok_or(none)?;
    let topic: Arc<str> = format!("{}/{}", repo.did, repo.key).into();
    let changes = index.changes.subscribe();
    let changed = stream::unfold(changes, move |mut changes| {
        let topic = topic.clone();
        async move {
            // Behind, it may have missed one of its own.
            while !changes.recv().await.ok().is_none_or(|t| t == topic) {}
            tokio::time::sleep(SETTLE).await;
            while changes.try_recv().is_ok() {}
            Some((Event::default().event("changed").data(NOTHING), changes))
        }
    });
    Ok(events(&app, changed))
}

/// `/_live/check`: a check's output as its runner sends it, from where
/// `Last-Event-ID` says the browser had it, as `output`, each a JSON string,
/// its id where it ends; its steps as they end, as `steps`; the step going,
/// as `going`, its name or null; and `ended`.
pub async fn follow_check(
    State(app): State<Arc<App>>,
    headers: HeaderMap,
    Query(p): Query<CheckParams>,
) -> Result<Response> {
    named_check(&app, &p).await?;
    let full = not_found(anyhow::anyhow!("too many runs followed"));
    let run = app.live.run(&p.uri).ok_or(full)?;
    let had = headers.get("last-event-id").and_then(|v| v.to_str().ok());
    let following = Following {
        news: run.news.subscribe(),
        run,
        had: had.and_then(|v| v.parse().ok()),
        steps: Value::Null,
        going: None,
        ended: false,
    };
    let said = stream::unfold(following, |mut f| async move {
        while !f.ended {
            let said = f.said();
            if !said.is_empty() {
                return Some((stream::iter(said), f));
            }
            let _ = f.news.recv().await;
        }
        None
    });
    Ok(events(&app, said.flatten()))
}

/// One page's following of a run.
struct Following {
    run: Arc<Run>,
    news: Receiver<()>,
    had: Option<u64>,
    steps: Value,
    going: Option<String>,
    ended: bool,
}

impl Following {
    /// What the run has that this following was not yet sent.
    fn said(&mut self) -> Vec<Event> {
        let kept = self.run.kept();
        let mut said = vec![];
        let (at, text) = kept.since(self.had);
        if !text.is_empty() {
            let end = at + text.len() as u64;
            let text = String::from_utf8_lossy(text);
            let output = Event::default().event("output").id(end.to_string());
            said.push(output.json_data(text).expect("a string"));
            self.had = Some(end);
        }
        if kept.steps != self.steps {
            self.steps = kept.steps.clone();
            let steps = Event::default().event("steps");
            said.push(steps.json_data(&self.steps).expect("JSON"));
        }
        if kept.going != self.going {
            self.going = kept.going.clone();
            let going = Event::default().event("going");
            said.push(going.json_data(&self.going).expect("JSON"));
        }
        if kept.ended {
            self.ended = true;
            said.push(Event::default().event("ended").data(NOTHING));
        }
        said
    }
}

/// `events` as server-sent events, for [`FOLLOW`] or until the server
/// stops.
fn events(app: &App, events: impl Stream<Item = Event> + Send + 'static) -> Response {
    let events = events.map(Ok::<_, Infallible>);
    let stopping = app.live.stopping.clone().cancelled_owned();
    let over = tokio::time::timeout(FOLLOW, stopping);
    let events = events.take_until(over);
    Sse::new(events)
        .keep_alive(KeepAlive::default())
        .into_response()
}