snyvi 1.42.0

//! The event stream (`/api/events`): one SSE connection per open page,
//! the mark that counts windows and agents, the resync when a stream falls
//! behind, and the focus beacon desktop notifications read.

use super::*;
use axum::body::Bytes;
use std::collections::HashMap;

/// Which pages have said they are in front, by the id each page picks for
/// itself, with the number of its last word. A page says so when it gains
/// focus or comes into view, says not when it loses it, goes out of view or
/// is left, and says it again when its stream comes back -- and nothing in
/// between: a window in front all afternoon is one request, where it was one
/// every three seconds (finding 13). By page, so two tabs cannot take each
/// other's word, and numbered, so a word that arrives late is not the last.
static FOCUSED: std::sync::LazyLock<std::sync::Mutex<HashMap<String, Said>>> =
    std::sync::LazyLock::new(Default::default);

/// A page's last word: its number, in front or not, seen at all or not, and
/// the desk it shows (its own, or the one a document is read over).
#[derive(Clone, Copy)]
struct Said {
    seq: u64,
    focused: bool,
    visible: bool,
    desk: Option<i64>,
}
/// More pages than anyone keeps open: past it the oldest words are dropped.
const FOCUS_PAGES: usize = 64;

/// What a page posts: in front or not, who it is, and which of its words
/// this is. No body is a page from before the words existed, which posted
/// only while in front, every three seconds.
#[derive(Deserialize, Default)]
pub(crate) struct FocusBody {
    focused: Option<bool>,
    /// Not hidden: in front or not, the page can be seen. A page from
    /// before 1.27 does not say, and counts as seen while in front.
    visible: Option<bool>,
    desk: Option<i64>,
    page: Option<String>,
    seq: Option<u64>,
}

/// Open tabs report focus so arrivals only raise a desktop notification when nobody is looking.
pub(crate) async fn focus(State(app): S, headers: HeaderMap, body: Bytes) -> Response {
    if let Some(no) = refuse_reader(&app, &headers) {
        return no;
    }
    let b: FocusBody = serde_json::from_slice(&body).unwrap_or_default();
    let focused = b.focused.unwrap_or(true);
    if let Some(page) = b.page.filter(|p| !p.is_empty() && p.len() <= 64) {
        let seq = b.seq.unwrap_or(0);
        let mut m = FOCUSED.lock().unwrap();
        if m.get(&page).is_none_or(|was| seq >= was.seq) {
            if m.len() >= FOCUS_PAGES && !m.contains_key(&page) {
                // Unseen words first: they say nothing a missing one does not.
                if let Some(k) = m.iter().find(|(_, v)| !v.visible).map(|(k, _)| k.clone()) {
                    m.remove(&k);
                } else {
                    m.clear();
                }
            }
            let visible = b.visible.unwrap_or(focused);
            m.insert(
                page,
                Said {
                    seq,
                    focused,
                    visible,
                    desk: b.desk.filter(|_| visible),
                },
            );
        }
    }
    if focused {
        *app.last_focus.lock().unwrap() = Instant::now();
    }
    StatusCode::NO_CONTENT.into_response()
}

/// Whether a page is in front now: one said so and pages are open still, or
/// one said so a moment ago (a page from before `focused` was a word).
pub(crate) fn focused(app: &App) -> bool {
    if app.last_focus.lock().unwrap().elapsed() < FOCUS_FOR {
        return true;
    }
    let mut m = FOCUSED.lock().unwrap();
    // No page open: whatever they said went with them.
    if app.pages.load(Ordering::Relaxed) == 0 {
        m.clear();
        return false;
    }
    m.values().any(|v| v.focused)
}

/// Whether a widget is in view: a global one (`None`) while any page can be
/// seen, a desk's while a page that can be seen shows that desk. What a
/// widget file's command runs on (`crate::widget::run`): nothing runs for a
/// sidebar nobody can see.
pub(crate) fn in_view(app: &App, desk: Option<i64>) -> bool {
    let mut m = FOCUSED.lock().unwrap();
    if app.pages.load(Ordering::Relaxed) == 0 {
        m.clear();
        return false;
    }
    m.values()
        .any(|v| v.visible && (desk.is_none() || v.desk == desk))
}

