use axum::{
    extract::State,
    response::{
        sse::{self, Sse},
        IntoResponse,
    },
    routing::get,
    Router,
};
use axum_extra::extract::Query;
use chrono::{format::SecondsFormat, DateTime};
use futures::{
    future,
    stream::{self, StreamExt as _, TryStreamExt as _},
};

use crate::{
    app::App,
    channel::repo::broadcast,
    error::{BoxedError, InternalError},
    header::LastEventId,
    repo::{channel, login::Login},
};

pub fn router() -> Router<App> {
    Router::new().route("/events", get(on_events))
}

#[derive(serde::Deserialize)]
struct EventsQuery {
    #[serde(default, rename = "channel")]
    channels: Vec<channel::Id>,
}

async fn on_events(
    State(app): State<App>,
    _: Login, // requires auth, but doesn't actually care who you are
    last_event_id: Option<LastEventId>,
    Query(query): Query<EventsQuery>,
) -> Result<impl IntoResponse, InternalError> {
    let resume_at = last_event_id
        .map(|LastEventId(header)| header)
        .map(|header| DateTime::parse_from_rfc3339(&header))
        .transpose()?
        .map(|ts| ts.to_utc());

    let streams = stream::iter(query.channels)
        .then(|channel| {
            let app = app.clone();
            async move {
                let events = app
                    .channels()
                    .events(&channel, resume_at.as_ref())
                    .await?
                    .map_ok(ChannelEvent::wrap(channel));

                Ok::<_, BoxedError>(events)
            }
        })
        .try_collect::<Vec<_>>()
        .await?;

    let stream = stream::select_all(streams).and_then(|msg| future::ready(to_sse_event(msg)));
    let sse = Sse::new(stream).keep_alive(sse::KeepAlive::default());

    Ok(sse)
}

fn to_sse_event(event: ChannelEvent<broadcast::Message>) -> Result<sse::Event, BoxedError> {
    let data = serde_json::to_string(&event)?;
    let event = sse::Event::default()
        .id(event
            .message
            .sent_at
            .to_rfc3339_opts(SecondsFormat::AutoSi, /* use_z */ true))
        .data(&data);

    Ok(event)
}

#[derive(serde::Serialize)]
struct ChannelEvent<M> {
    channel: channel::Id,
    #[serde(flatten)]
    message: M,
}

impl<M> ChannelEvent<M> {
    fn wrap(channel: channel::Id) -> impl Fn(M) -> Self {
        move |message| Self {
            channel: channel.clone(),
            message,
        }
    }
}