allsource-core 0.19.1

// Copyright 2024-2025 AllSource Team
// Licensed under the Business Source License 1.1 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     See LICENSE-BSL in the repository root
//
// Change Date: 2029-03-01
// Change License: Apache License, Version 2.0

//! WAL Receiver — connects to the leader and replays WAL entries on a follower.
//!
//! On startup (follower mode), connects to the leader's replication port, sends
//! a `Subscribe` message with the last local WAL offset, receives WAL entries,
//! validates CRC32 checksums, writes to local WAL, replays into the EventStore,
//! and sends ACKs back to the leader. Auto-reconnects with exponential backoff.
//!
//! ## Catch-up protocol
//!
//! When the follower is too far behind for WAL-only catch-up, the leader sends
//! a Parquet snapshot:
//! 1. `SnapshotStart` — list of Parquet files to expect
//! 2. `SnapshotChunk` — base64-encoded binary chunks for each file
//! 3. `SnapshotEnd` — WAL offset to resume from after loading snapshot
//!
//! The follower writes received Parquet files to its local storage directory,
//! then loads them into the EventStore via `ingest_replicated()`.

use anyhow::Context as _;

use crate::{
    infrastructure::{
        observability::metrics::MetricsRegistry,
        persistence::{
            storage::ParquetStorage,
            wal::{WALConfig, WALEntry, WriteAheadLog},
        },
        replication::protocol::{FollowerMessage, LeaderMessage},
    },
    store::EventStore,
};
use std::{
    collections::HashMap,
    sync::{
        Arc,
        atomic::{AtomicBool, AtomicU64, Ordering},
    },
    time::Duration,
};
use tokio::{
    io::{AsyncBufReadExt, AsyncWriteExt, BufReader},
    net::TcpStream,
};

/// Follower-side replication status exposed via the health endpoint.
#[derive(Debug, Clone, serde::Serialize)]
pub struct FollowerReplicationStatus {
    /// Whether the receiver is currently connected to the leader.
    pub connected: bool,
    /// Address of the leader replication port.
    pub leader: String,
    /// Estimated replication lag in milliseconds (offset-based proxy).
    pub replication_lag_ms: u64,
    /// Last WAL offset successfully replayed.
    pub last_replayed_offset: u64,
    /// Leader's current offset (from CaughtUp or latest entry).
    pub leader_offset: u64,
    /// Total entries replayed since this follower started.
    pub total_replayed: u64,
    /// Number of entries skipped due to CRC32 validation failure.
    pub corrupted_skipped: u64,
    /// Number of reconnection attempts since startup.
    pub reconnect_count: u64,
    /// Number of snapshot catch-ups completed since startup.
    pub snapshots_received: u64,
}

/// WAL Receiver manages the follower's connection to the leader.
pub struct WalReceiver {
    /// Leader's replication address (host:port). Wrapped in RwLock for repointing.
    leader_addr: Arc<tokio::sync::RwLock<String>>,
    /// Local WAL for crash recovery on the follower.
    local_wal: Arc<WriteAheadLog>,
    /// The event store to replay entries into.
    store: Arc<EventStore>,
    /// Directory for storing received Parquet snapshot files.
    snapshot_dir: std::path::PathBuf,
    /// Whether currently connected.
    connected: Arc<AtomicBool>,
    /// Last replayed WAL offset.
    last_replayed_offset: Arc<AtomicU64>,
    /// Leader's current offset.
    leader_offset: Arc<AtomicU64>,
    /// Total entries replayed.
    total_replayed: Arc<AtomicU64>,
    /// Corrupted entries skipped.
    corrupted_skipped: Arc<AtomicU64>,
    /// Reconnection attempts.
    reconnect_count: Arc<AtomicU64>,
    /// Snapshot catch-ups completed.
    snapshots_received: Arc<AtomicU64>,
    /// Prometheus metrics registry for replication counters/gauges.
    metrics: Option<Arc<MetricsRegistry>>,
    /// Shutdown flag — when set, the receiver stops reconnecting.
    shutdown: Arc<AtomicBool>,
    /// Notify channel to wake the receiver from backoff sleep on repoint/shutdown.
    wake: Arc<tokio::sync::Notify>,
}

