solidb 1.0.2

//! Sync worker for background replication
//!
//! Handles:
//! - Incremental sync with peers
//! - Full sync for new nodes
//! - Heartbeat sending/receiving
//! - Dead node detection and removal
//! - Shard rebalancing

use std::sync::Arc;
use std::time::Duration;
use tokio::sync::mpsc;
use tracing::{debug, error, info, warn};

use super::protocol::{NodeStats, Operation, ShardConfig, SyncEntry, SyncMessage};
use super::state::SyncState;
use super::transport::{ConnectionPool, SyncServer, TransportError};
use crate::storage::StorageEngine;

/// Configuration for the sync worker
#[derive(Clone)]
pub struct SyncConfig {
    /// Heartbeat interval
    pub heartbeat_interval: Duration,
    /// Timeout before considering a node dead
    pub dead_node_timeout: Duration,
    /// Maximum batch size in bytes
    pub max_batch_bytes: u32,
    /// Sync interval (how often to check for updates)
    pub sync_interval: Duration,
    /// How often to prune the sync log. Set to None to disable auto-prune.
    pub prune_interval: Option<Duration>,
    /// Minimum number of entries to retain even if every peer is caught up.
    /// Acts as a buffer for peers that briefly fall behind.
    pub prune_retain_buffer: u64,
}

impl Default for SyncConfig {
    fn default() -> Self {
        Self {
            heartbeat_interval: Duration::from_secs(5),
            dead_node_timeout: Duration::from_secs(15),
            max_batch_bytes: 1024 * 1024, // 1 MB
            sync_interval: Duration::from_millis(1000),
            prune_interval: Some(Duration::from_secs(300)), // 5 minutes
            prune_retain_buffer: 10_000,
        }
    }
}

/// Command to send to the sync worker
pub enum SyncCommand {
    /// Request full sync from a peer
    RequestFullSync { peer_addr: String },
    /// Add a new peer
    AddPeer {
        node_id: String,
        sync_addr: String,
        http_addr: String,
    },
    /// Remove a peer
    RemovePeer { node_id: String },
    /// Shutdown the worker
    Shutdown,
}

/// Sync worker running in background
pub struct SyncWorker {
    storage: Arc<StorageEngine>,
    state: Arc<SyncState>,
    pool: Arc<ConnectionPool>,
    sync_log: Arc<super::log::SyncLog>,
    config: SyncConfig,
    command_rx: mpsc::Receiver<SyncCommand>,
    local_node_id: String,
    keyfile_path: String,
    listen_addr: String,
    incoming_rx: Option<mpsc::Receiver<(super::transport::SyncStream, String)>>,
    cluster_manager: Option<Arc<crate::cluster::manager::ClusterManager>>,
    shard_coordinator: Option<Arc<crate::sharding::ShardCoordinator>>,
    system: sysinfo::System,
}

impl SyncWorker {
    #[allow(clippy::too_many_arguments)]
    pub fn new(
        storage: Arc<StorageEngine>,
        state: Arc<SyncState>,
        pool: Arc<ConnectionPool>,
        sync_log: Arc<super::log::SyncLog>,
        config: SyncConfig,
        command_rx: mpsc::Receiver<SyncCommand>,
        local_node_id: String,
        keyfile_path: String,
        listen_addr: String,
    ) -> Self {
        Self {
            storage,
            state,
            pool,
            sync_log,
            config,
            command_rx,
            local_node_id,
            keyfile_path,
            listen_addr,
            incoming_rx: None,
            cluster_manager: None,
            shard_coordinator: None,
            system: sysinfo::System::new(),
        }
    }

    pub fn with_cluster_manager(
        mut self,
        manager: Arc<crate::cluster::manager::ClusterManager>,
    ) -> Self {
        self.cluster_manager = Some(manager);
        self
    }

    pub fn with_shard_coordinator(
        mut self,
        coordinator: Arc<crate::sharding::ShardCoordinator>,
    ) -> Self {
        self.shard_coordinator = Some(coordinator);
        self
    }

    pub fn with_incoming_channel(
        mut self,
        rx: mpsc::Receiver<(super::transport::SyncStream, String)>,
    ) -> Self {
        self.incoming_rx = Some(rx);
        self
    }

    /// Start the sync worker
    pub async fn run(self) {
        info!("Starting sync worker for node {}", self.local_node_id);

        // Start TCP server
        let server = match SyncServer::bind(
            &self.listen_addr,
            self.keyfile_path.clone(),
            self.local_node_id.clone(),
        )
        .await
        {
            Ok(s) => Arc::new(s),
            Err(e) => {
                error!("Failed to start sync server: {}", e);
                return;
            }
        };

        // Spawn server accept loop
        let accept_pool = self.pool.clone();
        let accept_state = self.state.clone();
        let accept_storage = self.storage.clone();
        let accept_sync_log = self.sync_log.clone();
        let accept_cluster_manager = self.cluster_manager.clone();
        let server_clone = server.clone();
        tokio::spawn(async move {
            loop {
                match server_clone.accept().await {
                    Ok((stream, addr)) => {
                        let pool = accept_pool.clone();
                        let state = accept_state.clone();
                        let storage = accept_storage.clone();
                        let sync_log = accept_sync_log.clone();
                        let cluster_manager = accept_cluster_manager.clone();
                        tokio::spawn(async move {
                            if let Err(e) = Self::handle_connection(
                                stream,
                                addr,
                                pool,
                                state,
                                storage,
                                sync_log,
                                cluster_manager,
                            )
                            .await
                            {
                                error!("Connection handler error: {}", e);
                            }
                        });
                    }
                    Err(e) => {
                        error!("Accept error: {}", e);
                    }
                }
            }
        });

        self.run_background().await;
    }

