//! The dashboard (`--features dashboard`): a page showing the node's figures
//! (`/admin/stats`), its peers (`/peers`) and a map of the swarm, kept up to date
//! through Server-Sent Events.
//!
//! One producer per node serves every client: while at least one is connected (or
//! asked for the state within `LINGER`), it builds the snapshot every `TICK` and, if
//! something changed, publishes it, and a JSON merge patch (RFC 7386) from the one
//! before to every client connected. Nothing changed, nothing is sent (but SSE's
//! keep-alive comments). Ages (`…SecsAgo`, `uptimeSecs` and the like, `GROWING`) grow by
//! themselves: they alone changing is no change, and the page counts them on.
//!
//! The map: the graph the member list makes (`topology::intended_graph`, as this node
//! sees it), and the links this node measures (`own_links`). Other members' links the
//! page reads from them itself, from the nodes it subscribes to
//! (`GET /admin/dashboard/links/own/events`, there).

mod links;

use std::collections::{BTreeMap, BTreeSet};
use std::convert::Infallible;
use std::sync::{Arc, Mutex, OnceLock};
use std::time::Duration;

use axum::extract::Query;
use axum::http::{HeaderValue, StatusCode, header::CONTENT_TYPE};
use axum::response::sse::{Event, KeepAlive, Sse};
use axum::response::{Html, IntoResponse, Response};
use axum::routing::{get, post};
use axum::{Extension, Router};
use serde_json::{Map, Value as JsonValue, json};
use tokio::sync::{Notify, broadcast, watch};
use tokio::time::Instant;

use crate::locks::MutexExt;
use crate::state::AppState;
use crate::swarm::{members, topology};
use crate::transport::Hex;
use links::Link;

/// How long messages logged are gathered before they are sent on, after the first: a
/// busy node sends about ten events a second at most, a quiet one each message soon
/// after it went.
const MESSAGES_GATHERED: Duration = Duration::from_millis(100);

/// How often the snapshot is built and compared, while anyone looks.
const TICK: Duration = Duration::from_secs(2);
/// How often the member list's graph is recomputed (it costs about `members²`).
const GRAPH_EVERY: Duration = Duration::from_secs(30);
/// The producer runs on this long after the last request for the state.
const LINGER: Duration = Duration::from_secs(60);

/// A snapshot or a patch as published: its number, and its JSON. A snapshot's text is
/// made when first asked for (a client connecting, or one that fell behind), not every
/// time the state changes: the whole state may be large, and mostly nobody new comes.
struct Published {
    seq: u64,
    value: Option<Arc<JsonValue>>,
    json: std::sync::OnceLock<String>,
}

impl Published {
    fn patch(seq: u64, patch: &JsonValue) -> Self {
        Self {
            seq,
            value: None,
            json: std::sync::OnceLock::from(patch.to_string()),
        }
    }

    fn snapshot(seq: u64, snapshot: Arc<JsonValue>) -> Self {
        Self {
            seq,
            value: Some(snapshot),
            json: std::sync::OnceLock::new(),
        }
    }

    fn json(&self) -> &str {
        self.json.get_or_init(|| {
            self.value
                .as_ref()
                .map(|value| value.to_string())
                .unwrap_or_default()
        })
    }
}

/// The producer and what it published, shared by all clients.
struct Hub {
    state: AppState,
    /// The latest snapshot; None while the producer hasn't made one (since it started).
    latest: watch::Sender<Option<Arc<Published>>>,
    /// Each patch, to the clients connected.
    patches: broadcast::Sender<Arc<Published>>,
    /// Where the document changed since the last tick, to the clients connected:
    /// `{"paths": [...]}`, or `{"all": true}` if too many or not known. Shared by every page
    /// open; each is told of the paths its user may read at least partly (`changed_for`).
    documents: broadcast::Sender<Arc<String>>,
    /// The messages exchanged, one by one (`traffic::MessageLog`), to the clients
    /// connected as they come (`forward_messages`): a JSON list.
    messages: broadcast::Sender<Arc<String>>,
    control: Mutex<Control>,
    /// Someone asked for it now (`refresh`): a tick at once, the graph recomputed.
    poke: Notify,
    /// True once the node stops: every client gets an `offline` event, and its stream
    /// ends (so the HTTP server needn't wait for clients to go).
    closing: watch::Receiver<bool>,
}

struct Control {
    running: bool,
    /// The producer stops once no client is connected and this is past.
    wanted_until: Instant,
}

