//! Pushing changes to clients: every mutation publishes a [`ChangeSet`] of
//! the records it touched, broadcast over `GET /api/events` (SSE).
//! `GET /api/snapshot` gives clients a consistent starting point, stamped
//! with the revision the events continue from.

use std::{convert::Infallible, sync::atomic::Ordering};

use axum::{
    Json,
    extract::State,
    response::{
        IntoResponse,
        sse::{Event, KeepAlive, Sse},
    },
};
use entity::NanoId;
use fil_api::{Change, ChangeSet};
use fil_core::{Kasten, ZettelId};
use tokio_stream::{Stream, StreamExt as _, wrappers::BroadcastStream};
use tracing::{debug, instrument};

use super::{FilState, error::ApiError};

/// How often an idle event stream sends a keep-alive comment.
const KEEP_ALIVE: std::time::Duration = std::time::Duration::from_secs(15);

impl FilState {
    /// Broadcasts `changes` to every subscribed client under the next
    /// revision.
    ///
    /// Takes the `Kasten` only to prove the write lock is still held:
    /// publishing under it keeps revisions in mutation order and lets
    /// snapshots read a revision that matches their data.
    pub fn publish(&self, _locked: &mut Kasten, changes: Vec<Change>) {
        let rev = self.rev.fetch_add(1, Ordering::SeqCst) + 1;
        debug!(rev, count = changes.len(), "publishing changes");
        // no subscribers is fine: nobody needs telling
        let _ = self.diff_tx.send(ChangeSet {
            epoch: self.epoch,
            rev,
            changes,
        });
    }
}

/// `GET /api/snapshot`: every record, plus the revision they reflect, read
/// under one lock so they are consistent.
#[axum::debug_handler]
#[instrument(skip(state))]
pub async fn snapshot(State(state): State<FilState>) -> Result<impl IntoResponse, ApiError> {
    let kt = state.kh.read().await;
    let rev = state.rev.load(Ordering::SeqCst);
    let snapshot = fil_api::snapshot(&kt, state.epoch, rev).await?;
    drop(kt);
    Ok(Json(snapshot))
}

/// `GET /api/events`: an SSE stream of `changes` events, each a
/// [`ChangeSet`] as JSON. If this subscriber falls too far behind it gets a
/// `resync` event instead, telling it to fetch a fresh snapshot.
#[instrument(skip(state))]
pub async fn events(
    State(state): State<FilState>,
) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
    let stream = BroadcastStream::new(state.diff_tx.subscribe()).map(|set| {
        let event = set
            .ok()
            .and_then(|set| Event::default().event("changes").json_data(set).ok());
        Ok(event.unwrap_or_else(|| Event::default().event("resync").data("")))
    });
    // clients time a stalled stream out (see `client::READ_TIMEOUT`), so keep
    // this well under that
    Sse::new(stream).keep_alive(KeepAlive::new().interval(KEEP_ALIVE))
}

/// The current state of task `id`, as a change to publish.
pub async fn task_change(id: &NanoId, kt: &Kasten) -> Result<Change, ApiError> {
    fil_api::fetch_task(id, kt)
        .await?
        .map(Change::Task)
        .ok_or_else(|| ApiError::NotFound(format!("no task with id {id}")))
}

/// The current state of group `id`, as a change to publish.
pub async fn group_change(id: &NanoId, kt: &Kasten) -> Result<Change, ApiError> {
    fil_api::fetch_group(id, kt)
        .await?
        .map(Change::Group)
        .ok_or_else(|| ApiError::NotFound(format!("no group with id {id}")))
}

/// The current state of tag `id`, as a change to publish.
pub async fn tag_change(id: &NanoId, kt: &Kasten) -> Result<Change, ApiError> {
    fil_api::fetch_tag(id, kt)
        .await?
        .map(Change::Tag)
        .ok_or_else(|| ApiError::NotFound(format!("no tag with id {id}")))
}

/// The current state of zettel `id`, preceded by its tags (a client may not
/// have them yet, e.g. a tag just typed into the front matter).
pub async fn zettel_changes(id: &ZettelId, kt: &Kasten) -> Result<Vec<Change>, ApiError> {
    let (zettel, tags) = fil_api::fetch_zettel(id, kt)
        .await?
        .ok_or_else(|| ApiError::NotFound(format!("no zettel with id {id}")))?;
    let mut changes: Vec<Change> = tags.into_iter().map(Change::Tag).collect();
    changes.push(Change::Zettel(zettel));
    Ok(changes)
}