    /// Run background tasks (without binding a port)
    pub async fn run_background(mut self) {
        // Start incoming channel handler if present
        if let Some(mut rx) = self.incoming_rx.take() {
            let pool = self.pool.clone();
            let state = self.state.clone();
            let storage = self.storage.clone();
            let sync_log = self.sync_log.clone();
            let keyfile_path = self.keyfile_path.clone();
            let cluster_manager = self.cluster_manager.clone();

            tokio::spawn(async move {
                while let Some((stream, addr)) = rx.recv().await {
                    let pool = pool.clone();
                    let state = state.clone();
                    let storage = storage.clone();
                    let sync_log = sync_log.clone();
                    let keyfile = keyfile_path.clone();
                    let cluster_manager = cluster_manager.clone();

                    tokio::spawn(async move {
                        // We must authenticate the stream first, as it comes raw from the multiplexer
                        // The multiplexer already verified the magic header, so skip reading it again
                        match super::transport::SyncServer::authenticate_standalone_skip_magic(
                            stream, &keyfile,
                        )
                        .await
                        {
                            Ok(auth_stream) => {
                                if let Err(e) = Self::handle_connection(
                                    auth_stream,
                                    addr,
                                    pool,
                                    state,
                                    storage,
                                    sync_log,
                                    cluster_manager,
                                )
                                .await
                                {
                                    error!("Connection handler error: {}", e);
                                }
                            }
                            Err(e) => {
                                error!("Authentication failed for {}: {}", addr, e);
                            }
                        }
                    });
                }
            });
        }

        // Main worker loop
        let mut sync_interval = tokio::time::interval(self.config.sync_interval);
        let mut heartbeat_interval = tokio::time::interval(self.config.heartbeat_interval);
        let mut health_check_interval = tokio::time::interval(self.config.dead_node_timeout / 2);
        // Prune ticker is created even when prune is disabled — we just skip
        // the body in that case. Defaulting to a long interval keeps the
        // select! arm cheap.
        let prune_period = self
            .config
            .prune_interval
            .unwrap_or(Duration::from_secs(3600));
        let mut prune_interval_ticker = tokio::time::interval(prune_period);
        // Skip the immediate fire of `interval()` so we don't prune on startup.
        prune_interval_ticker.tick().await;

        loop {
            tokio::select! {
                // Handle commands
                Some(cmd) = self.command_rx.recv() => {
                    match cmd {
                        SyncCommand::Shutdown => {
                            info!("Sync worker shutting down");
                            break;
                        }
                        SyncCommand::AddPeer { node_id, sync_addr, http_addr } => {
                            self.state.add_peer(node_id, sync_addr, http_addr);
                            self.state.persist();
                        }
                        SyncCommand::RemovePeer { node_id } => {
                            self.state.remove_peer(&node_id);
                            self.state.persist();
                        }
                        SyncCommand::RequestFullSync { peer_addr } => {
                            if let Err(e) = self.request_full_sync(&peer_addr).await {
                                error!("Full sync request failed: {}", e);
                            }
                        }
                    }
                }

                // Periodic sync
                _ = sync_interval.tick() => {
                    self.sync_with_peers().await;
                }

                // Periodic heartbeat
                _ = heartbeat_interval.tick() => {
                    self.send_heartbeats().await;
                }

                // Health check
                _ = health_check_interval.tick() => {
                    self.check_dead_nodes().await;
                }

                // Periodic prune of the sync log
                _ = prune_interval_ticker.tick() => {
                    if self.config.prune_interval.is_some() {
                        self.prune_sync_log();
                    }
                }
            }
        }

        // Persist state on shutdown
        self.state.persist();
    }

    /// Sync with all connected peers
    async fn sync_with_peers(&self) {
        // Get peers from both SyncState (persisted) and ClusterManager (discovered)
        let mut peers = self.state.get_peers();
        debug!("sync_with_peers: found {} persisted peers", peers.len());

        // Also get peers from ClusterManager if available
        if let Some(ref manager) = self.cluster_manager {
            let cluster_members = manager.state().get_all_members();
            debug!(
                "sync_with_peers: ClusterManager has {} members",
                cluster_members.len()
            );
            for member in cluster_members {
                // Skip self
                if member.node.id == self.local_node_id {
                    continue;
                }
                // Check if we already have this peer
                let already_known = peers.iter().any(|p| p.node_id == member.node.id);
                if !already_known {
                    debug!(
                        "sync_with_peers: discovered peer {} at {}",
                        member.node.id, member.node.address
                    );
                    // Add to SyncState for future syncs
                    self.state.add_peer(
                        member.node.id.clone(),
                        member.node.address.clone(),     // sync address
                        member.node.api_address.clone(), // http address
                    );
                    // Add to local list for this sync cycle
                    peers.push(super::state::PeerInfo {
                        node_id: member.node.id,
                        sync_address: member.node.address,
                        http_address: member.node.api_address,
                        last_seen: std::time::Instant::now(),
                        is_connected: false,
                    });
                }
            }
        } else {
            debug!("sync_with_peers: no ClusterManager");
        }

        debug!("sync_with_peers: {} peers total", peers.len());

        for peer in peers {
            debug!(
                "sync_with_peers: syncing with {} at {}",
                peer.node_id, peer.sync_address
            );
            if !peer.is_connected {
                // Try to connect
                if self.pool.connect(&peer.sync_address).await.is_ok() {
                    self.state.set_peer_connected(&peer.node_id, true);
                }
            }

            if peer.is_connected || self.pool.connect(&peer.sync_address).await.is_ok() {
                self.state.set_peer_connected(&peer.node_id, true);

                // Mark ourselves as syncing in cluster state
                if let Some(ref mgr) = self.cluster_manager {
                    mgr.state().mark_status(
                        &self.local_node_id,
                        crate::cluster::state::NodeStatus::Syncing,
                    );
                }

                // Sync loop - keep fetching while there's more data
                // User requested no page limit to sync millions of documents
                let mut pages = 0;
                let mut has_more = true;

                while has_more {
                    match self.incremental_sync(&peer.sync_address).await {
                        Ok(more) => {
                            has_more = more;
                            pages += 1;
                        }
                        Err(e) => {
                            warn!("Sync with {} failed: {}", peer.node_id, e);
                            self.state.set_peer_connected(&peer.node_id, false);
                            self.pool.disconnect(&peer.sync_address).await;
                            has_more = false;
                        }
                    }
                }

                // Mark ourselves as active again
                if let Some(ref mgr) = self.cluster_manager {
                    mgr.state().mark_status(
                        &self.local_node_id,
                        crate::cluster::state::NodeStatus::Active,
                    );
                }

                if pages > 0 {
                    debug!(
                        "Synced {} batches from {} (finished, has_more={})",
                        pages, peer.node_id, has_more
                    );
                }
            } else {
                debug!("sync_with_peers: failed to connect to {}", peer.node_id);
            }
        }
    }

