//! J1939 Protocol Client Implementation
//!
//! Implements the igw Protocol traits for J1939/CAN communication.

use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;

use socketcan::{CanSocket, EmbeddedFrame, Frame, Socket};
use tokio::sync::{mpsc, RwLock};
use tokio::task::JoinHandle;
use voltage_j1939::{database_stats, decode_frame, extract_pgn, extract_source_address};

use crate::core::data::{DataBatch, DataPoint, DataType, Value};
use crate::core::error::{GatewayError, Result};
use crate::core::quality::Quality;
use crate::core::traits::{
    AdjustmentCommand, CommunicationMode, ConnectionState, ControlCommand, DataEvent,
    DataEventHandler, DataEventReceiver, DataEventSender, Diagnostics, EventDrivenProtocol,
    PollingConfig, Protocol, ProtocolCapabilities, ProtocolClient, ReadRequest, ReadResponse,
    WriteResult,
};

// ============================================================================
// Configuration
// ============================================================================

/// J1939 client configuration.
#[derive(Debug, Clone)]
pub struct J1939Config {
    /// CAN interface name (e.g., "can0").
    pub can_interface: String,

    /// Source address of the target device (ECU address).
    pub source_address: u8,

    /// Our address for sending request PGNs.
    pub our_address: u8,

    /// Request interval for on-demand PGNs in milliseconds.
    pub request_interval_ms: u64,
}

impl Default for J1939Config {
    fn default() -> Self {
        Self {
            can_interface: "can0".to_string(),
            source_address: 0x00,
            our_address: 0xFE,
            request_interval_ms: 1000,
        }
    }
}

// ============================================================================
// J1939Client
// ============================================================================

/// J1939 protocol client.
///
/// Implements event-driven communication over CAN bus using the SAE J1939 protocol.
/// Uses `voltage_j1939` crate for protocol parsing and SPN database.
pub struct J1939Client {
    config: J1939Config,

    // Connection state
    connection_state: Arc<RwLock<ConnectionState>>,
    is_connected: Arc<AtomicBool>,

    // Statistics
    read_count: AtomicU64,
    error_count: AtomicU64,
    last_error: Arc<RwLock<Option<String>>>,

    // Tasks
    receive_handle: Option<JoinHandle<()>>,

    // Event channel
    event_sender: Option<DataEventSender>,
    event_handler: Option<Arc<dyn DataEventHandler>>,

    // Cached data (latest values)
    cached_data: Arc<RwLock<HashMap<String, DataPoint>>>,
}

impl J1939Client {
    /// Create a new J1939 client with the given configuration.
    pub fn new(config: J1939Config) -> Self {
        Self {
            config,
            connection_state: Arc::new(RwLock::new(ConnectionState::Disconnected)),
            is_connected: Arc::new(AtomicBool::new(false)),
            read_count: AtomicU64::new(0),
            error_count: AtomicU64::new(0),
            last_error: Arc::new(RwLock::new(None)),
            receive_handle: None,
            event_sender: None,
            event_handler: None,
            cached_data: Arc::new(RwLock::new(HashMap::new())),
        }
    }

