//! Thing - The Consensus Council (Raft).
//!
//! Named after the Nordic governing assembly, the Thing ensures all
//! nodes in the Yggdrasil cluster agree on the state of the well.
//!
//! Uses raft 0.7 for actual Raft consensus with leader election,
//! log replication, and committed entry application.

use crate::event_bus::EventBus;
use crate::signal::MemorySignal;
use crate::wal::Saga;
use mimirs_core::{AgentId, MimirError};
use raft::Config as RaftConfig;
use raft::RawNode;
use raft::prelude::*;
use raft::storage::MemStorage;
use std::collections::HashMap;
/// ThingNode - A node in the consensus cluster.
pub struct ThingNode {
    /// The Raft raw node.
    raft_node: RawNode<MemStorage>,
    /// Local log of changes.
    pub saga: Saga,
    /// Current term in the consensus process.
    pub term: u64,
    /// Reference to the event bus for cluster coordination.
    pub bus: EventBus,
    /// Local agent identifier.
    pub agent_id: AgentId,
    /// Pending proposals awaiting commitment, keyed by Raft log index.
    /// The log index is read back after propose() and matches entry.index in advance().
    pending_proposals: HashMap<u64, tokio::sync::oneshot::Sender<Result<(), MimirError>>>,
}

impl ThingNode {
    /// Opna - Creates a new Thing node.
    pub fn new(agent_id: AgentId, bus: EventBus) -> Self {
        let storage = MemStorage::new();

        let cfg = RaftConfig {
            id: agent_id.0.as_u128() as u64,
            election_tick: 10,
            heartbeat_tick: 3,
            max_size_per_msg: 1024 * 1024,
            max_inflight_msgs: 256,
            ..Default::default()
        };

        let logger = raft::default_logger();
        let raft_node = RawNode::new(&cfg, storage, &logger).expect("Failed to create Raft node");

        Self {
            raft_node,
            saga: Saga::new(),
            term: 0,
            bus,
            agent_id,
            pending_proposals: HashMap::new(),
        }
    }

    /// Samykki - Propose and reach consensus on a change.
    ///
    /// CONTRACT: Registers a oneshot channel keyed by the Raft log index of the
    /// proposal. `advance()` resolves the channel when the entry is committed.
    /// The WAL is written only in `advance()` after consensus — never here.
    pub async fn samykki(&mut self, change: Vec<u8>) -> Result<(), MimirError> {
        self.raft_node.tick();

        let (tx, rx) = tokio::sync::oneshot::channel();

        self.raft_node
            .propose(vec![], change.clone())
            .map_err(|e| MimirError::Internal(format!("Raft proposal failed: {e}")))?;

        // Key by the log index that Raft just assigned to our proposal.
        // This matches entry.index in advance(), regardless of leader elections
        // or configuration entries that shift the global index sequence.
        let log_index = self.raft_node.raft.raft_log.last_index();
        self.pending_proposals.insert(log_index, tx);

        let _ = self.bus.broadcast(MemorySignal::ConsensusReached {
            agent_id: self.agent_id,
            term: self.term,
        });

        tracing::info!(
            term = self.term,
            log_index = log_index,
            "Change proposed to Raft consensus"
        );

        match tokio::time::timeout(std::time::Duration::from_secs(5), rx).await {
            Ok(Ok(result)) => result,
            Ok(Err(_)) => Err(MimirError::Internal("Proposal channel closed".to_string())),
            Err(_) => Err(MimirError::Internal(
                "Proposal timed out waiting for consensus".to_string(),
            )),
        }
    }

    /// Advance the Raft state machine and process any committed entries.
    pub fn advance(&mut self) -> Result<(), MimirError> {
        self.raft_node.tick();

        let mut ready = self.raft_node.ready();
        if !ready.committed_entries().is_empty() {
            for entry in ready.take_committed_entries() {
                if !entry.data.is_empty() {
                    let saga_entry: crate::wal::SagaEntry = bincode::deserialize(&entry.data)
                        .map_err(|e| MimirError::Serialization(e.to_string()))?;
                    self.saga.skrifa(saga_entry)?;
                }

                if let Some(tx) = self.pending_proposals.remove(&entry.index) {
                    let _ = tx.send(Ok(()));
                }
            }
        }

        self.raft_node.advance(ready);
        self.term = self.raft_node.raft.hard_state().term;

        Ok(())
    }

    /// Check if this node is the leader.
    pub fn is_leader(&self) -> bool {
        self.raft_node.raft.state == raft::StateRole::Leader
    }

    /// Get the current Raft term.
    pub fn current_term(&self) -> u64 {
        self.raft_node.raft.hard_state().term
    }

    /// Handle an incoming Raft message from a peer.
    pub fn step(&mut self, msg: Message) -> Result<(), MimirError> {
        self.raft_node
            .step(msg)
            .map_err(|e| MimirError::Internal(format!("Raft step failed: {e}")))
    }
}