impl WalReceiver {
    /// Create a new WAL receiver.
    ///
    /// `leader_addr` is the host:port of the leader's replication port (e.g. "core-leader:3910").
    /// `wal_dir` is the directory for the follower's local WAL files.
    pub fn new(
        leader_addr: String,
        wal_dir: impl Into<std::path::PathBuf>,
        store: Arc<EventStore>,
    ) -> anyhow::Result<Self> {
        let wal_dir = wal_dir.into();
        let wal_config = WALConfig {
            max_file_size: 64 * 1024 * 1024,
            sync_on_write: true,
            max_wal_files: 10,
            compress: false,
            ..WALConfig::default()
        };
        let local_wal = Arc::new(WriteAheadLog::new(&wal_dir, wal_config)?);

        // Recover last offset from local WAL (the max sequence from any recovered events)
        let last_offset = local_wal.current_sequence();

        // Snapshot directory is a sibling of the WAL directory
        let snapshot_dir = wal_dir
            .parent()
            .unwrap_or(&wal_dir)
            .join("follower-snapshots");

        Ok(Self {
            leader_addr: Arc::new(tokio::sync::RwLock::new(leader_addr)),
            local_wal,
            store,
            snapshot_dir,
            connected: Arc::new(AtomicBool::new(false)),
            last_replayed_offset: Arc::new(AtomicU64::new(last_offset)),
            leader_offset: Arc::new(AtomicU64::new(0)),
            total_replayed: Arc::new(AtomicU64::new(0)),
            corrupted_skipped: Arc::new(AtomicU64::new(0)),
            reconnect_count: Arc::new(AtomicU64::new(0)),
            snapshots_received: Arc::new(AtomicU64::new(0)),
            metrics: None,
            shutdown: Arc::new(AtomicBool::new(false)),
            wake: Arc::new(tokio::sync::Notify::new()),
        })
    }

    /// Set the Prometheus metrics registry for replication metrics.
    pub fn set_metrics(&mut self, metrics: Arc<MetricsRegistry>) {
        self.metrics = Some(metrics);
    }

    /// Get the current follower replication status for health reporting.
    pub fn status(&self) -> FollowerReplicationStatus {
        let last_replayed = self.last_replayed_offset.load(Ordering::Relaxed);
        let leader_off = self.leader_offset.load(Ordering::Relaxed);
        let lag = leader_off.saturating_sub(last_replayed);

        // Try to read leader_addr without blocking; fall back to "unknown" if locked
        let leader = self
            .leader_addr
            .try_read()
            .map_or_else(|_| "unknown".to_string(), |g| g.clone());

        FollowerReplicationStatus {
            connected: self.connected.load(Ordering::Relaxed),
            leader,
            replication_lag_ms: lag,
            last_replayed_offset: last_replayed,
            leader_offset: leader_off,
            total_replayed: self.total_replayed.load(Ordering::Relaxed),
            corrupted_skipped: self.corrupted_skipped.load(Ordering::Relaxed),
            reconnect_count: self.reconnect_count.load(Ordering::Relaxed),
            snapshots_received: self.snapshots_received.load(Ordering::Relaxed),
        }
    }

    /// Signal the receiver to stop reconnecting and shut down.
    ///
    /// Called during follower → leader promotion.
    pub fn shutdown(&self) {
        self.shutdown.store(true, Ordering::Relaxed);
        self.wake.notify_waiters();
    }

    /// Change the leader address and force a reconnect.
    ///
    /// Called by the sentinel via POST /internal/repoint to redirect
    /// this follower to a newly promoted leader.
    pub fn repoint(&self, new_leader: &str) {
        // Use try_write to avoid deadlocks in sync context
        if let Ok(mut guard) = self.leader_addr.try_write() {
            *guard = new_leader.to_string();
        } else {
            tracing::warn!(
                "REPOINT: Could not acquire write lock on leader_addr, will retry on next reconnect"
            );
        }
        // Wake the receiver from its backoff sleep so it reconnects immediately
        self.wake.notify_waiters();
    }