    /// Incremental sync with a peer
    /// Pulls entries FROM the peer that we haven't received yet
    /// Returns true if there are more entries to sync
    async fn incremental_sync(&self, peer_addr: &str) -> Result<bool, TransportError> {
        // Get the last sequence we received from this peer's perspective
        // For pull-based sync, we ask the peer: "give me entries after sequence X from YOUR log"
        // We track this using get_origin_sequence keyed by peer address
        let after_seq = self.state.get_origin_sequence(peer_addr);

        debug!("incremental_sync: {} after_seq={}", peer_addr, after_seq);

        // Request sync from peer
        let request = SyncMessage::IncrementalSyncRequest {
            from_node: self.local_node_id.clone(),
            after_sequence: after_seq,
            max_batch_bytes: self.config.max_batch_bytes,
        };

        self.pool.send(peer_addr, &request).await?;

        // Wait for response
        let response = self.pool.receive(peer_addr).await?;

        match response {
            SyncMessage::SyncBatch {
                entries,
                has_more,
                current_sequence,
                ..
            } => {
                debug!(
                    "incremental_sync: {} entries from {} (seq={}) has_more={}",
                    entries.len(),
                    peer_addr,
                    current_sequence,
                    has_more
                );

                // Group consecutive entries by (database, collection, operation) for batching
                let mut batch_start = 0;
                while batch_start < entries.len() {
                    let first = &entries[batch_start];

                    // Only batch data operations
                    if matches!(
                        first.operation,
                        Operation::Insert | Operation::Update | Operation::Delete
                    ) {
                        let mut batch_end = batch_start + 1;
                        while batch_end < entries.len() {
                            let next = &entries[batch_end];
                            if next.database == first.database
                                && next.collection == first.collection
                                && next.operation == first.operation
                            {
                                batch_end += 1;
                            } else {
                                break;
                            }
                        }

                        // Process batch
                        let batch = &entries[batch_start..batch_end];
                        self.apply_batch(batch).await?;
                        batch_start = batch_end;
                    } else {
                        // Single entry processing for schema changes
                        self.apply_entry(first).await?;
                        batch_start += 1;
                    }
                }

                // Only update origin_sequence if we ACTUALLY received entries.
                // We must NOT update to current_sequence if entries are empty, because that implies
                // the server hasn't persisted the data yet (race condition) or we'd skip data.
                if let Some(max_seq) = entries.iter().map(|e| e.sequence).max() {
                    if max_seq > after_seq {
                        self.state.update_origin_sequence(peer_addr, max_seq);
                    }
                }

                debug!("Applied {} entries from {}", entries.len(), peer_addr);

                // Calculate has_more based on what the server claims is the head vs what we have
                // If server has seq 100, and we are at 90 (either via max_seq or after_seq), we have more.
                // Note: current_sequence from server is the "head", entries max is what we just got.
                let latest_we_have = entries
                    .iter()
                    .map(|e| e.sequence)
                    .max()
                    .unwrap_or(after_seq);
                let actual_has_more = current_sequence > latest_we_have;

                Ok(actual_has_more)
            }
            _ => {
                warn!("Unexpected response from {}", peer_addr);
                Ok(false)
            }
        }
    }

    /// Apply a batch of sync entries to local storage
    async fn apply_batch(&self, entries: &[SyncEntry]) -> Result<(), TransportError> {
        if entries.is_empty() {
            return Ok(());
        }

        let first = &entries[0];
        let database = &first.database;
        let collection = &first.collection;
        let operation = first.operation;

        // Skip physical shard collections - sharded data is partitioned, NOT replicated cluster-wide
        // Physical shards have names like "users_s0", "users_s1" etc.
        let is_physical_shard = collection.contains("_s")
            && collection
                .chars()
                .last()
                .map(|c| c.is_ascii_digit())
                .unwrap_or(false);
        if is_physical_shard {
            debug!(
                "apply_batch: Skipping physical shard collection {} (partitioned, not replicated)",
                collection
            );
            return Ok(());
        }

        // Ensure database and collection exist for Write operations
        if matches!(operation, Operation::Insert | Operation::Update) {
            // Create database if it doesn't exist
            if self.storage.get_database(database).is_err() {
                let _ = self.storage.create_database(database.clone());
            }

            if let Ok(db) = self.storage.get_database(database) {
                if db.get_collection(collection).is_err() {
                    let _ = db.create_collection(collection.clone(), None);
                }
            }
        }

        match operation {
            Operation::Insert | Operation::Update => {
                if let Ok(db) = self.storage.get_database(database) {
                    if let Ok(coll) = db.get_collection(collection) {
                        let mut batch_data = Vec::with_capacity(entries.len());

                        for entry in entries {
                            // Check for duplicate
                            if self
                                .state
                                .is_duplicate(&entry.origin_node, entry.origin_sequence)
                            {
                                continue;
                            }

                            if let Some(ref data) = entry.document_data {
                                let doc: serde_json::Value =
                                    serde_json::from_slice(data).map_err(|e| {
                                        TransportError::DecodeError(format!(
                                            "Invalid document: {}",
                                            e
                                        ))
                                    })?;
                                batch_data.push((entry.document_key.clone(), doc));
                            }
                        }

                        if !batch_data.is_empty() {
                            // Replicated writes to _api_keys must also update the
                            // in-memory auth cache (the HTTP handlers that normally
                            // maintain it are bypassed on this path).
                            let api_key_docs: Vec<serde_json::Value> =
                                if database == "_system" && collection == "_api_keys" {
                                    batch_data.iter().map(|(_, doc)| doc.clone()).collect()
                                } else {
                                    Vec::new()
                                };

                            if let Err(e) = coll.upsert_batch(batch_data) {
                                warn!(
                                    "apply_batch: upsert failed for {}.{}: {}",
                                    database, collection, e
                                );
                            } else {
                                for doc in &api_key_docs {
                                    crate::server::auth::note_replicated_api_key_upsert(doc);
                                }
                            }
                        }
                    }
                }
            }
            Operation::Delete => {
                if let Ok(db) = self.storage.get_database(database) {
                    if let Ok(coll) = db.get_collection(collection) {
                        let mut keys_to_delete = Vec::with_capacity(entries.len());

                        for entry in entries {
                            // Check for duplicate
                            if self
                                .state
                                .is_duplicate(&entry.origin_node, entry.origin_sequence)
                            {
                                continue;
                            }
                            keys_to_delete.push(entry.document_key.clone());
                        }

                        if !keys_to_delete.is_empty() {
                            // Evict replicated _api_keys deletes from the in-memory
                            // auth cache so revoked keys stop authenticating here.
                            if database == "_system" && collection == "_api_keys" {
                                for key in &keys_to_delete {
                                    crate::server::auth::note_replicated_api_key_delete(key);
                                }
                            }
                            let _ = coll.delete_batch(keys_to_delete);
                        }
                    }
                }
            }
            _ => {
                // Other operations should be handled one by one via apply_entry
                // But if they came here, they are grouped, so we iterate
                for entry in entries {
                    self.apply_entry(entry).await?;
                }
            }
        }

        // Update origin sequence for all processed entries
        for entry in entries {
            self.state
                .update_origin_sequence(&entry.origin_node, entry.origin_sequence);
        }

        Ok(())
    }