impl Hub {
    fn new(state: AppState, closing: watch::Receiver<bool>) -> Arc<Self> {
        let (latest, _) = watch::channel(None);
        let (patches, _) = broadcast::channel(64);
        let (documents, _) = broadcast::channel(64);
        let (messages, _) = broadcast::channel(64);
        Arc::new(Self {
            state,
            latest,
            patches,
            documents,
            messages,
            poke: Notify::new(),
            control: Mutex::new(Control {
                running: false,
                wanted_until: Instant::now(),
            }),
            closing,
        })
    }

    /// Someone looks: the producer runs (started if it wasn't), for `LINGER` at least.
    fn want(self: &Arc<Self>) {
        let mut control = self.control.lock_or_recover();
        control.wanted_until = Instant::now() + LINGER;
        if !control.running && !*self.closing.borrow() {
            control.running = true;
            // What it published before it stopped is stale.
            self.latest.send_replace(None);
            tokio::spawn(Arc::clone(self).run());
        }
    }

    /// The latest snapshot, once there is one; None if the node stops first.
    async fn current(&self) -> Option<Arc<Published>> {
        let mut latest = self.latest.subscribe();
        let mut closing = self.closing.clone();
        // The borrows (lock guards, not Send) end within the arms.
        tokio::select! {
            published = latest.wait_for(Option::is_some) => {
                published.ok().and_then(|current| current.clone())
            }
            _ = closing.wait_for(|closing| *closing) => None,
        }
    }

    /// Resolves once the node stops (or its signal is gone).
    async fn closed(&self) {
        let mut closing = self.closing.clone();
        let _ = closing.wait_for(|closing| *closing).await;
    }

    async fn run(self: Arc<Self>) {
        // The latest snapshot, and it without its ages (`without_ages`): what the next
        // is told apart from.
        let mut last: Option<(Arc<JsonValue>, JsonValue)> = None;
        let mut seq = 0u64;
        let mut graph = Graph::default();
        let mut graph_due = Instant::now();
        let mut changes = self.state.change_events.subscribe();
        // The messages one by one, from now on, while it runs: sent on as they come.
        let log = &self.state.traffic.log;
        log.watch(true);
        let forwarding = tokio::spawn(Arc::clone(&self).forward_messages());
        let _forwarding = AbortOnDrop(forwarding);
        loop {
            if Instant::now() >= graph_due {
                graph.refresh(&self.state);
                graph_due = Instant::now() + GRAPH_EVERY;
            }
            let snapshot = build(&self.state, &graph).await;
            let stripped = without_ages(&snapshot);
            let changed = last.as_ref().is_none_or(|(_, old)| *old != stripped);
            if changed {
                seq += 1;
                if let Some((old, _)) = &last
                    && let Some(patch) = merge_patch(old, &snapshot)
                {
                    let _ = self.patches.send(Arc::new(Published::patch(seq, &patch)));
                }
                let snapshot = Arc::new(snapshot);
                self.latest.send_replace(Some(Arc::new(Published::snapshot(
                    seq,
                    Arc::clone(&snapshot),
                ))));
                last = Some((snapshot, stripped));
            }
            let poked = tokio::select! {
                () = tokio::time::sleep(TICK) => false,
                () = self.poke.notified() => true,
            };
            if poked {
                graph_due = Instant::now();
            }
            if let Some(changed) = document_changes(&mut changes) {
                let _ = self.documents.send(Arc::new(changed.to_string()));
            }
            let unwanted = || {
                self.patches.receiver_count() == 0
                    && Instant::now() >= self.control.lock_or_recover().wanted_until
            };
            if *self.closing.borrow() {
                log.watch(false);
                self.control.lock_or_recover().running = false;
                return;
            }
            if unwanted() {
                // Nobody looks: it stops. Unless someone came meanwhile, then it goes on.
                let mut control = self.control.lock_or_recover();
                if self.patches.receiver_count() == 0 && Instant::now() >= control.wanted_until {
                    log.watch(false);
                    control.running = false;
                    return;
                }
                drop(control);
                graph_due = Instant::now();
            }
        }
    }
}

impl Hub {
    /// The messages logged, to every client connected (`messages`), as they come:
    /// gathered for `MESSAGES_GATHERED` after the first, the latest `MESSAGES_AT_MOST`.
    /// For as long as the producer runs.
    async fn forward_messages(self: Arc<Self>) {
        let log = &self.state.traffic.log;
        let mut logged = log.latest();
        loop {
            log.logged_after(logged).await;
            tokio::time::sleep(MESSAGES_GATHERED).await;
            let mut messages = log.after(logged, MESSAGES_AT_MOST);
            let Some(last) = messages.last() else {
                continue;
            };
            logged = last.seq;
            by_node(&self.state, &mut messages).await;
            if let Ok(json) = serde_json::to_string(&messages) {
                let _ = self.messages.send(Arc::new(json));
            }
        }
    }
}

