remdb 0.3.1

// 心跳监视器实现

use crate::ha::HARole;
use crate::ha::{HAError, Result};
use crate::pubsub;
use core::sync::atomic::{AtomicBool, AtomicU64, Ordering};

#[cfg(feature = "log")]
use crate::log::{debug, error};

// 心跳主题ID
const HEARTBEAT_TOPIC: u16 = 3;

/// 心跳数据包结构
#[repr(C)]
pub struct HeartbeatPacket {
    /// 节点ID
    node_id: u64,
    /// 时间戳（毫秒）
    timestamp: u64,
    /// 节点角色
    role: u8,
    /// CRC校验值
    crc32: u32,
}

/// 心跳接收回调函数 - 简化实现，不依赖全局静态变量
fn handle_heartbeat_callback(topic_id: u16, data: &[u8]) -> bool {
    if topic_id != HEARTBEAT_TOPIC {
        return false;
    }

    #[cfg(feature = "log")]
    debug!("Heartbeat callback received data, len: {}", data.len());

    // 解析并处理心跳数据包
    if let Some(packet) = HeartbeatPacket::from_bytes(data) {
        // 验证CRC校验值
        if !packet.verify_crc() {
            #[cfg(feature = "log")]
            debug!("Heartbeat CRC check failed");
            return true;
        }

        // 安全访问字段
        let node_id = packet.node_id();
        let role = packet.role();
        let timestamp = packet.timestamp();

        #[cfg(feature = "log")]
        debug!(
            "Received heartbeat, node_id: {}, role: {:?}, timestamp: {}",
            node_id, role, timestamp
        );
    }

    true
}

// 注意：移除了静态指针，改用其他方式处理回调

/// 心跳监视器
pub struct HeartbeatMonitor {
    /// 节点ID
    node_id: u64,
    /// 节点角色
    role: HARole,
    /// 心跳间隔（毫秒）
    heartbeat_interval: u64,
    /// 故障检测时间（毫秒）
    failure_detection_time: u64,
    /// 最后收到心跳的时间
    last_heartbeat_time: AtomicU64,
    /// 主节点是否存活
    master_alive: AtomicBool,
    /// 是否初始化
    is_initialized: bool,
    /// 接收循环是否运行
    receiver_running: AtomicBool,
    /// 发送循环是否运行
    sender_running: AtomicBool,
}

impl HeartbeatPacket {
    /// 创建新的心跳数据包
    pub fn new(node_id: u64, role: HARole) -> Self {
        let timestamp = crate::platform::get_timestamp_us() / 1000; // 转换为毫秒
        let role_u8 = role as u8;

        let mut packet = Self {
            node_id,
            timestamp,
            role: role_u8,
            crc32: 0,
        };

        // 计算CRC校验值
        packet.update_crc();

        packet
    }

    /// 更新CRC校验值
    pub fn update_crc(&mut self) {
        // 先将CRC字段置为0
        self.crc32 = 0;

        // 计算CRC值，仅包含实际数据字段，不包含填充
        let mut crc_data = [0u8; 21]; // 21 bytes of actual data

        // 复制字段数据到临时缓冲区
        unsafe {
            // node_id (8 bytes)
            core::ptr::copy_nonoverlapping(
                &self.node_id as *const u64 as *const u8,
                crc_data.as_mut_ptr(),
                8,
            );

            // timestamp (8 bytes)
            core::ptr::copy_nonoverlapping(
                &self.timestamp as *const u64 as *const u8,
                crc_data.as_mut_ptr().add(8),
                8,
            );

            // role (1 byte)
            *crc_data.as_mut_ptr().add(16) = self.role;

            // crc32 (4 bytes) - already 0, no need to copy
        };

        // 计算CRC
        self.crc32 = crate::pubsub::crc32::calculate_crc32(&crc_data);
    }

    /// 验证CRC校验值
    pub fn verify_crc(&self) -> bool {
        // 计算当前数据包的CRC值，仅包含实际数据字段
        let mut crc_data = [0u8; 21]; // 21 bytes of actual data

        // 复制字段数据到临时缓冲区
        unsafe {
            // node_id (8 bytes)
            core::ptr::copy_nonoverlapping(
                &self.node_id as *const u64 as *const u8,
                crc_data.as_mut_ptr(),
                8,
            );

            // timestamp (8 bytes)
            core::ptr::copy_nonoverlapping(
                &self.timestamp as *const u64 as *const u8,
                crc_data.as_mut_ptr().add(8),
                8,
            );

            // role (1 byte)
            *crc_data.as_mut_ptr().add(16) = self.role;

            // crc32 (4 bytes) - use 0 for calculation
        };

        // 计算CRC
        let calculated_crc = crate::pubsub::crc32::calculate_crc32(&crc_data);

        // 比较计算出的CRC值与原始数据包的CRC值
        calculated_crc == self.crc32
    }

