//! Sync lanes: inbound sync messages are applied off the consumer, so one slow apply (a
//! large delta takes seconds) doesn't hold up keepalives, HELLOs and blob chunks behind
//! it. Each sender's messages go to the same lane, in arrival order; a few lanes are
//! enough, since applying takes the document's write lock anyway.

use std::collections::hash_map::DefaultHasher;
use std::hash::{Hash, Hasher};

use tokio::sync::mpsc::{self, error::TrySendError};
use tracing::{debug, warn};

use crate::state::AppState;
use crate::swarm::sync::handle_sync;
use crate::transport::SyncEnvelope;

const LANES: usize = 4;
/// Sync messages waiting per lane. A full lane drops what arrives: its sender gets no
/// answer and resends after its reply timeout, by which time the backlog is gone.
const QUEUED_PER_LANE: usize = 256;

pub(crate) struct SyncLanes {
    lanes: Vec<mpsc::Sender<(String, SyncEnvelope)>>,
}

impl SyncLanes {
    /// Starts a task per lane.
    pub(crate) fn start(state: &AppState) -> Self {
        let lanes = (0..LANES)
            .map(|_| {
                let (tx, mut rx) = mpsc::channel::<(String, SyncEnvelope)>(QUEUED_PER_LANE);
                let state = state.clone();
                tokio::spawn(async move {
                    while let Some((peer, envelope)) = rx.recv().await {
                        if let Err(error) = handle_sync(&state, &peer, envelope).await {
                            warn!(peer = %peer, error = %format!("{error:#}"), "Failed to process sync message");
                        }
                    }
                });
                tx
            })
            .collect();
        Self { lanes }
    }

    /// Queues `envelope` from `peer` on its lane.
    pub(crate) fn enqueue(&self, peer: &str, envelope: SyncEnvelope) {
        let mut hasher = DefaultHasher::new();
        peer.hash(&mut hasher);
        let lane = &self.lanes[hasher.finish() as usize % LANES];
        match lane.try_send((peer.to_owned(), envelope)) {
            Ok(()) => {}
            Err(TrySendError::Full(_)) => {
                debug!(peer = %peer, "Sync lane full; dropping the message (its sender resends)")
            }
            Err(TrySendError::Closed(_)) => {
                warn!(peer = %peer, "Sync lane stopped; dropping the message")
            }
        }
    }
}