/// The desks some page in view shows, for the runner's round.
pub(crate) fn desks_in_view(app: &App) -> Vec<i64> {
    if app.pages.load(Ordering::Relaxed) == 0 {
        return Vec::new();
    }
    let m = FOCUSED.lock().unwrap();
    let mut out: Vec<i64> = m
        .values()
        .filter(|v| v.visible)
        .filter_map(|v| v.desk)
        .collect();
    out.sort_unstable();
    out.dedup();
    out
}

/// How long since a page was in front, for the updater's doors: none while
/// one is.
pub(crate) fn focus_age(app: &App) -> std::time::Duration {
    if focused(app) {
        std::time::Duration::ZERO
    } else {
        app.last_focus.lock().unwrap().elapsed()
    }
}

/// How long a page's last word that it was in front is good for on its own.
const FOCUS_FOR: std::time::Duration = std::time::Duration::from_secs(4);

pub(crate) fn notify_desktop(app: &App, doc: &Doc) {
    if std::env::var("SNYVI_NOTIFY")
        .map(|v| v == "0")
        .unwrap_or(false)
    {
        return;
    }
    if focused(app) {
        return;
    }
    // Clicking it opens the document where the reader reads: the window it
    // belongs to, raised, or a browser when there is no window. The daemon is
    // the one process that knows which, so it decides here rather than handing
    // a URL to the desktop and hoping.
    let url = format!("{}/d/{}", config::base_url(), doc.id);
    let has_window = app.has_window();
    crate::platform::notify_open(
        &doc.title,
        &format!("{} · {}", doc.project, doc.workflow_title),
        move || {
            if has_window && crate::desktop::hand_to_window(&url) {
                return;
            }
            platform::open_url(&url);
        },
    );
}

/// When an ask last raised a desktop notification: a burst of them is one.
static ASKED_AT: std::sync::atomic::AtomicI64 = std::sync::atomic::AtomicI64::new(0);
/// Asks this close after the one that raised a notification raise none.
const ASK_BURST_SECS: i64 = 30;

/// Whether something that waits on the reader raises a desktop notification:
/// not when they turned them off, not while a page is in front -- its corner
/// rises and chimes -- and once for a burst.
pub(crate) fn ask_notifies(off: bool, focused: bool, last: i64, now: i64) -> bool {
    !off && !focused && now - last >= ASK_BURST_SECS
}

/// What waits on the reader -- a turn an agent put on On you, an offer to a
/// friend -- told to the desktop when no snyvi page is in front: the chime
/// plays only in a page, so the one thing that blocks an agent was the one
/// thing the desktop never said (plan b0bca0a11f). A click opens its desk
/// with the corner's tray open (`#onyou`, `ui/corner.js`).
pub(crate) fn notify_waiting(app: &App, title: &str, body: &str, desk: Option<i64>) {
    let off = std::env::var("SNYVI_NOTIFY").is_ok_and(|v| v == "0");
    let now = crate::store::now();
    if !ask_notifies(off, focused(app), ASKED_AT.load(Ordering::Relaxed), now) {
        return;
    }
    ASKED_AT.store(now, Ordering::Relaxed);
    let url = match desk {
        Some(d) => format!("{}/desk/{d}#onyou", config::base_url()),
        None => format!("{}/#onyou", config::base_url()),
    };
    let has_window = app.has_window();
    crate::platform::notify_open(title, body, move || {
        if has_window && crate::desktop::hand_to_window(&url) {
            return;
        }
        platform::open_url(&url);
    });
}

/// A turn put on On you, said to the desktop: who, what they want, where.
/// Claude's own question dialog is its panel's, already on screen there.
pub(crate) fn notify_turn(app: &App, t: &crate::thread::Turn, more: usize) {
    if t.via == "dialog" {
        return;
    }
    let from = crate::receive::pane_origin(&app.store, Some(t.pane.as_str()));
    let who = if t.by.starts_with("claude") {
        "Claude".to_string()
    } else {
        from.as_ref()
            .map_or("An agent".into(), |o| format!("Panel {}", o.slot))
    };
    let does = if t.kind == "decide" {
        "asks"
    } else {
        "hands you"
    };
    let title = match more {
        0 => format!("{who} {does}"),
        n => format!("{who} {does} (+{n} more)"),
    };
    let body = match &from {
        Some(o) => format!("{} · {}", crate::thread::line(&t.text, 140), o.name),
        None => crate::thread::line(&t.text, 140),
    };
    notify_waiting(app, &title, &body, Some(t.desk_id));
}

