nordisk/monetize

crates/monetize-ui-web/src/feed.rs

//! The live feed, over server-sent events. An `EventSource` reconnects after a
//! network drop but NOT after an HTTP error status: when the session ends the
//! server answers 401 and the browser gives up, so a closed source sends the
//! operator to the login page.
//!
//! The lines come off monetize's `Events.Watch` stream (the console's server
//! holds one and fans it out), with a five-second poll as the server's
//! fallback. The console does not render them: every delta bumps `reread`,
//! and the page re-reads its outcomes. The step-by-step feed that used to sit
//! on the ledger page was the dump `docs/v2/07-console.md` rule 6 deletes.

use leptos::prelude::*;
use wasm_bindgen::prelude::*;
use wasm_bindgen::JsCast;

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Connection {
    Connecting,
    Live,
    Ended,
}

#[derive(Clone, Copy)]
pub struct Feed {
    pub connection: RwSignal<Connection>,
    /// Bumped on every delta, so a list re-reads instead of patching itself.
    /// The delta's content is not rendered anywhere (`docs/v2/07-console.md` rule 6: outcomes,
    /// not steps) — the re-read is the whole point of the feed.
    pub reread: RwSignal<u32>,
}

/// **The page's feed**: the one the estate shell opened ([`provide_feed`]),
/// else a stream of the caller's own.
///
/// ONE stream per page, not one per panel: a browser keeps six HTTP/1.1
/// connections per origin, and every `EventSource` holds one for as long as it
/// lives. An estate's Customers draws a panel set per product and Money four
/// panels that each re-read on the feed — one stream each was enough to use up
/// the six and leave the page's own reads queued behind them for good
/// (MEASURED 2026-10-02 in the picture walk: the Customers page of a
/// four-product estate never finished loading).
pub fn use_feed() -> Feed {
    use_context::<Feed>().unwrap_or_else(open_feed)
}

/// Open the one stream every panel under the caller shares.
pub fn provide_feed() -> Feed {
    let feed = open_feed();
    provide_context(feed);
    feed
}

/// Open a stream for as long as the calling component lives.
fn open_feed() -> Feed {
    let feed = Feed {
        connection: RwSignal::new(Connection::Connecting),
        reread: RwSignal::new(0),
    };
    // ONE feed: the ONE monetize's, whichever estate the shell shows.
    let Ok(source) = web_sys::EventSource::new("/api/events") else {
        feed.connection.set(Connection::Ended);
        return feed;
    };
    let on_open =
        Closure::<dyn FnMut(web_sys::Event)>::new(move |_| feed.connection.set(Connection::Live));
    let _ = source.add_event_listener_with_callback("open", on_open.as_ref().unchecked_ref());
    let on_message =
        Closure::<dyn FnMut(web_sys::MessageEvent)>::new(move |_: web_sys::MessageEvent| {
            feed.connection.set(Connection::Live);
            feed.reread.update(|n| *n = n.wrapping_add(1));
        });
    let _ = source.add_event_listener_with_callback("feed", on_message.as_ref().unchecked_ref());
    let on_error = Closure::<dyn FnMut(web_sys::Event)>::new({
        let source = source.clone();
        move |_| {
            if source.ready_state() == web_sys::EventSource::CLOSED {
                feed.connection.set(Connection::Ended);
                // A closed stream is not a lost session (a restart, a proxy): ask the server.
                // Only a console that says nobody is signed in leaves for the login page; a
                // signed-in operator stays on the page (the Sites order is never sent to enrol).
                leptos::task::spawn_local(async move {
                    if let Ok(s) = crate::api::get::<crate::app::Status>("/auth/status").await {
                        if crate::fmt::leaves_for_login(s.authenticated, s.open, s.passkey_enabled)
                        {
                            crate::app::go_to_login_because("session");
                        } else {
                            feed.reread.update(|n| *n = n.wrapping_add(1));
                        }
                    }
                });
            } else {
                feed.connection.set(Connection::Connecting);
            }
        }
    });
    let _ = source.add_event_listener_with_callback("error", on_error.as_ref().unchecked_ref());
    // Nothing here leaves the browser's single thread; the wrapper says so, and
    // owning the closures means a page opened repeatedly does not leak a set per visit.
    let owned = send_wrapper::SendWrapper::new((source, on_open, on_message, on_error));
    on_cleanup(move || {
        owned.0.close();
        drop(owned);
    });
    feed
}