    /// Run the receiver loop with auto-reconnect. This runs until shutdown is signalled.
    ///
    /// Exponential backoff: 1s initial, doubles each attempt, capped at 30s.
    #[cfg_attr(feature = "hotpath", hotpath::measure)]
    pub async fn run(self: Arc<Self>) {
        let mut backoff = Duration::from_secs(1);
        let max_backoff = Duration::from_secs(30);

        loop {
            if self.shutdown.load(Ordering::Relaxed) {
                tracing::info!("WAL receiver shutdown requested — stopping");
                break;
            }

            let leader_addr = self.leader_addr.read().await.clone();

            tracing::info!(
                "Connecting to leader at {} (last_offset={})",
                leader_addr,
                self.last_replayed_offset.load(Ordering::Relaxed),
            );

            match self.connect_and_stream().await {
                Ok(()) => {
                    tracing::info!("Leader connection closed normally");
                }
                Err(e) => {
                    tracing::warn!("Leader connection error: {}", e);
                }
            }

            if self.shutdown.load(Ordering::Relaxed) {
                tracing::info!("WAL receiver shutdown requested — stopping");
                break;
            }

            self.connected.store(false, Ordering::Relaxed);
            self.reconnect_count.fetch_add(1, Ordering::Relaxed);
            if let Some(ref m) = self.metrics {
                m.replication_connected.set(0);
                m.replication_reconnects_total.inc();
            }

            tracing::info!(
                "Reconnecting to leader in {:?} (attempt {})",
                backoff,
                self.reconnect_count.load(Ordering::Relaxed),
            );

            // Sleep with early wake on repoint/shutdown
            tokio::select! {
                () = tokio::time::sleep(backoff) => {}
                () = self.wake.notified() => {
                    tracing::info!("WAL receiver woken early (repoint or shutdown)");
                    // Reset backoff on repoint so we reconnect quickly
                    backoff = Duration::from_secs(1);
                }
            }

            // Exponential backoff with cap
            backoff = (backoff * 2).min(max_backoff);
        }
    }

    /// Connect to the leader and stream WAL entries until disconnect.
    async fn connect_and_stream(&self) -> anyhow::Result<()> {
        let leader_addr = self.leader_addr.read().await.clone();
        let stream = TcpStream::connect(&leader_addr)
            .await
            .context(format!("TCP connect to leader at {leader_addr}"))?;
        let peer = stream.peer_addr()?;
        tracing::info!("Connected to leader at {}", peer);

        self.connected.store(true, Ordering::Relaxed);
        if let Some(ref m) = self.metrics {
            m.replication_connected.set(1);
        }

        let (reader, mut writer) = stream.into_split();
        let mut reader = BufReader::new(reader);

        // Step 1: Send Subscribe with our last known offset.
        let last_offset = self.last_replayed_offset.load(Ordering::Relaxed);
        let subscribe = FollowerMessage::Subscribe { last_offset };
        let mut json = serde_json::to_string(&subscribe)?;
        json.push('\n');
        writer
            .write_all(json.as_bytes())
            .await
            .context("sending Subscribe message to leader")?;
        writer
            .flush()
            .await
            .context("flushing Subscribe message to leader")?;

        tracing::info!("Subscribed to leader with last_offset={}", last_offset);

        // Step 2: Read messages from leader.
        let mut line = String::new();
        loop {
            line.clear();
            let bytes_read = reader
                .read_line(&mut line)
                .await
                .context("reading WAL message from leader")?;
            if bytes_read == 0 {
                // Connection closed
                anyhow::bail!("Leader closed the connection");
            }

            let trimmed = line.trim();
            if trimmed.is_empty() {
                continue;
            }

            let msg: LeaderMessage =
                serde_json::from_str(trimmed).context("parsing WAL LeaderMessage JSON")?;

            match msg {
                LeaderMessage::CaughtUp { current_offset } => {
                    tracing::info!("Caught up with leader at offset {}", current_offset,);
                    self.leader_offset.store(current_offset, Ordering::Relaxed);
                    if let Some(ref m) = self.metrics {
                        let last_replayed = self.last_replayed_offset.load(Ordering::Relaxed);
                        let lag = current_offset.saturating_sub(last_replayed);
                        m.replication_lag_seconds.set(lag as i64);
                    }
                }
                LeaderMessage::WalEntry { offset, data } => {
                    self.handle_wal_entry(offset, data, &mut writer).await?;
                }
                LeaderMessage::SnapshotStart { parquet_files } => {
                    self.handle_snapshot(&parquet_files, &mut reader, &mut writer)
                        .await?;
                }
                // SnapshotChunk and SnapshotEnd are handled inside handle_snapshot
                LeaderMessage::SnapshotChunk { .. } | LeaderMessage::SnapshotEnd { .. } => {
                    tracing::warn!(
                        "Received unexpected snapshot message outside of snapshot transfer"
                    );
                }
            }
        }
    }