#[derive(Deserialize)]
pub(crate) struct EventsQ {
    /// Set by a page that is inside the native window, so the daemon knows
    /// there is one to hand a link to. A string rather than a bool because a
    /// query string is not JSON: `?window=1` is what a page would naturally
    /// send, and it is not a bool to serde.
    ///
    /// Forgeable, and deliberately kept anyway: it is a count hint, not a
    /// credential. `EventSource` cannot set a header, so the capability cannot
    /// ride this stream and the count has nowhere else to live; what makes that
    /// safe is that nothing reachable from here grants anything. See
    /// `App::windows`.
    #[serde(default)]
    pub(crate) window: Option<String>,
    /// Set by the MCP server, with the name its client gave in `initialize`,
    /// so the daemon knows that agent is here for as long as the stream is.
    #[serde(default)]
    pub(crate) agent: Option<String>,
}

impl EventsQ {
    pub(crate) fn is_window(&self) -> bool {
        self.window
            .as_deref()
            .is_some_and(|v| !matches!(v, "" | "0" | "false" | "False"))
    }

    /// What the window says of itself: a window from 1.7 on stamps its own
    /// number on the mark, and the page passes it along; an older one says
    /// `1`, which the updater reads as older than any.
    pub(crate) fn window_version(&self) -> Option<String> {
        self.window.clone().filter(|_| self.is_window())
    }

    /// The agent's name, trimmed and cut to a length the header can hold.
    /// None when it is empty, which is a page and not an agent.
    pub(crate) fn agent(&self) -> Option<String> {
        let name = self.agent.as_deref()?.trim();
        if name.is_empty() {
            return None;
        }
        Some(name.chars().take(64).collect())
    }
}

/// Held by an event stream for as long as that stream lasts. A page that is
/// closed, reloaded or navigated away from takes its connection with it, and
/// an agent's process that ends takes its own; every count follows. Nothing
/// here times out, so a window that is quit is not a window a moment later,
/// a page that is gone is not a socket, and an agent whose session was
/// closed is not here.
pub(crate) struct StreamMark {
    pub(crate) app: Arc<App>,
    pub(crate) window: bool,
    pub(crate) agent: Option<String>,
}

impl StreamMark {
    pub(crate) fn new(
        app: Arc<App>,
        window: bool,
        agent: Option<String>,
        window_version: Option<String>,
    ) -> StreamMark {
        app.streams.fetch_add(1, Ordering::Relaxed);
        if agent.is_none() {
            app.pages.fetch_add(1, Ordering::Relaxed);
        }
        if window {
            app.windows.fetch_add(1, Ordering::Relaxed);
            // A window older than the update just applied wants: asked to
            // quit, and started again once its stream has ended, below.
            // Most releases never touch the window, and it is left alone.
            if let Some(u) = &app.update {
                if u.window_is_too_old(window_version.as_deref()) {
                    if let Some(bin) = u.app_path() {
                        eprintln!("snyvi: the window is {}, older than this release wants; relaunching it", window_version.as_deref().unwrap_or("?"));
                        app.relaunch_window.store(true, Ordering::Relaxed);
                        if let Err(e) = platform::spawn_detached(&bin, &["--quit"]) {
                            eprintln!("snyvi: could not ask the window to quit: {e}");
                            app.relaunch_window.store(false, Ordering::Relaxed);
                        }
                    }
                }
            }
        }
        if let Some(name) = &agent {
            let changed = {
                let mut m = app.online.lock().unwrap_or_else(|e| e.into_inner());
                *m.entry(name.clone()).or_insert(0) += 1;
                json!(*m)
            };
            emit(&app, "agents", json!({ "online": changed }));
        }
        StreamMark { app, window, agent }
    }
}