    /// Apply a sync entry to local storage
    async fn apply_entry(&self, entry: &SyncEntry) -> Result<(), TransportError> {
        // Check for duplicate
        if self
            .state
            .is_duplicate(&entry.origin_node, entry.origin_sequence)
        {
            return Ok(());
        }

        // Skip physical shard collections - sharded data is partitioned, NOT replicated cluster-wide
        let is_physical_shard = entry.collection.contains("_s")
            && entry
                .collection
                .chars()
                .last()
                .map(|c| c.is_ascii_digit())
                .unwrap_or(false);
        if is_physical_shard {
            return Ok(());
        }

        // Apply based on operation type
        match entry.operation {
            Operation::Insert | Operation::Update => {
                if let Some(ref data) = entry.document_data {
                    let doc: serde_json::Value = serde_json::from_slice(data).map_err(|e| {
                        TransportError::DecodeError(format!("Invalid document: {}", e))
                    })?;

                    // Create database if it doesn't exist
                    if self.storage.get_database(&entry.database).is_err() {
                        let _ = self.storage.create_database(entry.database.clone());
                    }

                    if let Ok(db) = self.storage.get_database(&entry.database) {
                        // Create collection if it doesn't exist
                        if db.get_collection(&entry.collection).is_err() {
                            let _ = db.create_collection(entry.collection.clone(), None);
                        }

                        if let Ok(coll) = db.get_collection(&entry.collection) {
                            // Replicated _api_keys writes must also refresh the
                            // in-memory auth cache (handlers are bypassed here).
                            let api_key_doc =
                                if entry.database == "_system" && entry.collection == "_api_keys" {
                                    Some(doc.clone())
                                } else {
                                    None
                                };

                            if let Err(e) =
                                coll.upsert_batch(vec![(entry.document_key.clone(), doc)])
                            {
                                warn!(
                                    "apply_entry: upsert failed for {}: {}",
                                    entry.document_key, e
                                );
                            } else if let Some(ref doc) = api_key_doc {
                                crate::server::auth::note_replicated_api_key_upsert(doc);
                            }
                        }
                    }
                }
            }
            Operation::Delete => {
                if let Ok(db) = self.storage.get_database(&entry.database) {
                    if let Ok(coll) = db.get_collection(&entry.collection) {
                        let _ = coll.delete(&entry.document_key);
                    }
                }
                // Evict replicated _api_keys deletes from the in-memory auth
                // cache so revoked keys stop authenticating here.
                if entry.database == "_system" && entry.collection == "_api_keys" {
                    crate::server::auth::note_replicated_api_key_delete(&entry.document_key);
                }
            }
            Operation::CreateDatabase => {
                let _ = self.storage.create_database(entry.database.clone());
            }
            Operation::DeleteDatabase => {
                let _ = self.storage.delete_database(&entry.database);
            }
            Operation::CreateCollection => {
                if let Ok(db) = self.storage.get_database(&entry.database) {
                    // Parse metadata from entry.document_data
                    let metadata: Option<serde_json::Value> = entry
                        .document_data
                        .as_ref()
                        .and_then(|d| serde_json::from_slice(d).ok());

                    // Extract collection type from metadata
                    let collection_type = metadata
                        .as_ref()
                        .and_then(|m| m.get("type"))
                        .and_then(|t| t.as_str())
                        .map(|s| s.to_string());

                    // Create the collection with type
                    if let Ok(()) = db.create_collection(entry.collection.clone(), collection_type)
                    {
                        // Apply shard configuration if present
                        if let Some(ref meta) = metadata {
                            if let Some(shard_config_obj) = meta.get("shardConfig") {
                                if !shard_config_obj.is_null() {
                                    if let Ok(coll) = db.get_collection(&entry.collection) {
                                        // Parse shard config (CollectionShardConfig uses u16)
                                        let num_shards = shard_config_obj
                                            .get("num_shards")
                                            .and_then(|v| v.as_u64())
                                            .unwrap_or(1)
                                            as u16;
                                        let shard_key = shard_config_obj
                                            .get("shard_key")
                                            .and_then(|v| v.as_str())
                                            .unwrap_or("_key")
                                            .to_string();
                                        let replication_factor = shard_config_obj
                                            .get("replication_factor")
                                            .and_then(|v| v.as_u64())
                                            .unwrap_or(1)
                                            as u16;

                                        let shard_config =
                                            crate::sharding::coordinator::CollectionShardConfig {
                                                num_shards,
                                                shard_key,
                                                replication_factor,
                                            };

                                        let _ = coll.set_shard_config(&shard_config);
                                        debug!(
                                            "Replicated collection {} with shard config: {:?}",
                                            entry.collection, shard_config
                                        );
                                    }
                                }
                            }
                        }
                    }
                }
            }
            Operation::DeleteCollection => {
                if let Ok(db) = self.storage.get_database(&entry.database) {
                    let _ = db.delete_collection(&entry.collection);
                }
            }
            Operation::TruncateCollection => {
                if let Ok(db) = self.storage.get_database(&entry.database) {
                    if let Ok(coll) = db.get_collection(&entry.collection) {
                        // Check if sharded and truncate physical shards first
                        if let Some(shard_config) = coll.get_shard_config() {
                            if shard_config.num_shards > 0 {
                                for shard_id in 0..shard_config.num_shards {
                                    let physical_name =
                                        format!("{}_s{}", entry.collection, shard_id);
                                    if let Ok(shard_coll) = db.get_collection(&physical_name) {
                                        let _ = shard_coll.truncate();
                                    }
                                }
                            }
                        }
                        // Truncate the logical collection
                        let _ = coll.truncate();
                    }
                }
            }
            Operation::PutBlobChunk | Operation::DeleteBlob => {
                // Intentionally a no-op: blob chunks do not travel through the
                // replication log. They are pushed directly at upload time
                // (`sync::blob_replication`) and any that were missed are
                // repaired by `sharding::BlobRebalanceWorker`, because chunks
                // are far too large to buffer in the log.
            }
            Operation::CreateIndex => {
                // The entry names either the logical collection or a physical
                // shard; a node applies whichever of those it actually holds
                // and ignores the rest.
                if let Some(ref data) = entry.document_data {
                    let spec: crate::storage::IndexSpec =
                        serde_json::from_slice(data).map_err(|e| {
                            TransportError::DecodeError(format!("Invalid index spec: {}", e))
                        })?;
                    if let Ok(db) = self.storage.get_database(&entry.database) {
                        if let Ok(coll) = db.get_collection(&entry.collection) {
                            if let Err(e) = coll.apply_index_spec(&spec) {
                                warn!(
                                    "apply_entry: failed to create index '{}' on {}.{}: {}",
                                    spec.name(),
                                    entry.database,
                                    entry.collection,
                                    e
                                );
                            }
                        }
                    }
                }
            }
            Operation::DropIndex => {
                if let Some(ref data) = entry.document_data {
                    let index_ref: crate::storage::IndexRef = serde_json::from_slice(data)
                        .map_err(|e| {
                            TransportError::DecodeError(format!("Invalid index ref: {}", e))
                        })?;
                    if let Ok(db) = self.storage.get_database(&entry.database) {
                        if let Ok(coll) = db.get_collection(&entry.collection) {
                            if let Err(e) = coll.apply_index_drop(index_ref.kind, &index_ref.name) {
                                warn!(
                                    "apply_entry: failed to drop index '{}' on {}.{}: {}",
                                    index_ref.name, entry.database, entry.collection, e
                                );
                            }
                        }
                    }
                }
            }
            Operation::ColumnarInsert => {
                if let Some(ref data) = entry.document_data {
                    let row: serde_json::Value = serde_json::from_slice(data).map_err(|e| {
                        TransportError::DecodeError(format!("Invalid columnar row: {}", e))
                    })?;

                    // Create database if it doesn't exist
                    if self.storage.get_database(&entry.database).is_err() {
                        let _ = self.storage.create_database(entry.database.clone());
                    }

                    if let Ok(db) = self.storage.get_database(&entry.database) {
                        // Load columnar collection and insert with specific UUID
                        match crate::storage::columnar::ColumnarCollection::load(
                            entry.collection.clone(),
                            &entry.database,
                            db.db_arc(),
                        ) {
                            Ok(col) => {
                                // Insert with specific UUID (idempotent)
                                if let Err(e) = col.insert_row_with_id(&entry.document_key, row) {
                                    warn!(
                                        "apply_entry: columnar insert failed for {}: {}",
                                        entry.document_key, e
                                    );
                                }
                            }
                            Err(e) => {
                                warn!(
                                    "apply_entry: failed to load columnar collection {}: {}",
                                    entry.collection, e
                                );
                            }
                        }
                    }
                }
            }
            Operation::ColumnarDelete => {
                if let Ok(db) = self.storage.get_database(&entry.database) {
                    match crate::storage::columnar::ColumnarCollection::load(
                        entry.collection.clone(),
                        &entry.database,
                        db.db_arc(),
                    ) {
                        Ok(col) => {
                            if let Err(e) = col.delete_row(&entry.document_key) {
                                warn!(
                                    "apply_entry: columnar delete failed for {}: {}",
                                    entry.document_key, e
                                );
                            }
                        }
                        Err(e) => {
                            warn!(
                                "apply_entry: failed to load columnar collection {}: {}",
                                entry.collection, e
                            );
                        }
                    }
                }
            }
            Operation::ColumnarCreateCollection => {
                // Create database if it doesn't exist
                if self.storage.get_database(&entry.database).is_err() {
                    let _ = self.storage.create_database(entry.database.clone());
                }

                if let Ok(db) = self.storage.get_database(&entry.database) {
                    // Parse column definitions from entry.document_data
                    if let Some(ref data) = entry.document_data {
                        if let Ok(columns) =
                            serde_json::from_slice::<Vec<crate::storage::columnar::ColumnDef>>(data)
                        {
                            let _ = crate::storage::columnar::ColumnarCollection::new(
                                entry.collection.clone(),
                                &entry.database,
                                db.db_arc(),
                                columns,
                                crate::storage::columnar::CompressionType::Lz4,
                            );
                        }
                    }
                }
            }
            Operation::ColumnarDropCollection => {
                if let Ok(db) = self.storage.get_database(&entry.database) {
                    if let Ok(col) = crate::storage::columnar::ColumnarCollection::load(
                        entry.collection.clone(),
                        &entry.database,
                        db.db_arc(),
                    ) {
                        let _ = col.drop();
                    }
                }
            }
            Operation::ColumnarTruncate => {
                if let Ok(db) = self.storage.get_database(&entry.database) {
                    if let Ok(col) = crate::storage::columnar::ColumnarCollection::load(
                        entry.collection.clone(),
                        &entry.database,
                        db.db_arc(),
                    ) {
                        if let Err(e) = col.truncate() {
                            warn!(
                                "apply_entry: columnar truncate failed for {}: {}",
                                entry.collection, e
                            );
                        }
                    }
                }
            }
        }

        // Update origin sequence
        self.state
            .update_origin_sequence(&entry.origin_node, entry.origin_sequence);

        Ok(())
    }

