//! Phase 4d (2026-07-13): WAL replay engine.
//!
//! Given a peer connection and a decoded `WalMutation`, re-issue
//! the correct RPC. This is the piece that closes the loop from
//! "durably logged at client" to "actually applied at peer" on
//! reconnect.
//!
//! Failure classification is deliberate:
//!
//! * `Applied` — peer accepted the mutation (Merged, or non-versioned
//!   OK).
//! * `Superseded` — peer rejected because a dominant version already
//!   exists (`PutRefVersioned` / `PutTagVersioned` returning
//!   `Ok(false)`; `DeleteTag` on an absent key). Not a failure —
//!   the log's intent is satisfied by peer state.
//! * `Err(_)` — genuine RPC failure. Caller decides retry vs abort.
//!
//! Callers walk the WAL with `wal_mutation::replay_mutations`,
//! then feed each decoded mutation here. On `Applied` or
//! `Superseded`, advance the watermark and `wal.truncate_up_to`.
//! On error, stop and retry later — the WAL still has the record.

use crate::cluster::rpc::{
    call_delete_tag, call_put_ref, call_put_ref_versioned, call_put_tag,
    call_put_tag_versioned, call_set_tag_expiry,
};
use crate::cluster::wal_mutation::WalMutation;
use anyhow::{Context, Result};
use quinn::Connection;

/// Outcome of replaying one mutation. Both variants mean "safe to
/// advance the watermark past this record".
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ReplayOutcome {
    /// Peer applied the mutation.
    Applied,
    /// Peer already had a dominant version (versioned mutations)
    /// or the target was absent (`DeleteTag` returning `false`).
    /// The mutation's intent is satisfied by current peer state.
    Superseded,
}

/// Replay one decoded mutation. See module docs for the outcome
/// contract.
pub async fn replay_one(
    conn: &Connection,
    mutation: &WalMutation,
) -> Result<ReplayOutcome> {
    match mutation {
        WalMutation::PutRef { key, value } => {
            call_put_ref(conn, key, value)
                .await
                .context("replay PutRef")?;
            Ok(ReplayOutcome::Applied)
        }
        WalMutation::PutRefVersioned { key, stamped } => {
            let merged = call_put_ref_versioned(conn, key, stamped)
                .await
                .context("replay PutRefVersioned")?;
            Ok(if merged {
                ReplayOutcome::Applied
            } else {
                ReplayOutcome::Superseded
            })
        }
        WalMutation::PutTag { key, value } => {
            call_put_tag(conn, key, value)
                .await
                .context("replay PutTag")?;
            Ok(ReplayOutcome::Applied)
        }
        WalMutation::PutTagVersioned { key, stamped } => {
            let merged = call_put_tag_versioned(conn, key, stamped)
                .await
                .context("replay PutTagVersioned")?;
            Ok(if merged {
                ReplayOutcome::Applied
            } else {
                ReplayOutcome::Superseded
            })
        }
        WalMutation::DeleteTag { key } => {
            let removed = call_delete_tag(conn, key)
                .await
                .context("replay DeleteTag")?;
            Ok(if removed {
                ReplayOutcome::Applied
            } else {
                ReplayOutcome::Superseded
            })
        }
        WalMutation::SetTagExpiry {
            key,
            expires_at_unix,
        } => {
            call_set_tag_expiry(conn, key, *expires_at_unix)
                .await
                .context("replay SetTagExpiry")?;
            Ok(ReplayOutcome::Applied)
        }
    }
}

/// Drive a full replay from `start_seq` against a peer connection.
/// Stops at the first hard error and returns the last-applied seq
/// so the caller can `wal.truncate_up_to(last_applied)` before
/// closing.
///
/// Unknown-kind records mid-stream are skipped with a `warn!` —
/// forward-compat when a newer writer wrote a record this reader
/// doesn't understand. Malformed records also skip (loud), since
/// aborting on one bad record would prevent good tail records from
/// ever replaying.
pub async fn drive_replay(
    conn: &Connection,
    wal: &crate::cluster::wal::WriteAheadLog,
    start_seq: u64,
) -> Result<DriveReport> {
    let items = crate::cluster::wal_mutation::replay_mutations(wal, start_seq).await?;
    let mut last_applied = 0u64;
    let mut applied = 0usize;
    let mut superseded = 0usize;
    let mut skipped = 0usize;
    for (seq, decoded) in items {
        match decoded {
            Ok(m) => match replay_one(conn, &m).await {
                Ok(ReplayOutcome::Applied) => {
                    applied += 1;
                    last_applied = seq;
                }
                Ok(ReplayOutcome::Superseded) => {
                    superseded += 1;
                    last_applied = seq;
                }
                Err(e) => {
                    return Ok(DriveReport {
                        last_applied,
                        applied,
                        superseded,
                        skipped,
                        stopped_at: Some((seq, format!("{e:#}"))),
                    });
                }
            },
            Err(e) => {
                tracing::warn!(seq, error = %e, "skipping undecodable WAL record");
                skipped += 1;
                // Advance the watermark past a skipped record too —
                // it's not going to become decodable on retry.
                last_applied = seq;
            }
        }
    }
    Ok(DriveReport {
        last_applied,
        applied,
        superseded,
        skipped,
        stopped_at: None,
    })
}

