use std::path::Path;

use serde::{Deserialize, Serialize};

use crate::types::{GroupName, MessageEnvelope, NamespaceId, TopicName, TopicOffset};

/// Monotonic sequence number across all replication entries.
pub type SequenceNum = u64;

/// Monotonic term counter, incremented on each leader promotion.
///
/// Persisted to disk so a restarted node knows its last term and can
/// detect stale leaders. Higher term always wins.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
pub struct Term(pub u64);

impl Term {
    /// Initial term for a fresh cluster.
    pub const ZERO: Self = Self(0);

    /// Increment this term (used on promotion).
    pub fn next(self) -> Self {
        Self(self.0 + 1)
    }

    /// Persist the term to `<dir>/term`.
    pub fn persist(&self, dir: &Path) -> std::io::Result<()> {
        std::fs::create_dir_all(dir)?;
        std::fs::write(dir.join("term"), self.0.to_string())
    }

    /// Load the term from `<dir>/term`. Returns `Term::ZERO` if the file
    /// does not exist.
    pub fn load(dir: &Path) -> std::io::Result<Self> {
        let path = dir.join("term");
        match std::fs::read_to_string(&path) {
            Ok(s) => {
                let val: u64 = s.trim().parse().map_err(|e| {
                    std::io::Error::new(std::io::ErrorKind::InvalidData, format!("bad term: {e}"))
                })?;
                Ok(Self(val))
            }
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(Self::ZERO),
            Err(e) => Err(e),
        }
    }
}

impl std::fmt::Display for Term {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.0)
    }
}

/// Streamed from leader to followers over the cluster TCP connection.
/// Length-prefixed, same framing pattern as client protocol.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum ReplicationEntry {
    Publish {
        term: Term,
        namespace: NamespaceId,
        topic: TopicName,
        envelope: MessageEnvelope,
    },
    CommitOffset {
        term: Term,
        namespace: NamespaceId,
        topic: TopicName,
        group: GroupName,
        offset: TopicOffset,
    },
    TopicCreated {
        term: Term,
        namespace: NamespaceId,
        topic: TopicName,
    },
}

/// Follower → Leader acknowledgment.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum ReplicationAck {
    Progress { last_applied_sequence: SequenceNum },
}

/// State of a follower as tracked by the leader.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum FollowerState {
    /// Handshake in progress.
    Joining,
    /// Receiving replay or snapshot.
    CatchingUp,
    /// Fully caught up, eligible for promotion.
    InSync,
    /// Heartbeat lost.
    Disconnected,
}

/// Unique identifier for a node in the cluster.
pub type NodeId = u64;