    /// Request full sync from a peer (for new nodes)
    async fn request_full_sync(&self, peer_addr: &str) -> Result<(), TransportError> {
        self.pool.connect(peer_addr).await?;

        let request = SyncMessage::FullSyncRequest {
            from_node: self.local_node_id.clone(),
        };

        self.pool.send(peer_addr, &request).await?;

        // Process full sync messages
        loop {
            let msg = self.pool.receive(peer_addr).await?;

            match msg {
                SyncMessage::FullSyncStart {
                    total_databases,
                    total_documents,
                    ..
                } => {
                    info!(
                        "Starting full sync: {} databases, {} documents",
                        total_databases, total_documents
                    );
                }
                SyncMessage::FullSyncDatabase { name } => {
                    let _ = self.storage.create_database(name.clone());
                }
                SyncMessage::FullSyncCollection {
                    database,
                    name,
                    collection_type,
                    ..
                } => {
                    if let Ok(db) = self.storage.get_database(&database) {
                        let _ = db.create_collection(name.clone(), collection_type.clone());
                        // The collection may already exist from an earlier sync
                        // with the type unset; make the type stick either way.
                        if let (Some(ref ctype), Ok(coll)) =
                            (collection_type, db.get_collection(&name))
                        {
                            if coll.get_type() != *ctype {
                                let _ = coll.set_type(ctype);
                            }
                        }
                    }
                }
                SyncMessage::FullSyncDocuments {
                    database,
                    collection,
                    data,
                    compressed,
                    doc_count,
                } => {
                    let docs_data = if compressed {
                        lz4_flex::decompress_size_prepended(&data).map_err(|e| {
                            TransportError::DecodeError(format!("Decompression failed: {}", e))
                        })?
                    } else {
                        data
                    };

                    let docs = super::protocol::decode_documents(&docs_data)
                        .map_err(TransportError::DecodeError)?;

                    for doc in docs {
                        if let Some(key) = doc.get("_key").and_then(|k| k.as_str()) {
                            if let Ok(db) = self.storage.get_database(&database) {
                                if let Ok(coll) = db.get_collection(&collection) {
                                    let _ = coll.upsert_batch(vec![(key.to_string(), doc)]);
                                }
                            }
                        }
                    }
                    debug!(
                        "Synced {} documents to {}.{}",
                        doc_count, database, collection
                    );
                }
                SyncMessage::FullSyncComplete { final_sequence } => {
                    info!("Full sync complete, final sequence: {}", final_sequence);
                    break;
                }
                _ => {
                    warn!("Unexpected message during full sync");
                }
            }
        }

        Ok(())
    }