    /// 转换为字节数组
    pub fn to_bytes(&self) -> alloc::vec::Vec<u8> {
        // 创建固定大小的缓冲区，避免多次分配
        let mut bytes = [0u8; core::mem::size_of::<Self>()];

        // 直接写入字段，避免切片操作
        let (node_id_bytes, rest) = bytes.split_at_mut(8);
        node_id_bytes.copy_from_slice(&self.node_id.to_le_bytes());

        let (timestamp_bytes, rest) = rest.split_at_mut(8);
        timestamp_bytes.copy_from_slice(&self.timestamp.to_le_bytes());

        rest[0] = self.role;
        // rest[1..4] 是填充字节，已经初始化为0

        let crc32_bytes = &mut rest[4..8];
        crc32_bytes.copy_from_slice(&self.crc32.to_le_bytes());

        bytes.to_vec()
    }

    /// 从字节数组解析
    pub fn from_bytes(bytes: &[u8]) -> Option<Self> {
        if bytes.len() != core::mem::size_of::<Self>() {
            return None;
        }

        // 安全读取字段，避免切片操作
        let mut node_id_bytes = [0u8; 8];
        node_id_bytes.copy_from_slice(&bytes[0..8]);
        let node_id = u64::from_le_bytes(node_id_bytes);

        let mut timestamp_bytes = [0u8; 8];
        timestamp_bytes.copy_from_slice(&bytes[8..16]);
        let timestamp = u64::from_le_bytes(timestamp_bytes);

        let role = bytes[16];

        let mut crc32_bytes = [0u8; 4];
        crc32_bytes.copy_from_slice(&bytes[20..24]);
        let crc32 = u32::from_le_bytes(crc32_bytes);

        Some(Self {
            node_id,
            timestamp,
            role,
            crc32,
        })
    }

    /// 获取节点ID
    pub fn node_id(&self) -> u64 {
        self.node_id
    }

    /// 获取时间戳
    pub fn timestamp(&self) -> u64 {
        self.timestamp
    }

    /// 获取节点角色
    pub fn role(&self) -> HARole {
        match self.role {
            0 => HARole::Master,
            1 => HARole::Slave,
            2 => HARole::Auto,
            _ => HARole::Auto,
        }
    }
}

impl HeartbeatMonitor {
    /// 创建新的心跳监视器
    pub fn new(heartbeat_interval: u64, failure_detection_time: u64) -> Result<Self> {
        Ok(Self {
            node_id: 0, // 默认为0，后续可以设置
            role: HARole::Auto,
            heartbeat_interval,
            failure_detection_time,
            last_heartbeat_time: AtomicU64::new(crate::platform::get_timestamp_us() / 1000),
            master_alive: AtomicBool::new(true),
            is_initialized: false,
            receiver_running: AtomicBool::new(false),
            sender_running: AtomicBool::new(false),
        })
    }

    /// 设置节点ID
    pub fn set_node_id(&mut self, node_id: u64) {
        self.node_id = node_id;
    }

    /// 设置节点角色
    pub fn set_role(&mut self, role: HARole) {
        self.role = role;
    }

    /// 初始化心跳监视器
    pub fn init(&self) -> Result<()> {
        #[cfg(feature = "log")]
        debug!(
            "Heartbeat monitor initialized, role: {:?}, node_id: {}",
            self.role, self.node_id
        );

        // 初始化pubsub系统
        self.init_pubsub()?;

        #[cfg(feature = "log")]
        debug!("Heartbeat monitor pubsub initialized");

        Ok(())
    }

    /// 初始化主节点
    pub fn init_master(&self) -> Result<()> {
        // 主节点：定期发送心跳
        #[cfg(feature = "log")]
        debug!("Initializing master node, starting heartbeat sender");

        self.start_heartbeat_sender()?;

        #[cfg(feature = "log")]
        debug!("Master node heartbeat sender started");

        Ok(())
    }

    /// 初始化从节点
    pub fn init_slave(&self) -> Result<()> {
        // 从节点：接收主节点心跳
        self.start_heartbeat_receiver()?;

        // 启动心跳检查
        self.start_heartbeat_check()?;

        Ok(())
    }

    /// 初始化pubsub系统
    fn init_pubsub(&self) -> Result<()> {
        // pubsub系统已经由HA管理器统一初始化，无需再次初始化
        // 这里只做日志记录
        #[cfg(feature = "log")]
        debug!("Heartbeat using existing pubsub system");

        Ok(())
    }

    /// 启动心跳发送器
    fn start_heartbeat_sender(&self) -> Result<()> {
        // 检查是否已经在运行
        if self.sender_running.load(Ordering::Relaxed) {
            return Ok(());
        }

        // 标记为运行中
        self.sender_running.store(true, Ordering::Relaxed);

        Ok(())
    }

