thirdshift 0.14.5

Turn a GitHub issue into a ready-for-review pull request, unattended
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

//! Slow browsers only need the latest change notice; no rendering while idle.
use std::convert::Infallible;

use axum::Extension;
use axum::extract::State;
use axum::response::{
    IntoResponse, Response, Sse,
    sse::{Event, KeepAlive},
};
pub(super) async fn events(
    State(changes): State<super::changes::Receiver>,
    Extension(viewer): Extension<super::auth::Viewer>,
    Extension(notifier): Extension<super::changes::Notifier>,
) -> Response {
    let following = notifier.follow(changes.clone());
    // Every reconnect catches up even if its last notice was lost.
    let stream = futures_util::stream::unfold(
        (changes, viewer, None, following),
        |(mut changes, viewer, mut seen, following): (
            _,
            _,
            Option<super::changes::Revision>,
            _,
        )| async move {
            loop {
                if seen.is_some() && changes.changed().await.is_err() {
                    return None;
                }
                let current = *changes.borrow_and_update();
                if seen.is_none_or(|old| old.accounts != current.accounts) {
                    let signed_in = viewer.clone();
                    match tokio::task::spawn_blocking(move || signed_in.still_allowed()).await {
                        Ok(Ok(true)) => {}
                        Ok(Ok(false)) => return None,
                        error => {
                            eprintln!(
                                "thirdshift Console: can't authorize Inbox stream: {error:?}"
                            );
                            return None;
                        }
                    }
                }
                let commands_changed = seen.is_none_or(|old| old.commands != current.commands);
                seen = Some(current);
                if commands_changed {
                    let event = Event::default().event("commands").data("changed");
                    return Some((
                        Ok::<_, Infallible>(event),
                        (changes, viewer, seen, following),
                    ));
                }
            }
        },
    );
    (
        [("x-accel-buffering", "no"), ("cache-control", "no-store")],
        Sse::new(stream).keep_alive(KeepAlive::default()),
    )
        .into_response()
}