/// Summary returned by `drive_replay`. `last_applied` is the
/// suggested argument to `wal.truncate_up_to`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DriveReport {
    pub last_applied: u64,
    pub applied: usize,
    pub superseded: usize,
    pub skipped: usize,
    /// `Some((seq, msg))` when replay stopped on a hard error at
    /// this record; `None` means we walked the whole log.
    pub stopped_at: Option<(u64, String)>,
}

impl DriveReport {
    pub fn is_clean(&self) -> bool {
        self.stopped_at.is_none()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::cluster::gossip::ClusterGossip;
    use crate::cluster::refs::{node_stamp_for, RefStore, StampedRef};
    use crate::cluster::rpc::{
        call_get_ref, call_get_tag_expiry, call_get_tag_versioned, RpcRouter,
    };
    use crate::cluster::tags::{StampedTagValue, TagStore};
    use crate::cluster::transport::{NodeIdentity, QuicClient, QuicServer};
    use crate::cluster::wal::WriteAheadLog;
    use crate::cluster::wal_mutation::{append_mutation, WalMutation};
    use crate::config::ClusterConfig;
    use std::net::SocketAddr;
    use std::sync::atomic::{AtomicU16, Ordering};
    use std::sync::Arc;
    use std::time::Duration;

    static NEXT_PORT: AtomicU16 = AtomicU16::new(48001);
    fn next_port() -> u16 {
        NEXT_PORT.fetch_add(1, Ordering::Relaxed)
    }
    fn loopback(port: u16) -> SocketAddr {
        format!("127.0.0.1:{port}").parse().unwrap()
    }

    async fn bootstrap_router(name: &str, port: u16) -> (tempfile::TempDir, Arc<RpcRouter>) {
        let cfg = ClusterConfig {
            zone: "fabric-10g".into(),
            bind_lan: Some(loopback(port)),
            ..Default::default()
        };
        let gossip = Arc::new(ClusterGossip::bootstrap(&cfg, name).await.unwrap());
        let tmp = tempfile::TempDir::new().unwrap();
        let blob_store = Arc::new(
            crate::cluster::blob::BlobStore::open(tmp.path().join("blobs")).unwrap(),
        );
        let ref_store = Arc::new(RefStore::open(tmp.path().join("refs")).unwrap());
        let tag_store = Arc::new(TagStore::open(tmp.path().join("tags")).unwrap());
        let router = Arc::new(
            RpcRouter::new(gossip, name.into(), "fabric-10g".into())
                .with_blob_store(blob_store)
                .with_ref_store(ref_store)
                .with_tag_store(tag_store),
        );
        (tmp, router)
    }

    async fn start_peer(
        id_a: NodeIdentity,
    ) -> (
        tempfile::TempDir,
        Arc<RpcRouter>,
        SocketAddr,
        tokio::task::JoinHandle<()>,
    ) {
        let (tmp, router) = bootstrap_router("a", next_port()).await;
        let server = QuicServer::bind(loopback(0), id_a).unwrap();
        let addr = server.local_addr().unwrap();
        let r = router.clone();
        let acc = tokio::spawn(async move {
            while let Some(Ok(conn)) = server.accept().await {
                let r = r.clone();
                tokio::spawn(async move {
                    let _ = crate::cluster::rpc::serve_connection(conn, r).await;
                });
            }
        });
        (tmp, router, addr, acc)
    }

    #[tokio::test]
    async fn replay_each_variant_end_to_end() {
        let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap(); let (_tmp, _router, addr, acc) = start_peer(id_a).await;
        let client = QuicClient::new(loopback(0), id_b).unwrap();
        let conn = client.connect(addr, "a").await.unwrap();

        // Build one of every variant, WAL them, replay, verify peer state.
        let key_r = [0xAAu8; 32];
        let val_r = [0xBBu8; 32];
        let stamped_r = StampedRef {
            value: [0xCC; 32],
            clock: 10,
            node: node_stamp_for("client"),
        };
        let stamped_t = StampedTagValue {
            value: [0xDD; 32],
            clock: 20,
            node: node_stamp_for("client"),
        };
        let mutations = vec![
            WalMutation::PutRef {
                key: key_r,
                value: val_r,
            },
            WalMutation::PutRefVersioned {
                key: [0x11; 32],
                stamped: stamped_r,
            },
            WalMutation::PutTag {
                key: "raw-tag".into(),
                value: [0xEE; 32],
            },
            WalMutation::PutTagVersioned {
                key: "stamped-tag".into(),
                stamped: stamped_t,
            },
            WalMutation::SetTagExpiry {
                key: "stamped-tag".into(),
                expires_at_unix: 1_800_000_000,
            },
        ];

        let tmp = tempfile::TempDir::new().unwrap();
        let mut wal = WriteAheadLog::open(tmp.path().join("wal")).await.unwrap();
        for m in &mutations {
            append_mutation(&mut wal, m).await.unwrap();
        }

        let report = drive_replay(&conn, &wal, 0).await.unwrap();
        assert!(report.is_clean(), "unexpected stop: {:?}", report.stopped_at);
        assert_eq!(report.applied, mutations.len());
        assert_eq!(report.superseded, 0);
        assert_eq!(report.skipped, 0);
        assert_eq!(report.last_applied, mutations.len() as u64);

        // Verify peer state.
        assert_eq!(call_get_ref(&conn, &key_r).await.unwrap(), Some(val_r));
        assert_eq!(
            call_get_tag_versioned(&conn, "stamped-tag").await.unwrap(),
            Some(stamped_t)
        );
        assert_eq!(
            call_get_tag_expiry(&conn, "stamped-tag").await.unwrap(),
            Some(1_800_000_000)
        );

        // Now demonstrate the truncate handoff.
        wal.truncate_up_to(report.last_applied).await.unwrap();
        assert!(wal.is_empty());

        conn.close(quinn::VarInt::from_u32(0), b"done");
        client.shutdown().await;
        tokio::time::sleep(Duration::from_millis(50)).await;
        acc.abort();
    }

    #[tokio::test]
    async fn versioned_reject_counts_as_superseded_not_error() {
        let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap(); let (_tmp, _router, addr, acc) = start_peer(id_a).await;
        let client = QuicClient::new(loopback(0), id_b).unwrap();
        let conn = client.connect(addr, "a").await.unwrap();

        // Pre-seed the peer with a dominant version.
        let key = [0x33u8; 32];
        let winner = StampedRef {
            value: [0xFF; 32],
            clock: 100,
            node: node_stamp_for("winner"),
        };
        assert!(call_put_ref_versioned(&conn, &key, &winner).await.unwrap());

        // WAL now contains an older-clock mutation. Replay must
        // classify it Superseded, not error.
        let older = WalMutation::PutRefVersioned {
            key,
            stamped: StampedRef {
                value: [0xAA; 32],
                clock: 1,
                node: node_stamp_for("loser"),
            },
        };
        let tmp = tempfile::TempDir::new().unwrap();
        let mut wal = WriteAheadLog::open(tmp.path().join("wal")).await.unwrap();
        append_mutation(&mut wal, &older).await.unwrap();

        let report = drive_replay(&conn, &wal, 0).await.unwrap();
        assert!(report.is_clean());
        assert_eq!(report.superseded, 1);
        assert_eq!(report.applied, 0);
        // Peer state unchanged — winner still wins.
        // (Verified via a fresh GetRefVersioned in the actual e2e above.)

        conn.close(quinn::VarInt::from_u32(0), b"done");
        client.shutdown().await;
        tokio::time::sleep(Duration::from_millis(50)).await;
        acc.abort();
    }

    #[tokio::test]
    async fn delete_missing_tag_is_superseded() {
        let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap(); let (_tmp, _router, addr, acc) = start_peer(id_a).await;
        let client = QuicClient::new(loopback(0), id_b).unwrap();
        let conn = client.connect(addr, "a").await.unwrap();

        let mutation = WalMutation::DeleteTag {
            key: "never-existed".into(),
        };
        let tmp = tempfile::TempDir::new().unwrap();
        let mut wal = WriteAheadLog::open(tmp.path().join("wal")).await.unwrap();
        append_mutation(&mut wal, &mutation).await.unwrap();

        let report = drive_replay(&conn, &wal, 0).await.unwrap();
        assert!(report.is_clean());
        assert_eq!(report.superseded, 1);

        conn.close(quinn::VarInt::from_u32(0), b"done");
        client.shutdown().await;
        tokio::time::sleep(Duration::from_millis(50)).await;
        acc.abort();
    }

    #[tokio::test]
    async fn undecodable_record_is_skipped_not_aborted() {
        let (id_a, id_b) = NodeIdentity::generate_test_pair("a", "b").unwrap(); let (_tmp, _router, addr, acc) = start_peer(id_a).await;
        let client = QuicClient::new(loopback(0), id_b).unwrap();
        let conn = client.connect(addr, "a").await.unwrap();

        let tmp = tempfile::TempDir::new().unwrap();
        let mut wal = WriteAheadLog::open(tmp.path().join("wal")).await.unwrap();
        append_mutation(
            &mut wal,
            &WalMutation::PutRef {
                key: [1; 32],
                value: [2; 32],
            },
        )
        .await
        .unwrap();
        // Unknown-kind record between two real mutations.
        wal.append(&[0x01, 0x77, 0xAA]).await.unwrap();
        append_mutation(
            &mut wal,
            &WalMutation::PutRef {
                key: [3; 32],
                value: [4; 32],
            },
        )
        .await
        .unwrap();

        let report = drive_replay(&conn, &wal, 0).await.unwrap();
        assert!(report.is_clean());
        assert_eq!(report.applied, 2);
        assert_eq!(report.skipped, 1);
        assert_eq!(report.last_applied, 3);

        conn.close(quinn::VarInt::from_u32(0), b"done");
        client.shutdown().await;
        tokio::time::sleep(Duration::from_millis(50)).await;
        acc.abort();
    }
}