    /// Handle a Parquet snapshot transfer from the leader.
    ///
    /// Called when the leader sends `SnapshotStart`. Receives all Parquet file
    /// chunks, writes them to the local snapshot directory, loads them into the
    /// EventStore, and updates the offset tracking.
    async fn handle_snapshot(
        &self,
        expected_files: &[String],
        reader: &mut BufReader<tokio::net::tcp::OwnedReadHalf>,
        writer: &mut tokio::net::tcp::OwnedWriteHalf,
    ) -> anyhow::Result<()> {
        tracing::info!(
            "Receiving Parquet snapshot ({} files: {:?})",
            expected_files.len(),
            expected_files,
        );

        // Create snapshot directory
        tokio::fs::create_dir_all(&self.snapshot_dir).await?;

        // Accumulate chunks per file
        let mut file_buffers: HashMap<String, Vec<u8>> = HashMap::new();
        for filename in expected_files {
            file_buffers.insert(filename.clone(), Vec::new());
        }

        // Read snapshot chunks until SnapshotEnd
        let mut line = String::new();
        let wal_offset_after_snapshot;

        loop {
            line.clear();
            let bytes_read = reader
                .read_line(&mut line)
                .await
                .context("reading snapshot message from leader")?;
            if bytes_read == 0 {
                anyhow::bail!("Leader closed connection during snapshot transfer");
            }

            let trimmed = line.trim();
            if trimmed.is_empty() {
                continue;
            }

            let msg: LeaderMessage =
                serde_json::from_str(trimmed).context("parsing snapshot LeaderMessage JSON")?;

            match msg {
                LeaderMessage::SnapshotChunk {
                    filename,
                    data,
                    chunk_offset: _,
                    is_last,
                } => {
                    use base64::Engine;
                    let decoded = base64::engine::general_purpose::STANDARD.decode(&data)?;

                    let buffer = file_buffers.entry(filename.clone()).or_default();
                    buffer.extend_from_slice(&decoded);

                    if is_last {
                        // Write completed file to disk
                        let file_path = self.snapshot_dir.join(&filename);
                        tokio::fs::write(&file_path, &buffer).await?;
                        tracing::info!(
                            "Received Parquet file {} ({} bytes)",
                            filename,
                            buffer.len(),
                        );
                    }
                }
                LeaderMessage::SnapshotEnd {
                    wal_offset_after_snapshot: offset,
                } => {
                    wal_offset_after_snapshot = offset;
                    tracing::info!(
                        "Snapshot transfer complete, WAL resume offset={}",
                        wal_offset_after_snapshot,
                    );
                    break;
                }
                LeaderMessage::WalEntry { .. } | LeaderMessage::CaughtUp { .. } => {
                    tracing::warn!("Received unexpected WAL message during snapshot transfer");
                }
                LeaderMessage::SnapshotStart { .. } => {
                    tracing::warn!("Received unexpected SnapshotStart during snapshot transfer");
                }
            }
        }

        // Load received Parquet files into the EventStore
        let snapshot_dir = self.snapshot_dir.clone();
        let store = Arc::clone(&self.store);

        // Use ParquetStorage to read the files and replay events
        let temp_storage = ParquetStorage::new(&snapshot_dir)?;
        let events = temp_storage.load_all_events()?;

        tracing::info!(
            "Loading {} events from snapshot into EventStore",
            events.len(),
        );

        let mut replayed = 0u64;
        for event in events {
            if let Err(e) = store.ingest_replicated(&event) {
                tracing::error!("Failed to replay snapshot event: {}", e);
            } else {
                replayed += 1;
            }
        }

        // Update tracking
        self.last_replayed_offset
            .store(wal_offset_after_snapshot, Ordering::Relaxed);
        self.total_replayed.fetch_add(replayed, Ordering::Relaxed);
        self.snapshots_received.fetch_add(1, Ordering::Relaxed);

        // Clean up snapshot files after loading
        for filename in expected_files {
            let file_path = self.snapshot_dir.join(filename);
            if let Err(e) = tokio::fs::remove_file(&file_path).await {
                tracing::debug!("Failed to clean up snapshot file {}: {}", filename, e);
            }
        }

        tracing::info!(
            "Snapshot catch-up complete: {} events loaded, resuming WAL from offset {}",
            replayed,
            wal_offset_after_snapshot,
        );

        // Send ACK for the snapshot offset
        self.send_ack(wal_offset_after_snapshot, writer).await?;

        Ok(())
    }

