//! `GET /live?topics=`: one stream of changes for a page (docs/live.md §4,
//! oxygenhub#204). Every topic must be one the viewer may read, or nothing
//! is streamed; an event names a topic, its version and a kind, never data,
//! and the page fetches what changed from its normal routes.

use crate::auth::Viewer;
use crate::db;
use crate::events::{LiveEvent, Replay};
use crate::Shared;
use axum::extract::{Query, State};
use axum::http::{HeaderMap, StatusCode};
use axum::response::sse::{Event, KeepAlive, Sse};
use axum::response::{IntoResponse, Response};
use serde::Deserialize;

/// The most topics one page asks for.
const MAX_TOPICS: usize = 16;

/// A topic, read (docs/live.md §4).
#[derive(Debug, PartialEq, Eq)]
pub enum Topic {
    Repo(i64),
    Proposal(i64, i64),
    Issue(i64, i64),
    Run(i64),
    User(i64),
    Diffusion(i64),
}

pub fn parse_topic(s: &str) -> Option<Topic> {
    let num = |v: &str| v.parse::<i64>().ok().filter(|n| *n > 0);
    if let Some(rest) = s.strip_prefix("repo:") {
        return match rest.split_once('/') {
            None => num(rest).map(Topic::Repo),
            Some((id, sub)) => {
                let id = num(id)?;
                if let Some(n) = sub.strip_prefix("proposal:") {
                    num(n).map(|n| Topic::Proposal(id, n))
                } else if let Some(n) = sub.strip_prefix("issue:") {
                    num(n).map(|n| Topic::Issue(id, n))
                } else {
                    None
                }
            }
        };
    }
    if let Some(id) = s.strip_prefix("run:") {
        return num(id).map(Topic::Run);
    }
    if let Some(id) = s.strip_prefix("user:") {
        return num(id).map(Topic::User);
    }
    s.strip_prefix("diffusion:").and_then(num).map(Topic::Diffusion)
}

/// Whether `viewer` may hear of changes to `topic`: read access to its
/// repository, or, for `user:`, being that user.
async fn may_hear(app: &Shared, viewer: &Viewer, topic: &Topic) -> bool {
    let repo_id = match topic {
        Topic::Repo(r) | Topic::Proposal(r, _) | Topic::Issue(r, _) | Topic::Diffusion(r) => *r,
        Topic::Run(run) => match db::flow::run_by_id(&app.db, *run).await {
            Ok(Some(r)) => r.repo_id,
            _ => return false,
        },
        Topic::User(id) => return viewer.account().is_some_and(|a| a.id == *id),
    };
    let Ok(Some(row)) = db::repos::get_by_id(&app.db, repo_id).await else { return false };
    matches!(crate::auth::roles::role_for(&app.db, viewer.account(), &row).await, Ok(Some(_)))
}

#[derive(Deserialize)]
pub struct LiveQuery {
    #[serde(default)]
    topics: String,
}

fn sse_event(e: &LiveEvent) -> Event {
    Event::default()
        .id(e.id.to_string())
        .event("change")
        .data(serde_json::json!({ "topic": e.topic, "version": e.version, "kind": e.kind }).to_string())
}

/// `GET /live?topics=repo:1,repo:1/proposal:4`.
pub async fn stream(State(app): State<Shared>, viewer: Viewer, headers: HeaderMap, Query(q): Query<LiveQuery>) -> Response {
    use futures::StreamExt;
    let mut topics: Vec<String> = q.topics.split(',').map(str::trim).filter(|t| !t.is_empty()).map(String::from).collect();
    topics.sort();
    topics.dedup();
    if topics.is_empty() || topics.len() > MAX_TOPICS {
        return (StatusCode::BAD_REQUEST, format!("ask for 1 to {MAX_TOPICS} topics, as /live?topics=repo:1")).into_response();
    }
    for t in &topics {
        let Some(topic) = parse_topic(t) else {
            return (StatusCode::BAD_REQUEST, format!("{t:?} is not a topic")).into_response();
        };
        if !may_hear(&app, &viewer, &topic).await {
            // The same answer for every topic one may not hear: whether it
            // exists is not said either.
            return (StatusCode::FORBIDDEN, "not every topic asked for may be heard").into_response();
        }
    }
    // Subscribed before the replay is read, so nothing falls between them.
    let rx = app.events.live.subscribe();
    let last: Option<u64> = headers.get("last-event-id").and_then(|v| v.to_str().ok()).and_then(|v| v.parse().ok());
    let (replay, after) = match last {
        Some(id) => match app.events.live.since(id, &topics) {
            Replay::Events(e) => {
                let after = e.last().map(|e| e.id).unwrap_or(id);
                (e.iter().map(|e| Ok(sse_event(e))).collect::<Vec<_>>(), after)
            }
            Replay::Reload => (vec![Ok(Event::default().event("reload").data("{}"))], app.events.live.last_id()),
        },
        None => (vec![], app.events.live.last_id()),
    };
    let live = tokio_stream::wrappers::BroadcastStream::new(rx).filter_map(move |item| {
        let topics = topics.clone();
        async move {
            match item {
                Ok(e) if e.id > after && topics.contains(&e.topic) => Some(Ok::<_, std::convert::Infallible>(sse_event(&e))),
                // A subscriber that fell behind is told to reload.
                Err(_) => Some(Ok(Event::default().event("reload").data("{}"))),
                _ => None,
            }
        }
    });
    // A first comment at once, so a proxy that waits for body bytes (the
    // tunnel does) sends the headers now rather than at the first keep-alive.
    let hello = std::iter::once(Ok(Event::default().comment("live")));
    Sse::new(futures::StreamExt::take_until(futures::stream::iter(hello.chain(replay)).chain(live), crate::shutdown::stopping()))
        .keep_alive(KeepAlive::default())
        .into_response()
}

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

    #[test]
    fn topics_are_read_strictly() {
        assert_eq!(parse_topic("repo:7"), Some(Topic::Repo(7)));
        assert_eq!(parse_topic("repo:7/proposal:4"), Some(Topic::Proposal(7, 4)));
        assert_eq!(parse_topic("repo:7/issue:9"), Some(Topic::Issue(7, 9)));
        assert_eq!(parse_topic("run:12"), Some(Topic::Run(12)));
        assert_eq!(parse_topic("user:3"), Some(Topic::User(3)));
        assert_eq!(parse_topic("diffusion:7"), Some(Topic::Diffusion(7)));
        for bad in ["repo:", "repo:x", "repo:-1", "repo:7/wiki:1", "repo:7/proposal:", "team:1", "", "run:0"] {
            assert_eq!(parse_topic(bad), None, "{bad}");
        }
    }
}