    /// Periodic prune of the sync log up to the highest sequence safely
    /// confirmed-received by every known peer.
    ///
    /// Decision rules:
    /// - If peers exist but none have ever pulled from us, we cannot know
    ///   what they need — do nothing (defensive).
    /// - If peers exist and have pulled, prune up to `min(sent_sequences)`,
    ///   capped at `current_sequence - retain_buffer`.
    /// - If no peers are configured at all, the log is dead weight — prune
    ///   everything older than `current_sequence - retain_buffer`.
    fn prune_sync_log(&self) {
        let current = self.sync_log.current_sequence();
        let retain = self.config.prune_retain_buffer;
        let upper_bound = current.saturating_sub(retain);
        if upper_bound == 0 {
            return;
        }

        let configured_peers = self.state.get_peers();
        let safe_seq = match self.state.min_sent_sequence() {
            Some(min_sent) => min_sent.min(upper_bound),
            None if configured_peers.is_empty() => upper_bound,
            None => {
                debug!("prune_sync_log: peers configured but none have pulled yet; skipping");
                return;
            }
        };

        // A peer that is permanently offline pins `min_sent_sequence` and the
        // log grows without bound until the disk fills. Pruning past an
        // unacked peer would silently lose its data, so don't — but make the
        // situation loud so an operator removes the dead peer (or runs the
        // manual prune endpoint) before the disk fills.
        const LAG_WARN_THRESHOLD: u64 = 1_000_000;
        let lag = upper_bound.saturating_sub(safe_seq);
        if lag > LAG_WARN_THRESHOLD {
            warn!(
                "Sync log retains {} unpruned entries because at least one peer \
                 has not acknowledged past sequence {} (head {}). If that peer is \
                 permanently gone, remove it from the cluster or prune manually — \
                 the log will otherwise grow until the disk fills.",
                lag, safe_seq, current
            );
        }

        // prune_before(N) deletes entries with sequence < N. We want to KEEP
        // up to and including safe_seq, so we pass safe_seq + 1.
        let before = safe_seq.saturating_add(1);
        match self.sync_log.prune_before(before) {
            Ok(0) => {}
            Ok(n) => {
                info!(
                    "Sync log auto-prune: removed {} entries (kept >= seq {})",
                    n, before
                );
            }
            Err(e) => {
                warn!("Sync log auto-prune failed: {}", e);
            }
        }
    }