    /// Start the receive task.
    fn start_receive_task(&mut self) -> Result<()> {
        let can_interface = self.config.can_interface.clone();
        let source_address = self.config.source_address;
        let is_connected = Arc::clone(&self.is_connected);
        let cached_data = Arc::clone(&self.cached_data);
        let read_count = self.read_count.clone();
        let error_count = self.error_count.clone();
        let last_error = Arc::clone(&self.last_error);
        let event_sender = self.event_sender.clone();
        let event_handler = self.event_handler.clone();

        let handle = tokio::spawn(async move {
            let socket = match CanSocket::open(&can_interface) {
                Ok(s) => s,
                Err(e) => {
                    *last_error.write().await = Some(format!("Failed to open CAN socket: {}", e));
                    error_count.fetch_add(1, Ordering::Relaxed);
                    return;
                }
            };

            loop {
                if !is_connected.load(Ordering::SeqCst) {
                    break;
                }

                match socket.read_frame() {
                    Ok(frame) => {
                        if let Some(id) = frame.id().as_extended() {
                            let can_id = id.as_raw();
                            let sa = extract_source_address(can_id);

                            // Filter by source address
                            if sa != source_address {
                                continue;
                            }

                            // Use voltage_j1939 to decode the frame
                            let decoded_spns = decode_frame(can_id, frame.data());
                            if decoded_spns.is_empty() {
                                continue;
                            }

                            let timestamp = chrono::Utc::now();
                            let mut batch = DataBatch::new();

                            for decoded in decoded_spns {
                                let data_point = DataPoint {
                                    id: decoded.spn.to_string(),
                                    data_type: DataType::Telemetry,
                                    value: Value::Float(decoded.value),
                                    quality: Quality::Good,
                                    timestamp,
                                };

                                batch.push(data_point.clone());

                                // Update cache
                                cached_data
                                    .write()
                                    .await
                                    .insert(decoded.spn.to_string(), data_point);
                            }

                            if !batch.is_empty() {
                                read_count.fetch_add(1, Ordering::Relaxed);

                                // Send event
                                if let Some(ref sender) = event_sender {
                                    let _ = sender.send(DataEvent::DataUpdate(batch.clone())).await;
                                }

                                // Call handler
                                if let Some(ref handler) = event_handler {
                                    handler.on_data_update(batch).await;
                                }
                            }
                        }
                    }
                    Err(e) => {
                        *last_error.write().await = Some(format!("CAN read error: {}", e));
                        error_count.fetch_add(1, Ordering::Relaxed);
                        tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
                    }
                }
            }
        });

        self.receive_handle = Some(handle);
        Ok(())
    }
}

// ============================================================================
// Trait Implementations
// ============================================================================

impl ProtocolCapabilities for J1939Client {
    fn name(&self) -> &'static str {
        "J1939"
    }

    fn supported_modes(&self) -> &[CommunicationMode] {
        &[CommunicationMode::EventDriven]
    }

    fn supports_client(&self) -> bool {
        true
    }

    fn supports_server(&self) -> bool {
        false
    }

    fn version(&self) -> &'static str {
        "SAE J1939-21"
    }
}

impl Protocol for J1939Client {
    fn connection_state(&self) -> ConnectionState {
        *futures::executor::block_on(self.connection_state.read())
    }

    async fn read(&self, request: ReadRequest) -> Result<ReadResponse> {
        let cached = self.cached_data.read().await;

        let mut batch = DataBatch::new();

        match (&request.data_type, &request.point_ids) {
            (None, None) => {
                // Return all cached data
                for point in cached.values() {
                    batch.push(point.clone());
                }
            }
            (Some(DataType::Telemetry), None) => {
                for point in cached.values() {
                    if point.data_type == DataType::Telemetry {
                        batch.push(point.clone());
                    }
                }
            }
            (None, Some(ids)) => {
                for id in ids {
                    if let Some(point) = cached.get(id) {
                        batch.push(point.clone());
                    }
                }
            }
            (Some(dtype), Some(ids)) => {
                for id in ids {
                    if let Some(point) = cached.get(id) {
                        if point.data_type == *dtype {
                            batch.push(point.clone());
                        }
                    }
                }
            }
            _ => {}
        }

        Ok(ReadResponse::success(batch))
    }

    async fn diagnostics(&self) -> Result<Diagnostics> {
        let (spn_count, pgn_count) = database_stats();

        Ok(Diagnostics {
            protocol: "J1939".to_string(),
            connection_state: *self.connection_state.read().await,
            read_count: self.read_count.load(Ordering::Relaxed),
            write_count: 0,
            error_count: self.error_count.load(Ordering::Relaxed),
            last_error: self.last_error.read().await.clone(),
            extra: serde_json::json!({
                "can_interface": self.config.can_interface,
                "source_address": format!("0x{:02X}", self.config.source_address),
                "spn_count": spn_count,
                "pgn_count": pgn_count,
            }),
        })
    }
}

impl ProtocolClient for J1939Client {
    async fn connect(&mut self) -> Result<()> {
        *self.connection_state.write().await = ConnectionState::Connecting;

        // Verify CAN interface exists
        let _socket = CanSocket::open(&self.config.can_interface).map_err(|e| {
            GatewayError::Connection(format!(
                "Failed to open CAN interface {}: {}",
                self.config.can_interface, e
            ))
        })?;

        self.is_connected.store(true, Ordering::SeqCst);
        *self.connection_state.write().await = ConnectionState::Connected;

        // Start receive task
        self.start_receive_task()?;

        // Notify connection change
        if let Some(ref sender) = self.event_sender {
            let _ = sender
                .send(DataEvent::ConnectionChanged(ConnectionState::Connected))
                .await;
        }
        if let Some(ref handler) = self.event_handler {
            handler
                .on_connection_changed(ConnectionState::Connected)
                .await;
        }

        Ok(())
    }