/// Sent on the peer protocol during Leader Suspect state.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum PeerMessage {
    PromotionCandidate {
        node_id: NodeId,
        last_applied_sequence: SequenceNum,
        term: Term,
    },
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::types::{MessageId, TopicOffset};

    #[test]
    fn term_next() {
        assert_eq!(Term::ZERO.next(), Term(1));
        assert_eq!(Term(41).next(), Term(42));
    }

    #[test]
    fn term_ordering() {
        assert!(Term(1) < Term(2));
        assert!(Term(2) > Term(1));
        assert_eq!(Term(3), Term(3));
    }

    #[test]
    fn term_persist_and_load() {
        let dir = tempfile::tempdir().unwrap();
        let term = Term(42);
        term.persist(dir.path()).unwrap();
        let loaded = Term::load(dir.path()).unwrap();
        assert_eq!(loaded, term);
    }

    #[test]
    fn term_load_missing_returns_zero() {
        let dir = tempfile::tempdir().unwrap();
        let loaded = Term::load(dir.path()).unwrap();
        assert_eq!(loaded, Term::ZERO);
    }

    #[test]
    fn term_persist_overwrite() {
        let dir = tempfile::tempdir().unwrap();
        Term(1).persist(dir.path()).unwrap();
        Term(5).persist(dir.path()).unwrap();
        assert_eq!(Term::load(dir.path()).unwrap(), Term(5));
    }

    #[test]
    fn replication_entry_serde_roundtrip_publish() {
        let entry = ReplicationEntry::Publish {
            term: Term(3),
            namespace: NamespaceId("ns1".into()),
            topic: TopicName("events".into()),
            envelope: crate::types::MessageEnvelope::new(
                MessageId(1),
                NamespaceId("ns1".into()),
                TopicName("events".into()),
                vec![1, 2, 3],
                1000,
            ),
        };
        let bytes = bincode::serialize(&entry).unwrap();
        let decoded: ReplicationEntry = bincode::deserialize(&bytes).unwrap();
        match decoded {
            ReplicationEntry::Publish {
                term,
                namespace,
                topic,
                envelope,
            } => {
                assert_eq!(term, Term(3));
                assert_eq!(namespace, NamespaceId("ns1".into()));
                assert_eq!(topic, TopicName("events".into()));
                assert_eq!(envelope.payload, vec![1, 2, 3]);
            }
            other => panic!("expected Publish, got: {other:?}"),
        }
    }

    #[test]
    fn replication_entry_serde_roundtrip_commit_offset() {
        let entry = ReplicationEntry::CommitOffset {
            term: Term(5),
            namespace: NamespaceId("ns1".into()),
            topic: TopicName("orders".into()),
            group: GroupName("workers".into()),
            offset: TopicOffset(99),
        };
        let bytes = bincode::serialize(&entry).unwrap();
        let decoded: ReplicationEntry = bincode::deserialize(&bytes).unwrap();
        match decoded {
            ReplicationEntry::CommitOffset {
                term,
                group,
                offset,
                ..
            } => {
                assert_eq!(term, Term(5));
                assert_eq!(group, GroupName("workers".into()));
                assert_eq!(offset, TopicOffset(99));
            }
            other => panic!("expected CommitOffset, got: {other:?}"),
        }
    }

    #[test]
    fn replication_entry_serde_roundtrip_topic_created() {
        let entry = ReplicationEntry::TopicCreated {
            term: Term(1),
            namespace: NamespaceId("ns1".into()),
            topic: TopicName("new-topic".into()),
        };
        let bytes = bincode::serialize(&entry).unwrap();
        let decoded: ReplicationEntry = bincode::deserialize(&bytes).unwrap();
        match decoded {
            ReplicationEntry::TopicCreated {
                term,
                namespace,
                topic,
            } => {
                assert_eq!(term, Term(1));
                assert_eq!(namespace, NamespaceId("ns1".into()));
                assert_eq!(topic, TopicName("new-topic".into()));
            }
            other => panic!("expected TopicCreated, got: {other:?}"),
        }
    }

    #[test]
    fn replication_ack_serde_roundtrip() {
        let ack = ReplicationAck::Progress {
            last_applied_sequence: 12345,
        };
        let bytes = bincode::serialize(&ack).unwrap();
        let decoded: ReplicationAck = bincode::deserialize(&bytes).unwrap();
        match decoded {
            ReplicationAck::Progress {
                last_applied_sequence,
            } => {
                assert_eq!(last_applied_sequence, 12345);
            }
        }
    }

    #[test]
    fn follower_state_serde_roundtrip() {
        for state in [
            FollowerState::Joining,
            FollowerState::CatchingUp,
            FollowerState::InSync,
            FollowerState::Disconnected,
        ] {
            let bytes = bincode::serialize(&state).unwrap();
            let decoded: FollowerState = bincode::deserialize(&bytes).unwrap();
            assert_eq!(decoded, state);
        }
    }

    #[test]
    fn peer_message_serde_roundtrip() {
        let msg = PeerMessage::PromotionCandidate {
            node_id: 2,
            last_applied_sequence: 50000,
            term: Term(10),
        };
        let bytes = bincode::serialize(&msg).unwrap();
        let decoded: PeerMessage = bincode::deserialize(&bytes).unwrap();
        match decoded {
            PeerMessage::PromotionCandidate {
                node_id,
                last_applied_sequence,
                term,
            } => {
                assert_eq!(node_id, 2);
                assert_eq!(last_applied_sequence, 50000);
                assert_eq!(term, Term(10));
            }
        }
    }
}