    /// Send heartbeats to all peers
    async fn send_heartbeats(&mut self) {
        let peers = self.state.get_peers();
        if peers.is_empty() {
            return;
        }

        let stats = self.collect_local_stats();
        let heartbeat = SyncMessage::Heartbeat {
            node_id: self.local_node_id.clone(),
            sequence: self.state.current_sequence(),
            stats,
        };

        for peer in self.state.get_peers() {
            if peer.is_connected {
                if let Err(e) = self.pool.send(&peer.sync_address, &heartbeat).await {
                    debug!("Failed to send heartbeat to {}: {}", peer.node_id, e);
                }
            }
        }
    }

    /// Check for dead nodes and remove them
    async fn check_dead_nodes(&self) {
        let dead = self.state.dead_nodes(self.config.dead_node_timeout);

        if dead.is_empty() {
            return;
        }

        for node_id in &dead {
            warn!("Node {} is dead, removing from cluster", node_id);
            self.state.remove_peer(node_id);

            // Also update cluster manager if present
            if let Some(ref mgr) = self.cluster_manager {
                mgr.state().remove_member(node_id);
            }
        }

        if !self.state.get_peers().is_empty() {
            self.state.persist();
        }

        // Trigger shard rebalancing if we have a coordinator
        if let Some(ref coordinator) = self.shard_coordinator {
            info!("Triggering automatic shard rebalance after node death");
            let coordinator = coordinator.clone();
            tokio::spawn(async move {
                if let Err(e) = coordinator.rebalance().await {
                    error!("Failed to rebalance shards after node death: {}", e);
                }
            });
        }
    }

    /// Collect local node statistics
    fn collect_local_stats(&mut self) -> NodeStats {
        self.system.refresh_cpu_usage();
        self.system.refresh_memory();

        let cpu_usage = self
            .system
            .cpus()
            .first()
            .map(|c| c.cpu_usage())
            .unwrap_or(0.0);
        let memory_used = self.system.used_memory();
        let disk_used = 0;

        // Count documents and collections
        let mut document_count = 0u64;
        let mut collections_count = 0u32;

        let dbs = self.storage.list_databases();
        for db_name in dbs {
            if db_name.starts_with('_') {
                continue; // Skip system databases
            }
            if let Ok(db) = self.storage.get_database(&db_name) {
                let colls = db.list_collections();
                collections_count = collections_count.saturating_add(colls.len() as u32);
                for coll_name in colls {
                    if let Ok(coll) = db.system_collection(&coll_name) {
                        document_count = document_count.saturating_add(coll.count() as u64);
                    }
                }
            }
        }

        NodeStats {
            cpu_usage,
            memory_used,
            disk_used,
            document_count,
            collections_count,
        }
    }

