//! `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,

            }

        }

    });

    Sse::new(futures::stream::iter(replay).chain(live)).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}");

        }

    }

}