impl Drop for StreamMark {
    fn drop(&mut self) {
        self.app.streams.fetch_sub(1, Ordering::Relaxed);
        if self.agent.is_none() {
            self.app.pages.fetch_sub(1, Ordering::Relaxed);
        }
        if self.window {
            let left = self
                .app
                .windows
                .fetch_sub(1, Ordering::Relaxed)
                .saturating_sub(1);
            if left == 0 && self.app.relaunch_window.load(Ordering::Relaxed) {
                relaunch_window_when_gone(self.app.clone());
            }
        }
        // The last window closing is one of the updater's doors, and a page
        // going is worth a look at a pending restart.
        self.app.restart_wake.notify_one();
        if let Some(name) = &self.agent {
            let changed = {
                let mut m = self.app.online.lock().unwrap_or_else(|e| e.into_inner());
                if let Some(n) = m.get_mut(name) {
                    *n -= 1;
                    if *n == 0 {
                        m.remove(name);
                    }
                }
                json!(*m)
            };
            emit(&self.app, "agents", json!({ "online": changed }));
        }
    }
}

/// The window that was asked to quit has closed its stream: once no window
/// has come back for `WINDOW_GONE_FOR` -- a page that reloads itself closes
/// its stream too, and is back in a moment -- the new one is started. A
/// window that did come back is still the old one, and its stream ending
/// later is looked at again.
pub(crate) fn relaunch_window_when_gone(app: Arc<App>) {
    let Ok(rt) = tokio::runtime::Handle::try_current() else {
        return;
    };
    rt.spawn(async move {
        tokio::time::sleep(WINDOW_GONE_FOR).await;
        if app.windows.load(Ordering::Relaxed) != 0
            || !app.relaunch_window.swap(false, Ordering::Relaxed)
        {
            return;
        }
        if let Some(exe) = app.exe.as_ref().map(|e| e.path.clone()) {
            if let Err(e) = platform::spawn_detached(&exe, &["app"]) {
                eprintln!("snyvi: could not start the window again: {e}");
            }
        }
    });
}

pub(crate) const WINDOW_GONE_FOR: std::time::Duration = std::time::Duration::from_secs(2);

/// Broadcast payloads are "<event name>\n<json>".
///
/// The stream ends when the daemon is asked to stop. A graceful shutdown
/// closes the listener and then waits for every response in flight to
/// finish, and a stream that never ends is a response that never finishes:
/// the old daemon stayed up for as long as the window did, listening on
/// nothing, and the window stayed on it -- it heard no more arrivals, and the
/// daemon that took the port counted no window and handed every agent a link
/// to open in a browser instead. Seen on this machine: ten hours, one tab per
/// document. Ended here, the page reconnects to whatever is on the port now.
pub(crate) async fn events(
    State(app): S,
    Query(q): Query<EventsQ>,
) -> Sse<impl tokio_stream::Stream<Item = Result<Event, Infallible>>> {
    let rx = app.events.subscribe();
    let stop = BroadcastStream::new(app.shutdown.subscribe()).map(|_| None);
    let mark = StreamMark::new(app.clone(), q.is_window(), q.agent(), q.window_version());
    // The first thing on every stream is what the updater has to say, so
    // the pill is right at first paint without a field on every boot
    // payload or a poll.
    let first = tokio_stream::once(Some(Ok(Event::default()
        .event("update")
        .data(update_json(&app).to_string()))));
    let stream = first
        .chain(BroadcastStream::new(rx).filter_map(move |m| {
            // Captured so that the mark lives exactly as long as the stream does.
            let _keep = &mark;
            let ev = match m {
                Ok(msg) => {
                    let (name, data) = msg.split_once('\n').unwrap_or(("doc", msg.as_str()));
                    Event::default().event(name).data(data)
                }
                // The stream fell behind the channel and the events between
                // are gone. One `resync` in their place: the page refetches
                // the tree, the queue, the notes, the agents and the desks,
                // which is what each of them would have had it do.
                Err(_lagged) => Event::default().event("resync").data("{}"),
            };
            Some(Some(Ok(ev)))
        }))
        .merge(stop)
        .take_while(Option::is_some)
        .map(Option::unwrap);
    Sse::new(stream).keep_alive(KeepAlive::default())
}