    /// Process a single WAL entry received from the leader.
    async fn handle_wal_entry(
        &self,
        offset: u64,
        entry: WALEntry,
        writer: &mut tokio::net::tcp::OwnedWriteHalf,
    ) -> anyhow::Result<()> {
        // Update leader offset tracking.
        self.leader_offset.store(offset, Ordering::Relaxed);

        // Track received entry
        if let Some(ref m) = self.metrics {
            m.replication_wal_received_total.inc();
        }

        // Validate CRC32 checksum.
        if !entry.verify() {
            tracing::error!(
                "CRC32 validation failed for WAL entry at offset {} — skipping",
                offset,
            );
            self.corrupted_skipped.fetch_add(1, Ordering::Relaxed);
            return Ok(());
        }

        // Skip entries we've already replayed (idempotency on reconnect).
        let current = self.last_replayed_offset.load(Ordering::Relaxed);
        if offset <= current {
            tracing::debug!("Skipping already-replayed offset {}", offset);
            // Still ACK so leader updates our position
            self.send_ack(offset, writer).await?;
            return Ok(());
        }

        // Write to local WAL for crash recovery.
        let event = entry.event.clone();
        if let Err(e) = self.local_wal.append(event.clone()) {
            tracing::error!("Failed to write to local WAL at offset {}: {}", offset, e);
            // Continue anyway — the event is still in the leader's WAL.
            // On restart, the follower will re-request from the last ACKed offset.
        }

        // Replay into EventStore (bypasses validation and local WAL write).
        if let Err(e) = self.store.ingest_replicated(&event) {
            tracing::error!(
                "Failed to replay event at offset {} into store: {}",
                offset,
                e
            );
            // Don't ACK — we'll retry on reconnect from this offset.
            return Ok(());
        }

        // Update tracking.
        self.last_replayed_offset.store(offset, Ordering::Relaxed);
        self.total_replayed.fetch_add(1, Ordering::Relaxed);

        // Update metrics
        if let Some(ref m) = self.metrics {
            m.replication_wal_replayed_total.inc();
            let lag = self
                .leader_offset
                .load(Ordering::Relaxed)
                .saturating_sub(offset);
            m.replication_lag_seconds.set(lag as i64);
        }

        // Send ACK to leader.
        self.send_ack(offset, writer).await?;

        tracing::trace!("Replayed WAL entry at offset {}", offset);

        Ok(())
    }

