//! Axum SSE adapter. Behind the `axum` feature; the rest of the crate has no
//! framework dependency.
//!
//! SSE rather than a WebSocket, on purpose. Upstream traffic is one `POST` per
//! message typed, so bidirectionality buys nothing, and a `POST` passes through
//! the CSRF, rate-limit, and session middleware both consumers already run. It
//! also avoids adding a `connect-src` CSP directive on the Multithreaded side,
//! which currently has none and falls back to `default-src 'self'`.

use std::convert::Infallible;
use std::time::Duration;

use axum::response::sse::{Event, KeepAlive, Sse};
use tokio_stream::Stream;

use crate::event::ChatEvent;
use crate::stream::ChatStream;

/// Keepalive interval.
///
/// Cloudflare fronts makenot.work and idles a connection out at 100 seconds. A
/// 30 second comment clears that with room for a missed tick, and matches what
/// the SyncKit SSE endpoint already runs in production.
pub const KEEPALIVE_INTERVAL: Duration = Duration::from_secs(30);

/// Serialize one event into an SSE frame.
///
/// The event name is carried in the SSE `event:` field so the client can dispatch
/// without parsing the payload first, and the payload is the same tagged JSON, so
/// a client that prefers to read `type` can.
fn frame(event: &ChatEvent) -> Event {
    let data = serde_json::to_string(event).unwrap_or_else(|_| {
        // A ChatEvent is plain data and cannot fail to serialize. If that ever
        // changes, a dead connection is worse than a frame the client ignores.
        tracing::error!("chat event failed to serialize");
        r#"{"type":"gap"}"#.to_owned()
    });
    Event::default().event(event.name()).data(data)
}

/// Wrap a [`ChatStream`] as an axum SSE response.
///
/// The stream owns the hub subscription, so dropping the response (which axum
/// does when the client disconnects) releases the connection slot and prunes the
/// room if it was the last listener. There is nothing for a caller to remember to
/// clean up.
pub fn sse_response(
    stream: ChatStream,
) -> Sse<impl Stream<Item = Result<Event, Infallible>> + Send> {
    let events = async_stream::stream! {
        let mut stream = stream;
        while let Some(event) = stream.next().await {
            yield Ok(frame(&event));
        }
    };

    Sse::new(events).keep_alive(
        KeepAlive::new()
            .interval(KEEPALIVE_INTERVAL)
            .text("keepalive"),
    )
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::ids::{MessageId, RoomId, UserId};
    use crate::message::Message;
    use uuid::Uuid;

    fn message(id: i64) -> Message {
        Message {
            id: MessageId(id),
            room_id: RoomId(Uuid::nil()),
            author_id: UserId(Uuid::nil()),
            body_html: "hi".into(),
            created_at: 0,
            nonce: None,
        }
    }

    /// The SSE wire format is a contract with the client island, so assert on the
    /// bytes rather than on the builder.
    ///
    /// `Event` exposes no accessor for its encoded form, so this reads the `Debug`
    /// output, which embeds the real buffer. If an axum upgrade ever makes that
    /// opaque these assertions go quiet rather than failing, so
    /// [`debug_output_still_exposes_the_buffer`] guards the technique itself.
    fn wire(event: &ChatEvent) -> String {
        format!("{:?}", frame(event))
    }

    #[test]
    fn debug_output_still_exposes_the_buffer() {
        assert!(
            wire(&ChatEvent::Gap).contains("event: gap\\n"),
            "Event's Debug no longer embeds the encoded frame, so the wire-format \
             assertions in this module have stopped checking anything"
        );
    }

    #[test]
    fn every_variant_frames_with_its_own_event_name() {
        for (event, name) in [
            (ChatEvent::Message(message(1)), "message"),
            (ChatEvent::Delete { id: MessageId(1) }, "delete"),
            (
                ChatEvent::Purge {
                    author_id: UserId(Uuid::nil()),
                },
                "purge",
            ),
            (ChatEvent::Gap, "gap"),
        ] {
            // Match the SSE `event:` field specifically. A bare `contains(name)`
            // would also be satisfied by the tagged JSON in the data field, which
            // means it would pass even if the event name were never set.
            assert!(
                wire(&event).contains(&format!("event: {name}\\n")),
                "expected an SSE event field named {name}, got: {}",
                wire(&event)
            );
        }
    }

    #[test]
    fn body_html_is_carried_verbatim() {
        // The host renders through docengine before publishing; the transport
        // must not re-encode or re-escape what it was given.
        let mut m = message(1);
        m.body_html = "<em>hi</em>".into();
        let json = serde_json::to_string(&ChatEvent::Message(m)).unwrap();
        assert!(json.contains(r#"<em>hi</em>"#), "{json}");
    }

    #[test]
    fn a_frame_never_contains_a_bare_newline() {
        // SSE is newline-delimited. A payload with a raw newline would split one
        // event into two malformed ones, so the JSON encoding has to be the thing
        // that protects the framing.
        let mut m = message(1);
        m.body_html = "line one\nline two".into();
        let json = serde_json::to_string(&ChatEvent::Message(m)).unwrap();
        assert!(
            !json.contains('\n'),
            "raw newline reached the frame: {json}"
        );
        assert!(
            json.contains(r"\n"),
            "newline should survive escaped: {json}"
        );
    }
}