    /// Handle incoming connection
    pub async fn handle_connection(
        mut stream: super::transport::SyncStream,
        _addr: String,
        _pool: Arc<ConnectionPool>,
        state: Arc<SyncState>,
        storage: Arc<StorageEngine>,
        sync_log: Arc<super::log::SyncLog>,
        cluster_manager: Option<Arc<crate::cluster::manager::ClusterManager>>,
    ) -> Result<(), TransportError> {
        use crate::cluster::HybridLogicalClock;
        let hlc = HybridLogicalClock::now(sync_log.node_id());

        loop {
            // Read message header. Waiting for the FIRST byte may block
            // indefinitely (persistent connections legitimately idle between
            // messages), but once a message has started, the rest of the
            // header and the payload must arrive promptly — otherwise a peer
            // sending one byte and stalling holds the connection forever
            // (slowloris on the replication port).
            let mut header = [0u8; 5];
            if tokio::io::AsyncReadExt::read_exact(&mut stream, &mut header[..1])
                .await
                .is_err()
            {
                break;
            }
            match tokio::time::timeout(
                std::time::Duration::from_secs(10),
                tokio::io::AsyncReadExt::read_exact(&mut stream, &mut header[1..]),
            )
            .await
            {
                Ok(Ok(_)) => {}
                _ => break,
            }

            let compressed = header[0] == 1;
            let len = u32::from_be_bytes([header[1], header[2], header[3], header[4]]);

            if len > 10 * 1024 * 1024 {
                break;
            }

            let mut data = vec![0u8; len as usize];
            match tokio::time::timeout(
                std::time::Duration::from_secs(60),
                tokio::io::AsyncReadExt::read_exact(&mut stream, &mut data),
            )
            .await
            {
                Ok(Ok(_)) => {}
                _ => break,
            }

            let payload = if compressed {
                match lz4_flex::decompress_size_prepended(&data) {
                    Ok(p) => p,
                    Err(e) => {
                        tracing::error!("Decompression failed: {}, closing connection", e);
                        break;
                    }
                }
            } else {
                data
            };

            let msg: SyncMessage = match bincode::deserialize(&payload) {
                Ok(m) => m,
                Err(_) => break,
            };

            // Handle message
            match msg {
                SyncMessage::Heartbeat {
                    node_id,
                    sequence,
                    stats,
                } => {
                    // Update sync state heartbeat
                    state.update_heartbeat(&node_id, stats);

                    // Also update cluster state heartbeat so admin UI shows nodes as connected
                    if let Some(ref cm) = cluster_manager {
                        cm.state().update_heartbeat(&node_id, sequence, None);
                    }
                }
                SyncMessage::IncrementalSyncRequest {
                    from_node,
                    after_sequence,
                    max_batch_bytes,
                } => {
                    // Record what this peer has confirmed receipt of: by asking
                    // for entries strictly after `after_sequence`, they prove
                    // they already hold every entry with sequence <= that
                    // value. This is the high-watermark used by prune logic.
                    if !from_node.is_empty() {
                        state.update_sent_sequence(&from_node, after_sequence);
                    }

                    // Fetch entries from log
                    let limit = (max_batch_bytes / 1024).max(100) as usize; // Rough estimate
                    let log_entries = sync_log.get_entries_after(after_sequence, limit);
                    let current_seq = sync_log.current_sequence();

                    debug!(
                        "IncrementalSyncRequest: from={} after_seq={}, current_seq={}, found {} entries",
                        from_node,
                        after_sequence,
                        current_seq,
                        log_entries.len()
                    );

                    // Convert to SyncEntry
                    let entries: Vec<SyncEntry> =
                        log_entries.iter().map(|e| e.to_sync_entry(&hlc)).collect();

                    let has_more = !entries.is_empty()
                        && entries.last().map(|e| e.sequence).unwrap_or(0) < current_seq;

                    let response = SyncMessage::SyncBatch {
                        entries,
                        has_more,
                        current_sequence: current_seq,
                        compressed: false,
                    };

                    // Use proper message framing that the client expects
                    if let Err(e) =
                        super::transport::ConnectionPool::write_message(&mut stream, &response)
                            .await
                    {
                        warn!("Failed to send SyncBatch response: {}", e);
                        break;
                    }
                }
                SyncMessage::FullSyncRequest { from_node } => {
                    info!("Full sync request from {}", from_node);

                    // Enumerate databases and collections
                    let databases = storage.list_databases();
                    let mut total_collections = 0u32;
                    let mut total_documents = 0u64;

                    for db_name in &databases {
                        if let Ok(db) = storage.get_database(db_name) {
                            let colls = db.list_collections();
                            total_collections += colls.len() as u32;
                            for coll_name in &colls {
                                if let Ok(coll) = db.system_collection(coll_name) {
                                    total_documents += coll.count() as u64;
                                }
                            }
                        }
                    }

                    // Send start message
                    let start = SyncMessage::FullSyncStart {
                        total_databases: databases.len() as u32,
                        total_collections,
                        total_documents,
                    };
                    let _ = tokio::io::AsyncWriteExt::write_all(&mut stream, &start.encode()).await;

                    // Send each database
                    for db_name in &databases {
                        let db_msg = SyncMessage::FullSyncDatabase {
                            name: db_name.clone(),
                        };
                        let _ = tokio::io::AsyncWriteExt::write_all(&mut stream, &db_msg.encode())
                            .await;

                        if let Ok(db) = storage.get_database(db_name) {
                            let colls = db.list_collections();
                            for coll_name in colls {
                                // Send collection
                                // Carry the collection's real type and shard
                                // config; recreating everything as an untyped
                                // document collection silently downgraded blob
                                // and timeseries collections on the receiver.
                                let (collection_type, shard_config) =
                                    match db.system_collection(&coll_name) {
                                        Ok(c) => (
                                            Some(c.get_type().to_string()),
                                            c.get_shard_config().map(|cfg| ShardConfig {
                                                num_shards: cfg.num_shards,
                                                shard_key: cfg.shard_key.clone(),
                                                replication_factor: cfg.replication_factor,
                                            }),
                                        ),
                                        Err(_) => (None, None),
                                    };
                                let coll_msg = SyncMessage::FullSyncCollection {
                                    database: db_name.clone(),
                                    name: coll_name.clone(),
                                    shard_config,
                                    collection_type,
                                };
                                let _ = tokio::io::AsyncWriteExt::write_all(
                                    &mut stream,
                                    &coll_msg.encode(),
                                )
                                .await;

                                // Send documents in batches
                                if let Ok(coll) = db.get_collection(&coll_name) {
                                    let mut batch = Vec::new();
                                    let mut batch_count = 0u32;

                                    for doc in coll.scan(None) {
                                        batch.push(doc.to_value());
                                        batch_count += 1;

                                        if batch.len() >= 1000 {
                                            // Send batch
                                            // Not `unwrap_or_default()`: that
                                            // turned an encoding failure into an
                                            // empty batch, so a sync reported
                                            // success having sent nothing.
                                            let data =
                                                match super::protocol::encode_documents(&batch) {
                                                    Ok(data) => data,
                                                    Err(e) => {
                                                        return Err(TransportError::DecodeError(
                                                            format!("full sync aborted: {e}"),
                                                        ));
                                                    }
                                                };
                                            let compress = data.len() > 10 * 1024;
                                            let final_data = if compress {
                                                lz4_flex::compress_prepend_size(&data)
                                            } else {
                                                data
                                            };

                                            let doc_msg = SyncMessage::FullSyncDocuments {
                                                database: db_name.clone(),
                                                collection: coll_name.clone(),
                                                data: final_data,
                                                compressed: compress,
                                                doc_count: batch_count,
                                            };
                                            let _ = tokio::io::AsyncWriteExt::write_all(
                                                &mut stream,
                                                &doc_msg.encode(),
                                            )
                                            .await;

                                            batch.clear();
                                            batch_count = 0;
                                        }
                                    }

                                    // Send remaining
                                    if !batch.is_empty() {
                                        let data = match super::protocol::encode_documents(&batch) {
                                            Ok(data) => data,
                                            Err(e) => {
                                                return Err(TransportError::DecodeError(format!(
                                                    "full sync aborted: {e}"
                                                )));
                                            }
                                        };
                                        let compress = data.len() > 10 * 1024;
                                        let final_data = if compress {
                                            lz4_flex::compress_prepend_size(&data)
                                        } else {
                                            data
                                        };

                                        let doc_msg = SyncMessage::FullSyncDocuments {
                                            database: db_name.clone(),
                                            collection: coll_name.clone(),
                                            data: final_data,
                                            compressed: compress,
                                            doc_count: batch_count,
                                        };
                                        let _ = tokio::io::AsyncWriteExt::write_all(
                                            &mut stream,
                                            &doc_msg.encode(),
                                        )
                                        .await;
                                    }
                                }
                            }
                        }
                    }

                    // Send complete
                    let complete = SyncMessage::FullSyncComplete {
                        final_sequence: sync_log.current_sequence(),
                    };
                    let _ =
                        tokio::io::AsyncWriteExt::write_all(&mut stream, &complete.encode()).await;
                }
                _ => {}
            }
        }

        Ok(())
    }
}

/// Create a command channel for the sync worker
pub fn create_command_channel() -> (mpsc::Sender<SyncCommand>, mpsc::Receiver<SyncCommand>) {
    mpsc::channel(100)
}