/// A task stopped when this is dropped.
struct AbortOnDrop(tokio::task::JoinHandle<()>);

impl Drop for AbortOnDrop {
    fn drop(&mut self) {
        self.0.abort();
    }
}

/// Paths changed in a tick listed at most; more: `all`.
/// The messages the paths layer logged under an address (its own PINGs and PONGs) under
/// the node there instead, where one is known: the address it proved itself at, or is
/// said to be at (`Transport::identities_of`), named in `address`.
async fn by_node(state: &AppState, messages: &mut [crate::transport::traffic::Logged]) {
    let peers = state.peers_snapshot().await;
    let known = |peer: &str| peers.iter().any(|(known, _)| known == peer);
    if messages.iter().all(|message| known(&message.peer)) {
        return;
    }
    let mut nodes: std::collections::HashMap<String, String> = std::collections::HashMap::new();
    for (peer, _) in &peers {
        for address in state.transport.identities_of(peer) {
            nodes.entry(address).or_insert_with(|| peer.clone());
        }
    }
    for message in messages {
        if let Some(node) = nodes.get(&message.peer) {
            message.address = Some(std::mem::replace(&mut message.peer, node.clone()));
        }
    }
}

const MAX_CHANGED: usize = 256;
/// Messages sent to the clients per tick at most, the latest (the console keeps fewer).
const MESSAGES_AT_MOST: usize = 500;

/// Where the document changed since last asked (`Hub::documents`), if anywhere.
fn document_changes(
    changes: &mut broadcast::Receiver<Arc<crate::document::store::ChangeSet>>,
) -> Option<JsonValue> {
    let mut paths = BTreeSet::new();
    let mut all = false;
    loop {
        match changes.try_recv() {
            Ok(set) => {
                paths.extend(set.paths.iter().cloned());
                all |= paths.len() > MAX_CHANGED;
            }
            Err(broadcast::error::TryRecvError::Lagged(_)) => all = true,
            Err(_) => break,
        }
    }
    if all {
        Some(json!({ "all": true }))
    } else if paths.is_empty() {
        None
    } else {
        Some(json!({ "paths": paths }))
    }
}

/// What a client's stream waits for.
enum Next {
    Patch(Result<Arc<Published>, broadcast::error::RecvError>),
    Document(Result<Arc<String>, broadcast::error::RecvError>),
    Messages(Result<Arc<String>, broadcast::error::RecvError>),
    Closing,
}

/// The intended graph, recomputed every `GRAPH_EVERY`.
#[derive(Default)]
struct Graph {
    intended: BTreeMap<String, BTreeSet<String>>,
}

impl Graph {
    fn refresh(&mut self, state: &AppState) {
        self.intended = topology::intended_graph(state).unwrap_or_default();
    }
}

/// The snapshot: the node, its figures, its peers, and the map.
async fn build(state: &AppState, graph: &Graph) -> JsonValue {
    // The peers once, for their list and the figures' totals over them.
    let list = super::peers::peer_list(state).await;
    let stats = serde_json::to_value(super::stats::collect(state, &list).await).unwrap_or_default();
    let peers = serde_json::to_value(&list).unwrap_or_default();
    let (aliases, dashboards, remakes) = {
        let replica = state.doc.read().await;
        // The remakes of our epoch, with who voted: the map shows them.
        let remakes = crate::remake::current(
            &state.remake,
            &replica.doc,
            replica.epoch,
            state.clock.now_millis(),
            state.timing.dormant_after.as_millis() as u64,
        );
        let dashboards: std::collections::HashMap<String, String> = members::members(&replica.doc)
            .into_iter()
            .filter_map(|pk| members::dashboard(&replica.doc, &pk).map(|url| (pk, url)))
            .collect();
        (members::aliases(&replica.doc), dashboards, remakes)
    };
    let local = state.local_id();
    let intended: Map<String, JsonValue> = graph
        .intended
        .iter()
        .map(|(member, neighbours)| (member.clone(), json!(neighbours)))
        .collect();
    let own = links::own_links(state).await;
    // With its epoch: the page shows who entered a new one.
    let mut measured = Map::new();
    measured.insert(
        local.to_owned(),
        json!({ "secsAgo": 0, "links": links_json(&own),
            "epoch": crate::document::replica::EpochRecord::from(state.local_epoch()) }),
    );
    // Everyone on the map: in the member list's graph, measured, or at the other end of
    // a link measured. A node that joined through this one is linked to it before its
    // registration reaches this node, and before the graph is computed again
    // (`GRAPH_EVERY`); it shows then, hollow (no links of its own reported).
    let linked: BTreeSet<&String> = own.iter().map(|link| &link.peer).collect();
    let mut members_json = Map::new();
    for member in graph.intended.keys().chain(measured.keys()).chain(linked) {
        members_json.insert(
            member.clone(),
            json!({ "alias": aliases.get(member), "self": member == local,
                "dashboard": dashboards.get(member) }),
        );
    }
    json!({
        "node": { "id": local, "alias": state.alias },
        "page": page().0,
        "stats": stats,
        "peers": peers,
        "graph": {
            "members": members_json,
            "intended": intended,
            "measured": measured,
        },
        "remakes": remakes,
    })
}