    async fn disconnect(&mut self) -> Result<()> {
        self.is_connected.store(false, Ordering::SeqCst);

        if let Some(handle) = self.receive_handle.take() {
            handle.abort();
        }

        *self.connection_state.write().await = ConnectionState::Disconnected;

        // Notify connection change
        if let Some(ref sender) = self.event_sender {
            let _ = sender
                .send(DataEvent::ConnectionChanged(ConnectionState::Disconnected))
                .await;
        }
        if let Some(ref handler) = self.event_handler {
            handler
                .on_connection_changed(ConnectionState::Disconnected)
                .await;
        }

        Ok(())
    }

    async fn write_control(&mut self, _commands: &[ControlCommand]) -> Result<WriteResult> {
        // J1939 control requires proprietary PGN support
        Err(GatewayError::NotSupported(
            "J1939 control commands require proprietary PGN implementation".to_string(),
        ))
    }

    async fn write_adjustment(
        &mut self,
        _adjustments: &[AdjustmentCommand],
    ) -> Result<WriteResult> {
        // J1939 adjustment requires proprietary PGN support
        Err(GatewayError::NotSupported(
            "J1939 adjustment commands require proprietary PGN implementation".to_string(),
        ))
    }

    async fn start_polling(&mut self, _config: PollingConfig) -> Result<()> {
        // J1939 is event-driven, no polling needed
        Ok(())
    }

    async fn stop_polling(&mut self) -> Result<()> {
        Ok(())
    }
}

impl EventDrivenProtocol for J1939Client {
    fn subscribe(&self) -> DataEventReceiver {
        // This is a simplified implementation
        // In a real implementation, you'd want to support multiple subscribers
        let (_tx, rx) = mpsc::channel(100);
        // Note: This overwrites any existing sender, which is not ideal
        // A proper implementation would use a broadcast channel
        rx
    }

    fn set_event_handler(&mut self, handler: Arc<dyn DataEventHandler>) {
        self.event_handler = Some(handler);
    }
}

// ============================================================================
// Tests
// ============================================================================

#[cfg(test)]
mod tests {
    use super::*;
    use voltage_j1939::{extract_pgn, parse_can_id};

    #[test]
    fn test_parse_can_id() {
        // EEC1 from SA=0x00: CAN ID = 0x0CF00400
        let id = parse_can_id(0x0CF00400);
        assert_eq!(id.priority, 3);
        assert_eq!(id.pgn, 61444);
        assert_eq!(id.source_address, 0x00);

        // ET1 from SA=0x00: CAN ID = 0x18FEEE00
        let id = parse_can_id(0x18FEEE00);
        assert_eq!(id.priority, 6);
        assert_eq!(id.pgn, 65262);
        assert_eq!(id.source_address, 0x00);
    }

    #[test]
    fn test_decode_frame() {
        // EEC1 frame
        let can_id = 0x0CF00400;
        let data = [0x00, 0x00, 0x00, 0x20, 0x4E, 0x00, 0x00, 0x00];
        let decoded = decode_frame(can_id, &data);
        assert!(!decoded.is_empty());

        // Find engine speed (SPN 190)
        let engine_speed = decoded.iter().find(|d| d.spn == 190);
        assert!(engine_speed.is_some());
        assert_eq!(engine_speed.unwrap().value, 2500.0);
    }

    #[test]
    fn test_config_default() {
        let config = J1939Config::default();
        assert_eq!(config.can_interface, "can0");
        assert_eq!(config.source_address, 0x00);
        assert_eq!(config.our_address, 0xFE);
    }

    #[test]
    fn test_client_creation() {
        let config = J1939Config::default();
        let client = J1939Client::new(config);
        assert_eq!(client.name(), "J1939");
        assert_eq!(client.supported_modes(), &[CommunicationMode::EventDriven]);
    }
}