    /// 启动心跳接收器
    fn start_heartbeat_receiver(&self) -> Result<()> {
        // 检查是否已经在运行
        if self.receiver_running.load(Ordering::Relaxed) {
            return Ok(());
        }

        // 标记为运行中
        self.receiver_running.store(true, Ordering::Relaxed);

        // 订阅心跳主题 - 忽略订阅失败（测试环境可能没有网络）
        match pubsub::subscribe(HEARTBEAT_TOPIC, handle_heartbeat_callback) {
            Ok(_) => {
                #[cfg(feature = "log")]
                debug!("Successfully subscribed to heartbeat topic");
            }
            Err(e) => {
                #[cfg(feature = "log")]
                error!("Failed to subscribe to heartbeat topic: {:?}", e);
                // 忽略订阅失败，继续运行（测试环境可能没有网络）
            }
        }

        Ok(())
    }

    /// 启动心跳检查
    fn start_heartbeat_check(&self) -> Result<()> {
        // 注意：简化设计，移除线程相关逻辑
        // 实际应用中，心跳检查应该由外部定时器或主循环定期调用
        Ok(())
    }

    /// 处理接收到的心跳
    fn handle_heartbeat(&self, data: &[u8]) {
        // 解析心跳数据包
        if let Some(packet) = HeartbeatPacket::from_bytes(data) {
            // 验证CRC校验值
            if !packet.verify_crc() {
                #[cfg(feature = "log")]
                debug!("Heartbeat CRC check failed");
                return;
            }

            // 安全访问字段
            let node_id = packet.node_id();
            let role = packet.role();
            let timestamp = packet.timestamp();

            #[cfg(feature = "log")]
            debug!(
                "Received heartbeat, node_id: {}, role: {:?}, timestamp: {}",
                node_id, role, timestamp
            );

            // 更新最后心跳时间
            let now = crate::platform::get_timestamp_us() / 1000; // 转换为毫秒
            self.last_heartbeat_time.store(now, Ordering::Relaxed);
            self.master_alive.store(true, Ordering::Relaxed);

            #[cfg(feature = "log")]
            debug!("Updated master alive status to true");
        }
    }

    /// 检查心跳状态
    pub fn check_status(&self) -> Result<()> {
        // 只有从节点需要检查心跳
        if self.role != HARole::Slave {
            return Ok(());
        }

        // 检查最后心跳时间
        let now = crate::platform::get_timestamp_us() / 1000; // 转换为毫秒
        let last_heartbeat = self.last_heartbeat_time.load(Ordering::Relaxed);

        // 如果超过故障检测时间，触发故障转移
        if now - last_heartbeat > self.failure_detection_time {
            self.master_alive.store(false, Ordering::Relaxed);

            // 触发故障转移
            // 注意：这里需要调用角色管理器进行角色切换
            // 由于当前设计中HeartbeatMonitor没有引用RoleManager，
            // 这里只设置状态，由HA管理器定期检查
            return Err(HAError::HeartbeatTimeout);
        }

        Ok(())
    }

    /// 发送心跳
    fn send_heartbeat(&self) -> Result<()> {
        // 构建心跳数据包
        let packet = HeartbeatPacket::new(self.node_id, self.role);

        // 安全访问字段，避免未对齐访问
        let timestamp = packet.timestamp();

        #[cfg(feature = "log")]
        debug!(
            "Sending heartbeat, node_id: {}, role: {:?}, timestamp: {}",
            self.node_id, self.role, timestamp
        );

        // 复制数据到临时缓冲区，确保publish函数使用完数据之前不会释放内存
        let bytes = packet.to_bytes();
        let mut buffer = [0u8; core::mem::size_of::<HeartbeatPacket>()];
        let copy_len = core::cmp::min(bytes.len(), buffer.len());
        unsafe {
            core::ptr::copy_nonoverlapping(bytes.as_ptr(), buffer.as_mut_ptr(), copy_len);
        }

        // 发布心跳消息
        match pubsub::publish(HEARTBEAT_TOPIC, &buffer) {
            Ok(_) => {
                #[cfg(feature = "log")]
                debug!("Heartbeat sent successfully");
                Ok(())
            }
            Err(e) => {
                #[cfg(feature = "log")]
                error!("Failed to send heartbeat: {:?}", e);
                Err(HAError::NetworkError)
            }
        }
    }

    /// 关闭心跳监视器
    pub fn shutdown(&self) -> Result<()> {
        // 停止发送器
        self.sender_running.store(false, Ordering::Relaxed);

        // 停止接收器
        self.receiver_running.store(false, Ordering::Relaxed);

        Ok(())
    }

    /// 检查主节点是否存活
    pub fn is_master_alive(&self) -> bool {
        self.master_alive.load(Ordering::Relaxed)
    }

    /// 获取最后心跳时间
    pub fn get_last_heartbeat_time(&self) -> u64 {
        self.last_heartbeat_time.load(Ordering::Relaxed)
    }
}