fn links_json(links: &[Link]) -> JsonValue {
    links
        .iter()
        .map(|link| {
            json!({
                "peer": link.peer,
                "transport": link.transport,
                "relayed": link.relayed,
                "down": link.down,
                "rttMs": link.rtt_us.map(|us| f64::from(us) / 1000.0),
                "dormant": link.dormant,
                "clock": link.clock,
                "others": link.others.iter().map(|other| json!({
                    "transport": other.transport,
                    "rttMs": other.rtt_us.map(|us| f64::from(us) / 1000.0),
                    "relayed": other.relayed,
                })).collect::<Vec<_>>(),
            })
        })
        .collect()
}

/// Fields that grow by themselves besides those named `…SecsAgo`; the page gets the
/// list when served (`GROWING_MARK`).
const GROWING: &[&str] = &[
    "uptimeSecs",
    "dormantForSecs",
    "divergedForSecs",
    "observedSecs",
    "lastFromNeighbourSecs",
    "oldestQuarantineSecs",
];

/// Whether the field `key` is an age, which grows by itself.
fn is_age(key: &str) -> bool {
    key == "secsAgo" || key.ends_with("SecsAgo") || GROWING.contains(&key)
}

/// `value` without its ages, to tell whether anything else changed.
fn without_ages(value: &JsonValue) -> JsonValue {
    match value {
        JsonValue::Object(object) => JsonValue::Object(
            object
                .iter()
                .filter(|(key, _)| !is_age(key))
                .map(|(key, value)| (key.clone(), without_ages(value)))
                .collect(),
        ),
        JsonValue::Array(items) => JsonValue::Array(items.iter().map(without_ages).collect()),
        other => other.clone(),
    }
}

/// The JSON merge patch (RFC 7386) from `old` to `new`, if they differ: objects field
/// by field (a field gone is null), anything else replaced whole. A null in `new` reads
/// as the field gone, as a merge patch can't tell them apart.
fn merge_patch(old: &JsonValue, new: &JsonValue) -> Option<JsonValue> {
    match (old, new) {
        (JsonValue::Object(old), JsonValue::Object(new)) => {
            let mut patch = Map::new();
            for key in old.keys() {
                if !new.contains_key(key) {
                    patch.insert(key.clone(), JsonValue::Null);
                }
            }
            for (key, value) in new {
                match old.get(key) {
                    Some(previous) => {
                        if let Some(change) = merge_patch(previous, value) {
                            patch.insert(key.clone(), change);
                        }
                    }
                    None => {
                        patch.insert(key.clone(), value.clone());
                    }
                }
            }
            (!patch.is_empty()).then_some(JsonValue::Object(patch))
        }
        (old, new) if old == new => None,
        (_, new) => Some(new.clone()),
    }
}

/// The dashboard's routes (authenticated, as the rest of `/admin`), with their hub.
pub(crate) fn router(state: AppState) -> Router<AppState> {
    // Its streams end when the node shuts down (`AppState::closing`).
    let closing = state.closing.subscribe();
    Router::new()
        .route("/admin/dashboard", get(page_html))
        .route("/admin/dashboard/state", get(current_state))
        .route("/admin/dashboard/events", get(events))
        .route("/admin/dashboard/refresh", post(refresh))
        .route("/admin/dashboard/tree", get(writer_tree))
        .route("/admin/dashboard/links/own", get(own_links_now))
        .route("/admin/dashboard/links/own/events", get(own_links_events))
        .route(
            "/admin/dashboard/messages/own/events",
            get(own_messages_events),
        )
        .layer(Extension(Hub::new(state, closing)))
}

