//! The change publisher: after document changes, which store paths they touched
//! (`AppState::change_events`, for the API's streams, the dashboard and remakes), and
//! a nudge to topology when the member list changed, and to the peer worker when the
//! forget records did.

use std::sync::Arc;

use tracing::{info, warn};

use crate::document::changes;
use crate::document::store::{self, ChangeSet, InLists};
use crate::state::AppState;

pub(crate) async fn run_change_publisher(state: AppState) {
    let mut changed = state.change_tx.subscribe();
    // The replica's heads (sorted, as of its last refresh: every change refreshes them).
    let mut last_heads = state.doc.read().await.heads().to_vec();

    info!("Starting change publisher");

    while changed.changed().await.is_ok() {
        let before = last_heads.clone();

        let doc = Arc::clone(&state.doc);
        let (heads, paths, conflicts) = tokio::task::spawn_blocking(move || {
            let replica = doc.blocking_read();
            let heads = replica.heads().to_vec();
            if heads == before {
                return (heads, Vec::new(), Vec::new());
            }

            // After an epoch switch the previous heads belong to the old document:
            // report the root as changed, which every stream is related to.
            if !changes::missing(&replica.doc, &before).is_empty() {
                return (heads, vec![Vec::new()], Vec::new());
            }

            let mut paths = Vec::new();
            let mut conflicts = Vec::new();
            for changed in store::changed(&replica.doc, &before, &heads, InLists::Index) {
                if changed.conflict {
                    conflicts.push(changed.path.clone());
                }
                paths.push(changed.path);
            }
            (heads, paths, conflicts)
        })
        .await
        .expect("change publisher task panicked");

        if heads == last_heads {
            continue;
        }
        last_heads = heads;

        for path in &conflicts {
            warn!(path = ?path, "Concurrent writes produced a conflict");
        }

        if paths
            .iter()
            .any(|path| crate::swarm::topology::affects_neighbours(path))
        {
            state.topology_changed.notify_one();
        }
        if paths
            .iter()
            .any(|path| crate::swarm::peer::affects_forget_records(path))
        {
            state
                .forget_records_changed
                .store(true, std::sync::atomic::Ordering::Release);
            state.peers_due.now();
        }

        // Fails only when nobody is subscribed.
        let _ = state.change_events.send(Arc::new(ChangeSet { paths }));
    }
}