    /// Send an ACK message to the leader.
    async fn send_ack(
        &self,
        offset: u64,
        writer: &mut tokio::net::tcp::OwnedWriteHalf,
    ) -> anyhow::Result<()> {
        let ack = FollowerMessage::Ack { offset };
        let mut json = serde_json::to_string(&ack)?;
        json.push('\n');
        writer
            .write_all(json.as_bytes())
            .await
            .context("sending ACK to leader")?;
        writer.flush().await.context("flushing ACK to leader")?;
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_follower_replication_status_serialization() {
        let status = FollowerReplicationStatus {
            connected: true,
            leader: "core-leader:3910".to_string(),
            replication_lag_ms: 42,
            last_replayed_offset: 100,
            leader_offset: 142,
            total_replayed: 100,
            corrupted_skipped: 0,
            reconnect_count: 1,
            snapshots_received: 0,
        };
        let json = serde_json::to_value(&status).unwrap();
        assert_eq!(json["connected"], true);
        assert_eq!(json["leader"], "core-leader:3910");
        assert_eq!(json["replication_lag_ms"], 42);
        assert_eq!(json["last_replayed_offset"], 100);
        assert_eq!(json["leader_offset"], 142);
        assert_eq!(json["total_replayed"], 100);
        assert_eq!(json["corrupted_skipped"], 0);
        assert_eq!(json["reconnect_count"], 1);
        assert_eq!(json["snapshots_received"], 0);
    }

    #[test]
    fn test_follower_replication_status_defaults() {
        let status = FollowerReplicationStatus {
            connected: false,
            leader: "localhost:3910".to_string(),
            replication_lag_ms: 0,
            last_replayed_offset: 0,
            leader_offset: 0,
            total_replayed: 0,
            corrupted_skipped: 0,
            reconnect_count: 0,
            snapshots_received: 0,
        };
        let json = serde_json::to_value(&status).unwrap();
        assert_eq!(json["connected"], false);
        assert_eq!(json["replication_lag_ms"], 0);
        assert_eq!(json["snapshots_received"], 0);
    }

    #[test]
    fn test_wal_receiver_creation() {
        let store = Arc::new(EventStore::new());
        let temp_dir = tempfile::TempDir::new().unwrap();
        let receiver = WalReceiver::new(
            "localhost:3910".to_string(),
            temp_dir.path().join("follower-wal"),
            store,
        );
        assert!(receiver.is_ok());

        let receiver = receiver.unwrap();
        let status = receiver.status();
        assert!(!status.connected);
        assert_eq!(status.leader, "localhost:3910");
        assert_eq!(status.last_replayed_offset, 0);
        assert_eq!(status.total_replayed, 0);
        assert_eq!(status.snapshots_received, 0);
    }

    #[test]
    fn test_wal_receiver_recovers_offset_from_local_wal() {
        let store = Arc::new(EventStore::new());
        let temp_dir = tempfile::TempDir::new().unwrap();
        let wal_dir = temp_dir.path().join("follower-wal");

        // Write some events to a local WAL first
        {
            let wal = WriteAheadLog::new(&wal_dir, WALConfig::default()).unwrap();
            let event = crate::test_utils::test_event("test-entity", "test.replicated");
            wal.append(event).unwrap();
            let event2 = crate::test_utils::test_event("test-entity", "test.replicated");
            wal.append(event2).unwrap();
        }

        // Create receiver — it should recover the sequence from existing WAL
        let receiver = WalReceiver::new("localhost:3910".to_string(), &wal_dir, store).unwrap();

        // The local WAL recovery doesn't replay into the new WAL instance's sequence
        // counter automatically — the WAL itself starts fresh. But on follower startup
        // we recover events separately. The receiver reads current_sequence() from its
        // local WAL, which starts at 0 for a new WriteAheadLog instance.
        // The actual offset tracking is maintained via last_replayed_offset atomic.
        let status = receiver.status();
        assert_eq!(status.last_replayed_offset, 0);
    }

    #[test]
    fn test_snapshot_dir_created_correctly() {
        let store = Arc::new(EventStore::new());
        let temp_dir = tempfile::TempDir::new().unwrap();
        let wal_dir = temp_dir.path().join("follower-wal");

        let receiver = WalReceiver::new("localhost:3910".to_string(), &wal_dir, store).unwrap();

        // snapshot_dir should be a sibling of follower-wal
        assert_eq!(
            receiver.snapshot_dir,
            temp_dir.path().join("follower-snapshots"),
        );
    }
}