/// How often a stream of this node's own links looks whether they changed
/// (`own_links_events`).
const OWN_LINKS_EVERY: Duration = Duration::from_secs(2);

/// This node's own links, as it measures them: its ID, its epoch, the links
/// (`links_json`), what a member's report on the map holds.
async fn own_report(state: &AppState) -> (Vec<Link>, JsonValue) {
    let own = links::own_links(state).await;
    let report = json!({
        "node": state.local_id(),
        "epoch": crate::document::replica::EpochRecord::from(state.local_epoch()),
        "links": links_json(&own),
    });
    (own, report)
}

/// `GET /admin/dashboard/links/own`: this node's own links (`own_report`), for a page
/// of this node or, through CORS, another member's (`api::cors`).
async fn own_links_now(Extension(hub): Extension<Arc<Hub>>) -> axum::Json<JsonValue> {
    axum::Json(own_report(&hub.state).await.1)
}

/// `GET /admin/dashboard/links/own/events`: server-sent events, `links` with this
/// node's own links at once, then whenever they changed (`links::changed`, looked at
/// every `OWN_LINKS_EVERY`); `offline` when the node stops, which ends it.
async fn own_links_events(Extension(hub): Extension<Arc<Hub>>) -> impl IntoResponse {
    let stream = async_stream::stream! {
        let (mut last, report) = own_report(&hub.state).await;
        yield Ok::<_, Infallible>(Event::default().event("links").data(report.to_string()));
        let mut every = tokio::time::interval(OWN_LINKS_EVERY);
        every.tick().await;
        loop {
            tokio::select! {
                _ = every.tick() => {}
                () = hub.closed() => {
                    yield Ok(offline_event());
                    break;
                }
            }
            let (own, report) = own_report(&hub.state).await;
            if links::changed(&last, &own) {
                last = own;
                yield Ok(Event::default().event("links").data(report.to_string()));
            }
        }
    };
    (
        [(super::UNBUFFERED.0, super::UNBUFFERED.1)],
        Sse::new(stream).keep_alive(KeepAlive::default()),
    )
}

/// Stops looking at the message log when dropped (`own_messages_events`).
struct Watching(Arc<Hub>);

impl Drop for Watching {
    fn drop(&mut self) {
        self.0.state.traffic.log.watch(false);
    }
}

/// `GET /admin/dashboard/messages/own/events`: server-sent events, `messages` with the
/// messages this node sent and received since the last, as they come (gathered for
/// `MESSAGES_GATHERED` after the first), as the page's own `messages` events have them,
/// without each message written out, for a page of another member subscribed to this node: its console lists them,
/// and its map runs them along this node's links. `offline` when the node stops, which
/// ends it. The message log is kept while it is open.
async fn own_messages_events(Extension(hub): Extension<Arc<Hub>>) -> impl IntoResponse {
    let log = &hub.state.traffic.log;
    log.watch(true);
    let watching = Watching(Arc::clone(&hub));
    let mut logged = log.latest();
    let stream = async_stream::stream! {
        let _watching = watching;
        loop {
            // As they come, gathered a moment (`MESSAGES_GATHERED`).
            let log = &hub.state.traffic.log;
            tokio::select! {
                () = async {
                    log.logged_after(logged).await;
                    tokio::time::sleep(MESSAGES_GATHERED).await;
                } => {}
                () = hub.closed() => {
                    yield Ok::<_, Infallible>(offline_event());
                    break;
                }
            }
            let mut messages = log.after(logged, MESSAGES_AT_MOST);
            let Some(last) = messages.last() else {
                continue;
            };
            logged = last.seq;
            by_node(&hub.state, &mut messages).await;
            let messages: Vec<JsonValue> = messages
                .into_iter()
                .filter_map(|message| {
                    let mut json = serde_json::to_value(message).ok()?;
                    json.as_object_mut()?.remove("text");
                    Some(json)
                })
                .collect();
            yield Ok(Event::default().event("messages").data(JsonValue::Array(messages).to_string()));
        }
    };
    (
        [(super::UNBUFFERED.0, super::UNBUFFERED.1)],
        Sse::new(stream).keep_alive(KeepAlive::default()),
    )
}

/// The page as built into this binary.
const PAGE: &str = include_str!("dashboard.html");
/// Where the page says which version of it it is.
const PAGE_VERSION_MARK: &str = "__PAGE_VERSION__";
/// Where the page gets the age fields (`GROWING`), so the list is kept once.
const GROWING_MARK: &str = "__GROWING__";

/// The page's version (a hash of it), and the page saying so: a page whose version
/// differs from the snapshot's `page` was served by another build, and reloads.
fn page() -> &'static (String, String) {
    static PAGE_SERVED: OnceLock<(String, String)> = OnceLock::new();
    PAGE_SERVED.get_or_init(|| {
        use sha2::{Digest, Sha256};
        let growing = serde_json::to_string(GROWING).unwrap_or_else(|_| "[]".to_owned());
        let page = PAGE.replace(GROWING_MARK, &growing);
        // Of the page with its list: a new list is a new version too.
        let version = Hex(&Sha256::digest(page.as_bytes())[..8]).to_string();
        let served = page.replace(PAGE_VERSION_MARK, &version);
        (version, served)
    })
}

/// What a page of ours may do, `page` (whose one script is inline) being it: run its
/// own script and nothing else (an injected one, should a value shown slip through, gets
/// no token it holds), be framed by no other page (`frame-ancestors`: one framing it
/// would have it act with its user's token), and fetch from its node and the members'
/// dashboards (any origin: those the member list names).
fn content_policy(page: &str) -> String {
    use base64::{Engine, engine::general_purpose::STANDARD};
    use sha2::{Digest, Sha256};
    let script = page
        .split_once("<script>")
        .and_then(|(_, rest)| rest.split_once("</script>"))
        .map_or("", |(script, _)| script);
    let hash = STANDARD.encode(Sha256::digest(script.as_bytes()));
    format!(
        "default-src 'self'; script-src 'sha256-{hash}'; style-src 'self' 'unsafe-inline'; \
         img-src 'self' data:; connect-src 'self' http: https:; frame-ancestors 'none'; \
         base-uri 'none'; form-action 'self'; object-src 'none'"
    )
}

/// `page`, with its content policy (`content_policy`) and framed by none.
fn guarded(page: &'static str, policy: &'static str) -> Response {
    (
        [
            (axum::http::header::CONTENT_SECURITY_POLICY, policy),
            (axum::http::header::X_FRAME_OPTIONS, "DENY"),
        ],
        Html(page),
    )
        .into_response()
}

/// `GET /login`: a page to log in on, by password or with a token, without
/// authentication (it holds nothing): it keeps a session for the browser
/// (`POST /auth/session`) and opens the dashboard.
pub(crate) async fn login_page() -> Response {
    const LOGIN: &str = include_str!("login.html");
    static POLICY: OnceLock<String> = OnceLock::new();
    guarded(LOGIN, POLICY.get_or_init(|| content_policy(LOGIN)))
}

/// `GET /admin/dashboard`: the page.
async fn page_html() -> Response {
    static POLICY: OnceLock<String> = OnceLock::new();
    let page = page().1.as_str();
    guarded(page, POLICY.get_or_init(|| content_policy(page)))
}

/// `POST /admin/dashboard/refresh`: a tick now rather than at the next `TICK`, and the
/// graph recomputed; what changed comes as events, as ever.
async fn refresh(Extension(hub): Extension<Arc<Hub>>) -> StatusCode {
    hub.want();
    hub.poke.notify_one();
    StatusCode::NO_CONTENT
}

#[derive(serde::Deserialize)]
struct TreeQuery {
    writer: String,
}

/// `GET /admin/dashboard/tree?writer=…`: the writer's tree (`swarm::tree::whole`), as
/// this node computes it: from the member list, and whom it and its neighbours reach
/// and how far, as they said (others as reaching their neighbours by the member list):
/// every member's parent, `null` for one without. `404` if the writer isn't a member.
async fn writer_tree(
    Extension(hub): Extension<Arc<Hub>>,
    Query(query): Query<TreeQuery>,
) -> Response {
    let state = &hub.state;
    match crate::swarm::tree::whole(state, &query.writer).await {
        Some(parents) => {
            axum::Json(json!({ "writer": query.writer, "parents": parents })).into_response()
        }
        None => StatusCode::NOT_FOUND.into_response(),
    }
}

/// `GET /admin/dashboard/state`: the latest snapshot, for a client that wants it whole.
async fn current_state(Extension(hub): Extension<Arc<Hub>>) -> Response {
    hub.want();
    let Some(current) = hub.current().await else {
        return StatusCode::SERVICE_UNAVAILABLE.into_response();
    };
    (
        [(CONTENT_TYPE, HeaderValue::from_static("application/json"))],
        current.json().to_owned(),
    )
        .into_response()
}

/// `GET /admin/dashboard/events`: Server-Sent Events: a `snapshot` first, then a
/// `patch` whenever something changed (each with its number as the event's ID; a
/// client that fell behind gets a `snapshot` again), a `document` where the document
/// changed (`Hub::documents`, at most every tick), `messages` with the messages since
/// the last tick (`Hub::messages`), and `offline` when the node stops, which ends the
/// stream.
async fn events(
    Extension(hub): Extension<Arc<Hub>>,
    Extension(access): Extension<super::access::Access>,
) -> impl IntoResponse {
    // Subscribed before the snapshot is read: no patch after it is missed.
    let mut patches = hub.patches.subscribe();
    let mut documents = hub.documents.subscribe();
    let mut messages = hub.messages.subscribe();
    hub.want();
    let first = hub.current().await;
    let stream = async_stream::stream! {
        let Some(first) = first else {
            yield Ok::<_, Infallible>(offline_event());
            return;
        };
        let mut seen = first.seq;
        yield Ok(snapshot_event(&first));
        loop {
            let next = tokio::select! {
                received = patches.recv() => Next::Patch(received),
                changed = documents.recv() => Next::Document(changed),
                logged = messages.recv() => Next::Messages(logged),
                () = hub.closed() => Next::Closing,
            };
            let received = match next {
                Next::Patch(received) => received,
                Next::Document(Ok(changed)) => {
                    if let Some(changed) = changed_for(&changed, &access) {
                        yield Ok(Event::default().event("document").data(changed));
                    }
                    continue;
                }
                Next::Document(Err(broadcast::error::RecvError::Lagged(_))) => {
                    yield Ok(Event::default().event("document").data(r#"{"all":true}"#));
                    continue;
                }
                Next::Document(Err(broadcast::error::RecvError::Closed)) => break,
                Next::Messages(Ok(logged)) => {
                    yield Ok(Event::default().event("messages").data(logged.as_str()));
                    continue;
                }
                // Missed ones are only missed in the console.
                Next::Messages(Err(broadcast::error::RecvError::Lagged(_))) => continue,
                Next::Messages(Err(broadcast::error::RecvError::Closed)) => break,
                Next::Closing => {
                    yield Ok(offline_event());
                    break;
                }
            };
            match received {
                Ok(patch) if patch.seq <= seen => {}
                Ok(patch) => {
                    seen = patch.seq;
                    yield Ok(Event::default()
                        .event("patch")
                        .id(patch.seq.to_string())
                        .data(patch.json()));
                }
                Err(broadcast::error::RecvError::Lagged(_)) => {
                    let Some(current) = hub.current().await else {
                        yield Ok(offline_event());
                        break;
                    };
                    seen = current.seq;
                    yield Ok(snapshot_event(&current));
                }
                Err(broadcast::error::RecvError::Closed) => break,
            }
        }
    };
    (
        // Through nginx too as it comes, not buffered (`operations.md`).
        [(super::UNBUFFERED.0, super::UNBUFFERED.1)],
        Sse::new(stream).keep_alive(KeepAlive::default()),
    )
}

/// The node stops: the client may reconnect once it is back.
fn offline_event() -> Event {
    Event::default().event("offline").data("{}")
}

fn snapshot_event(published: &Published) -> Event {
    Event::default()
        .event("snapshot")
        .id(published.seq.to_string())
        .data(published.json())
}

/// A `document` event (`Hub::documents`) as the page of a user with `access` is told
/// it: of the changed paths, those it may read at least partly (`access::Visible`), so
/// that it isn't told where in the document others write what it may not read. None if
/// none is left. A few prefix comparisons per path and capability, once a tick per
/// page.
fn changed_for(changed: &str, access: &super::access::Access) -> Option<String> {
    use super::access::{Access, Visible};
    if matches!(access, Access::Everything) {
        return Some(changed.to_owned());
    }
    let mut parsed: JsonValue = serde_json::from_str(changed).ok()?;
    let Some(paths) = parsed.get_mut("paths").and_then(JsonValue::as_array_mut) else {
        // `{"all": true}`: the page reads again what it shows, as its user may.
        return Some(changed.to_owned());
    };
    paths.retain(|path| {
        let path: Vec<String> = path
            .as_array()
            .map(|segments| {
                segments
                    .iter()
                    .filter_map(|segment| segment.as_str().map(str::to_owned))
                    .collect()
            })
            .unwrap_or_default();
        access.visible(&path) != Visible::Nothing
    });
    (!paths.is_empty()).then(|| parsed.to_string())
}

#[cfg(test)]
mod tests {
    use super::*;

    /// A page's content policy lets its own inline script run, by its hash, and no
    /// other page frame it.
    #[test]
    fn pages_run_their_own_script_only() {
        use base64::{Engine, engine::general_purpose::STANDARD};
        use sha2::{Digest, Sha256};
        let policy = content_policy("<p>x</p><script>go()</script>");
        let hash = STANDARD.encode(Sha256::digest(b"go()"));
        assert!(policy.contains(&format!("script-src 'sha256-{hash}';")));
        assert!(policy.contains("frame-ancestors 'none'"));
        // Each page's one script, found.
        for page in [PAGE, include_str!("login.html")] {
            assert_eq!(page.matches("<script>").count(), 1);
            assert!(!content_policy(page).contains(&STANDARD.encode(Sha256::digest(b""))));
        }
    }

    /// The page in a browser (`check.js`): it loads without errors, the peers table and
    /// the map draw what the state says, and a large map doesn't hold the page up.
    /// Needs Node.js with Playwright's library and its Chromium (`check.nix` gives them),
    /// so only when asked: `cargo test --features dashboard dashboard_in_a_browser --
    /// --ignored`.
    #[test]
    #[ignore]
    fn dashboard_in_a_browser() {
        let file =
            std::env::temp_dir().join(format!("continuum-dashboard-{}.html", std::process::id()));
        std::fs::write(&file, &page().1).expect("page written");
        let check =
            std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("src/api/dashboard/check.js");
        let status = std::process::Command::new("node")
            .arg(&check)
            .arg(&file)
            .status()
            .unwrap_or_else(|error| {
                panic!(
                    "Node.js doesn't run ({error}): the browser checks need `node` on the \
                     PATH, with Playwright's library and its Chromium (check.nix)"
                )
            });
        let _ = std::fs::remove_file(&file);
        assert!(status.success(), "the browser checks failed");
    }

    #[test]
    fn merge_patches() {
        let old = json!({ "a": 1, "b": { "c": 2, "d": [1, 2] }, "gone": true });
        let new = json!({ "a": 1, "b": { "c": 3, "d": [1, 2] }, "added": "x" });
        assert_eq!(
            merge_patch(&old, &new),
            Some(json!({ "b": { "c": 3 }, "gone": null, "added": "x" }))
        );
        assert_eq!(merge_patch(&old, &old), None);
        assert_eq!(
            merge_patch(&json!([1]), &json!([1, 2])),
            Some(json!([1, 2]))
        );
    }

    #[test]
    fn each_page_is_told_of_what_its_user_may_read() {
        use super::super::access::Access;
        use crate::access::{Capability, Caps};
        let reader = Access::Caps(Caps::new(vec![Capability {
            path: vec!["servers".to_owned(), "eu".to_owned()],
            rights: ["read".to_owned()].into(),
            by: String::new(),
            since: 0,
        }]));
        let changed = json!({ "paths": [["servers", "eu", "a"], ["servers", "us"], ["servers"]] })
            .to_string();
        let told: JsonValue =
            serde_json::from_str(&changed_for(&changed, &reader).unwrap()).unwrap();
        assert_eq!(
            told,
            json!({ "paths": [["servers", "eu", "a"], ["servers"]] })
        );
        let elsewhere = json!({ "paths": [["accounts"]] }).to_string();
        assert_eq!(changed_for(&elsewhere, &reader), None);
        assert_eq!(
            changed_for(r#"{"all":true}"#, &reader).as_deref(),
            Some(r#"{"all":true}"#)
        );
        assert_eq!(
            changed_for(&elsewhere, &Access::Everything),
            Some(elsewhere.clone())
        );
    }

    #[test]
    fn ages_alone_are_no_change() {
        let old =
            json!({ "uptimeSecs": 1, "peers": { "p": { "lastSeenSecsAgo": 3, "rttMs": 2 } } });
        let older =
            json!({ "uptimeSecs": 9, "peers": { "p": { "lastSeenSecsAgo": 7, "rttMs": 2 } } });
        assert_eq!(without_ages(&old), without_ages(&older));
        let moved =
            json!({ "uptimeSecs": 9, "peers": { "p": { "lastSeenSecsAgo": 7, "rttMs": 5 } } });
        assert_ne!(without_ages(&old), without_ages(&moved));
    }
}