//! Channel management module
//!
//! Provides channel registry for tracking active messaging channels
//! and their connection states.

pub mod activity;
pub mod console;
pub mod discord;
pub mod discord_gateway;
pub mod inbound;
pub mod matrix;
pub(crate) mod media_fetch;
pub mod signal;
pub mod signal_receive;
pub mod slack;
pub mod slack_inbound;
pub mod telegram;
pub mod telegram_inbound;
pub mod telegram_receive;
pub mod webhook;

use parking_lot::RwLock;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;

const CHANNEL_RETRY_AFTER_MAX: Duration = Duration::from_secs(60 * 60);
const CHANNEL_RETRY_AFTER_MAX_MS: i64 = 60 * 60 * 1_000;

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ChannelAuthErrorKind {
    Auth,
    Transient,
}

#[derive(Debug, Clone)]
pub struct ChannelAuthError {
    kind: ChannelAuthErrorKind,
    message: String,
}

impl ChannelAuthError {
    pub fn auth(message: impl Into<String>) -> Self {
        Self {
            kind: ChannelAuthErrorKind::Auth,
            message: message.into(),
        }
    }

    pub fn transient(message: impl Into<String>) -> Self {
        Self {
            kind: ChannelAuthErrorKind::Transient,
            message: message.into(),
        }
    }

    pub fn is_auth(&self) -> bool {
        self.kind == ChannelAuthErrorKind::Auth
    }

    pub fn message(&self) -> &str {
        &self.message
    }
}

pub type ChannelAuthResult = Result<(), ChannelAuthError>;

pub(crate) fn retry_after_ms_from_headers(headers: &reqwest::header::HeaderMap) -> Option<i64> {
    let raw = headers
        .get(reqwest::header::RETRY_AFTER)
        .and_then(|value| value.to_str().ok())?
        .trim();
    if raw.is_empty() {
        return None;
    }
    if let Ok(seconds) = raw.parse::<u64>() {
        return duration_to_retry_after_ms(Duration::from_secs(seconds));
    }
    let deadline = retry_after_http_date(raw)?;
    let now = chrono::Utc::now();
    let delta = deadline.with_timezone(&chrono::Utc) - now;
    if delta <= chrono::Duration::zero() {
        return Some(0);
    }
    Some(
        delta
            .num_milliseconds()
            .clamp(0, CHANNEL_RETRY_AFTER_MAX_MS),
    )
}

fn retry_after_http_date(raw: &str) -> Option<chrono::DateTime<chrono::FixedOffset>> {
    if let Ok(deadline) = chrono::DateTime::parse_from_rfc2822(raw) {
        return Some(deadline);
    }
    if let Ok(deadline) = chrono::NaiveDateTime::parse_from_str(raw, "%A, %d-%b-%y %H:%M:%S GMT") {
        return Some(deadline.and_utc().fixed_offset());
    }
    if let Ok(deadline) = chrono::NaiveDateTime::parse_from_str(raw, "%a %b %e %H:%M:%S %Y") {
        return Some(deadline.and_utc().fixed_offset());
    }
    None
}

pub(crate) fn retry_after_ms_from_seconds_f64(seconds: f64) -> Option<i64> {
    if !seconds.is_finite() || seconds < 0.0 {
        return None;
    }
    duration_to_retry_after_ms(Duration::try_from_secs_f64(seconds).ok()?)
}

fn duration_to_retry_after_ms(duration: Duration) -> Option<i64> {
    i64::try_from(duration.min(CHANNEL_RETRY_AFTER_MAX).as_millis()).ok()
}

/// Connection status of a channel
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ChannelStatus {
    /// Channel is connected and ready
    Connected,
    /// Channel is disconnected
    #[default]
    Disconnected,
    /// Channel is in the process of connecting
    Connecting,
    /// Channel encountered an error
    Error,
    /// Channel is paused/suspended
    Paused,
    /// Channel logged out explicitly
    LoggedOut,
}

impl std::fmt::Display for ChannelStatus {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Connected => write!(f, "{}", Self::CONNECTED_WIRE),
            Self::Disconnected => write!(f, "disconnected"),
            Self::Connecting => write!(f, "connecting"),
            Self::Error => write!(f, "error"),
            Self::Paused => write!(f, "paused"),
            Self::LoggedOut => write!(f, "logged_out"),
        }
    }
}

impl ChannelStatus {
    /// Wire-format string emitted by `Display::fmt` for the
    /// `Connected` state. Exported so callers polling
    /// `channels.status` (e.g. `cara verify --outcome matrix`) can
    /// match against the same constant the producer uses, rather
    /// than duplicating the literal `"connected"` at every consumer.
    /// A future Display rename (e.g. to `"online"`) would touch this
    /// constant and trip every consumer that imported it, instead of
    /// silently breaking the polling loops.
    pub const CONNECTED_WIRE: &'static str = "connected";
}

/// Metadata associated with a channel
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct ChannelMetadata {
    /// Human-readable description
    #[serde(skip_serializing_if = "Option::is_none")]
    pub description: Option<String>,
    /// Channel-specific configuration or state
    #[serde(skip_serializing_if = "Option::is_none")]
    pub extra: Option<serde_json::Value>,
    /// Last error message if status is Error
    #[serde(skip_serializing_if = "Option::is_none")]
    pub last_error: Option<String>,
    /// Timestamp of last successful connection (Unix ms)
    #[serde(skip_serializing_if = "Option::is_none")]
    pub last_connected_at: Option<i64>,
    /// Timestamp of last status change (Unix ms)
    #[serde(skip_serializing_if = "Option::is_none")]
    pub status_changed_at: Option<i64>,
}

/// Information about a registered channel
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ChannelInfo {
    /// Unique channel identifier (e.g., "telegram", "discord", "slack")
    pub id: String,
    /// Human-readable channel name
    pub name: String,
    /// Current connection status
    pub status: ChannelStatus,
    /// Additional metadata
    #[serde(default)]
    pub metadata: ChannelMetadata,
}

impl ChannelInfo {
    /// Create a new channel info with the given ID and name
    pub fn new(id: impl Into<String>, name: impl Into<String>) -> Self {
        Self {
            id: id.into(),
            name: name.into(),
            status: ChannelStatus::Disconnected,
            metadata: ChannelMetadata::default(),
        }
    }

    /// Create a channel info with status
    pub fn with_status(mut self, status: ChannelStatus) -> Self {
        self.status = status;
        self
    }

    /// Create a channel info with metadata
    pub fn with_metadata(mut self, metadata: ChannelMetadata) -> Self {
        self.metadata = metadata;
        self
    }
}

/// Snapshot of the channel registry state for API responses
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RegistrySnapshot {
    /// All registered channels
    pub channels: Vec<ChannelInfo>,
    /// Timestamp when snapshot was taken (Unix ms)
    pub timestamp: i64,
}

/// Thread-safe registry for tracking active channels
#[derive(Debug)]
pub struct ChannelRegistry {
    channels: RwLock<HashMap<String, ChannelInfo>>,
}

impl Default for ChannelRegistry {
    fn default() -> Self {
        Self::new()
    }
}

impl ChannelRegistry {
    /// Create a new empty channel registry
    pub fn new() -> Self {
        Self {
            channels: RwLock::new(HashMap::new()),
        }
    }

    /// Register a channel in the registry
    ///
    /// If a channel with the same ID already exists, it will be replaced.
    pub fn register(&self, info: ChannelInfo) {
        let mut channels = self.channels.write();
        channels.insert(info.id.clone(), info);
    }

    /// Unregister a channel from the registry
    ///
    /// Returns the removed channel info if it existed.
    pub fn unregister(&self, channel_id: &str) -> Option<ChannelInfo> {
        let mut channels = self.channels.write();
        channels.remove(channel_id)
    }

    /// Get the status of a specific channel
    pub fn get_status(&self, channel_id: &str) -> Option<ChannelStatus> {
        let channels = self.channels.read();
        channels.get(channel_id).map(|info| info.status)
    }

    /// Get full info for a specific channel
    pub fn get(&self, channel_id: &str) -> Option<ChannelInfo> {
        let channels = self.channels.read();
        channels.get(channel_id).cloned()
    }

    /// List all registered channels
    pub fn list(&self) -> Vec<ChannelInfo> {
        let channels = self.channels.read();
        channels.values().cloned().collect()
    }

    /// Update the status of a channel
    ///
    /// Returns true if the channel existed and was updated.
    ///
    /// Clears `last_error` on every transition out of `Error`, not just on
    /// transitions to `Connected`. Earlier behavior left a stale error
    /// message on the channel after a clean `Disconnected`/`LoggedOut`
    /// transition, which surfaced misleading errors in operator UIs after
    /// graceful disconnects.
    pub fn update_status(&self, channel_id: &str, status: ChannelStatus) -> bool {
        let mut channels = self.channels.write();
        if let Some(info) = channels.get_mut(channel_id) {
            let prev_status = info.status;
            if prev_status == status {
                return true;
            }
            info.status = status;
            info.metadata.status_changed_at = Some(now_millis());
            if status == ChannelStatus::Connected {
                info.metadata.last_connected_at = Some(now_millis());
            }
            if status != ChannelStatus::Error && prev_status == ChannelStatus::Error {
                info.metadata.last_error = None;
            }
            true
        } else {
            false
        }
    }

    /// Update a channel's metadata `extra` field under the registry write lock.
    pub(crate) fn update_metadata_extra(&self, channel_id: &str, extra: serde_json::Value) -> bool {
        let mut channels = self.channels.write();
        if let Some(info) = channels.get_mut(channel_id) {
            info.metadata.extra = Some(extra);
            true
        } else {
            false
        }
    }

    /// Update the status and set an error message.
    ///
    /// `status_changed_at` is bumped only on a real transition (status flip
    /// or last_error message change). Idempotent re-stamps with the same
    /// `(Error, message)` are no-ops, preserving the first-occurrence
    /// timestamp for dashboards polling `statusChangedAt` to detect new
    /// errors. Hot path under the give-up policy stamps SyncLoopGaveUp
    /// every retry interval; without this guard `status_changed_at` would
    /// advance every tick despite no observable state change.
    pub fn set_error(&self, channel_id: &str, error: impl Into<String>) -> bool {
        let mut channels = self.channels.write();
        if let Some(info) = channels.get_mut(channel_id) {
            let new_msg = error.into();
            let unchanged = info.status == ChannelStatus::Error
                && info.metadata.last_error.as_deref() == Some(new_msg.as_str());
            info.status = ChannelStatus::Error;
            info.metadata.last_error = Some(new_msg);
            if !unchanged {
                info.metadata.status_changed_at = Some(now_millis());
            }
            true
        } else {
            false
        }
    }

    /// Handle logout for a channel (sets status to LoggedOut)
    pub fn logout(&self, channel_id: &str) -> bool {
        self.update_status(channel_id, ChannelStatus::LoggedOut)
    }

    /// Create a serializable snapshot of the registry state
    pub fn snapshot(&self) -> RegistrySnapshot {
        RegistrySnapshot {
            channels: self.list(),
            timestamp: now_millis(),
        }
    }

    /// Check if a channel is currently connected
    pub fn is_connected(&self, channel_id: &str) -> bool {
        self.get_status(channel_id) == Some(ChannelStatus::Connected)
    }

    /// Get count of channels by status
    pub fn count_by_status(&self, status: ChannelStatus) -> usize {
        let channels = self.channels.read();
        channels.values().filter(|c| c.status == status).count()
    }

    /// Get total number of registered channels
    pub fn len(&self) -> usize {
        self.channels.read().len()
    }

    /// Check if registry is empty
    pub fn is_empty(&self) -> bool {
        self.channels.read().is_empty()
    }
}

/// Create a shared channel registry
pub fn create_registry() -> Arc<ChannelRegistry> {
    Arc::new(ChannelRegistry::new())
}

/// Get current time in milliseconds since Unix epoch
fn now_millis() -> i64 {
    use std::time::{SystemTime, UNIX_EPOCH};
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_millis() as i64)
        .unwrap_or(0)
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_channel_status_default() {
        assert_eq!(ChannelStatus::default(), ChannelStatus::Disconnected);
    }

    #[test]
    fn test_channel_status_display() {
        assert_eq!(ChannelStatus::Connected.to_string(), "connected");
        assert_eq!(ChannelStatus::Disconnected.to_string(), "disconnected");
        assert_eq!(ChannelStatus::Error.to_string(), "error");
        assert_eq!(ChannelStatus::LoggedOut.to_string(), "logged_out");
    }

    #[test]
    fn test_retry_after_header_seconds_parses_to_millis() {
        let mut headers = reqwest::header::HeaderMap::new();
        headers.insert(
            reqwest::header::RETRY_AFTER,
            reqwest::header::HeaderValue::from_static("2"),
        );

        assert_eq!(retry_after_ms_from_headers(&headers), Some(2_000));
    }

    #[test]
    fn test_retry_after_accepts_rfc7231_http_date_formats() {
        for value in [
            "Sun, 06 Nov 1994 08:49:37 GMT",
            "Sunday, 06-Nov-94 08:49:37 GMT",
            "Sun Nov  6 08:49:37 1994",
        ] {
            let mut headers = reqwest::header::HeaderMap::new();
            headers.insert(
                reqwest::header::RETRY_AFTER,
                reqwest::header::HeaderValue::from_static(value),
            );
            assert_eq!(
                retry_after_ms_from_headers(&headers),
                Some(0),
                "{value} must parse as a valid past Retry-After HTTP-date"
            );
        }
    }

    #[test]
    fn test_retry_after_seconds_f64_parses_to_millis() {
        assert_eq!(retry_after_ms_from_seconds_f64(1.5), Some(1_500));
        assert_eq!(retry_after_ms_from_seconds_f64(-1.0), None);
        assert_eq!(retry_after_ms_from_seconds_f64(f64::INFINITY), None);
    }

    #[test]
    fn test_retry_after_values_are_capped() {
        let mut headers = reqwest::header::HeaderMap::new();
        headers.insert(
            reqwest::header::RETRY_AFTER,
            reqwest::header::HeaderValue::from_static("999999"),
        );

        assert_eq!(
            retry_after_ms_from_headers(&headers),
            Some(CHANNEL_RETRY_AFTER_MAX_MS)
        );
        assert_eq!(
            retry_after_ms_from_seconds_f64(999_999.0),
            Some(CHANNEL_RETRY_AFTER_MAX_MS)
        );
    }

    #[test]
    fn test_channel_info_builder() {
        let info = ChannelInfo::new("telegram", "Telegram")
            .with_status(ChannelStatus::Connected)
            .with_metadata(ChannelMetadata {
                description: Some("Main bot".into()),
                ..Default::default()
            });

        assert_eq!(info.id, "telegram");
        assert_eq!(info.name, "Telegram");
        assert_eq!(info.status, ChannelStatus::Connected);
        assert_eq!(info.metadata.description, Some("Main bot".into()));
    }

    #[test]
    fn test_registry_register_and_get() {
        let registry = ChannelRegistry::new();
        let info = ChannelInfo::new("telegram", "Telegram");

        registry.register(info.clone());

        let retrieved = registry.get("telegram").unwrap();
        assert_eq!(retrieved.id, "telegram");
        assert_eq!(retrieved.name, "Telegram");
    }

    #[test]
    fn test_registry_unregister() {
        let registry = ChannelRegistry::new();
        registry.register(ChannelInfo::new("telegram", "Telegram"));

        let removed = registry.unregister("telegram");
        assert!(removed.is_some());
        assert_eq!(removed.unwrap().id, "telegram");

        assert!(registry.get("telegram").is_none());
    }

    #[test]
    fn test_registry_get_status() {
        let registry = ChannelRegistry::new();
        registry.register(
            ChannelInfo::new("telegram", "Telegram").with_status(ChannelStatus::Connected),
        );

        assert_eq!(
            registry.get_status("telegram"),
            Some(ChannelStatus::Connected)
        );
        assert_eq!(registry.get_status("nonexistent"), None);
    }

    #[test]
    fn test_registry_update_status() {
        let registry = ChannelRegistry::new();
        registry.register(ChannelInfo::new("telegram", "Telegram"));

        assert!(registry.update_status("telegram", ChannelStatus::Connected));
        assert_eq!(
            registry.get_status("telegram"),
            Some(ChannelStatus::Connected)
        );

        // Verify last_connected_at was set
        let info = registry.get("telegram").unwrap();
        assert!(info.metadata.last_connected_at.is_some());
        assert!(info.metadata.status_changed_at.is_some());
    }

    #[test]
    fn test_registry_update_status_is_idempotent_for_same_status() {
        let registry = ChannelRegistry::new();
        registry.register(
            ChannelInfo::new("telegram", "Telegram").with_status(ChannelStatus::Connected),
        );
        {
            let mut channels = registry.channels.write();
            let info = channels.get_mut("telegram").unwrap();
            info.metadata.status_changed_at = Some(10);
            info.metadata.last_connected_at = Some(20);
        }

        assert!(registry.update_status("telegram", ChannelStatus::Connected));

        let info = registry.get("telegram").unwrap();
        assert_eq!(info.metadata.status_changed_at, Some(10));
        assert_eq!(info.metadata.last_connected_at, Some(20));
    }

    #[test]
    fn test_registry_set_error() {
        let registry = ChannelRegistry::new();
        registry.register(ChannelInfo::new("telegram", "Telegram"));

        assert!(registry.set_error("telegram", "Connection failed"));
        let info = registry.get("telegram").unwrap();
        assert_eq!(info.status, ChannelStatus::Error);
        assert_eq!(info.metadata.last_error, Some("Connection failed".into()));
    }

    #[test]
    fn test_registry_clear_last_error_on_reconnect() {
        let registry = ChannelRegistry::new();
        registry.register(ChannelInfo::new("matrix", "Matrix"));

        assert!(registry.set_error("matrix", "transient failure"));
        assert!(registry.update_status("matrix", ChannelStatus::Connected));

        let info = registry.get("matrix").unwrap();
        assert_eq!(info.status, ChannelStatus::Connected);
        assert!(info.metadata.last_error.is_none());
    }

    #[test]
    fn test_registry_logout() {
        let registry = ChannelRegistry::new();
        registry.register(
            ChannelInfo::new("telegram", "Telegram").with_status(ChannelStatus::Connected),
        );

        assert!(registry.logout("telegram"));
        assert_eq!(
            registry.get_status("telegram"),
            Some(ChannelStatus::LoggedOut)
        );
    }

    #[test]
    fn test_registry_list() {
        let registry = ChannelRegistry::new();
        registry.register(ChannelInfo::new("telegram", "Telegram"));
        registry.register(ChannelInfo::new("discord", "Discord"));

        let list = registry.list();
        assert_eq!(list.len(), 2);
    }

    #[test]
    fn test_registry_snapshot() {
        let registry = ChannelRegistry::new();
        registry.register(ChannelInfo::new("telegram", "Telegram"));

        let snapshot = registry.snapshot();
        assert_eq!(snapshot.channels.len(), 1);
        assert!(snapshot.timestamp > 0);
    }

    #[test]
    fn test_registry_is_connected() {
        let registry = ChannelRegistry::new();
        registry.register(
            ChannelInfo::new("telegram", "Telegram").with_status(ChannelStatus::Connected),
        );
        registry.register(ChannelInfo::new("discord", "Discord"));

        assert!(registry.is_connected("telegram"));
        assert!(!registry.is_connected("discord"));
        assert!(!registry.is_connected("nonexistent"));
    }

    #[test]
    fn test_registry_count_by_status() {
        let registry = ChannelRegistry::new();
        registry.register(
            ChannelInfo::new("telegram", "Telegram").with_status(ChannelStatus::Connected),
        );
        registry
            .register(ChannelInfo::new("discord", "Discord").with_status(ChannelStatus::Connected));
        registry.register(ChannelInfo::new("slack", "Slack").with_status(ChannelStatus::Error));

        assert_eq!(registry.count_by_status(ChannelStatus::Connected), 2);
        assert_eq!(registry.count_by_status(ChannelStatus::Error), 1);
        assert_eq!(registry.count_by_status(ChannelStatus::Disconnected), 0);
    }

    #[test]
    fn test_registry_len_and_is_empty() {
        let registry = ChannelRegistry::new();
        assert!(registry.is_empty());
        assert_eq!(registry.len(), 0);

        registry.register(ChannelInfo::new("telegram", "Telegram"));
        assert!(!registry.is_empty());
        assert_eq!(registry.len(), 1);
    }

    #[test]
    fn test_registry_thread_safety() {
        use std::thread;

        let registry = Arc::new(ChannelRegistry::new());
        let mut handles = vec![];

        // Spawn multiple threads that register channels
        for i in 0..10 {
            let reg = Arc::clone(&registry);
            handles.push(thread::spawn(move || {
                reg.register(ChannelInfo::new(
                    format!("channel_{}", i),
                    format!("Channel {}", i),
                ));
            }));
        }

        for handle in handles {
            handle.join().unwrap();
        }

        assert_eq!(registry.len(), 10);
    }

    #[test]
    fn test_channel_info_serialization() {
        let info = ChannelInfo::new("telegram", "Telegram")
            .with_status(ChannelStatus::Connected)
            .with_metadata(ChannelMetadata {
                description: Some("Test channel".into()),
                ..Default::default()
            });

        let json = serde_json::to_string(&info).unwrap();
        let parsed: ChannelInfo = serde_json::from_str(&json).unwrap();

        assert_eq!(parsed.id, info.id);
        assert_eq!(parsed.name, info.name);
        assert_eq!(parsed.status, info.status);
    }

    #[test]
    fn test_registry_snapshot_serialization() {
        let registry = ChannelRegistry::new();
        registry.register(ChannelInfo::new("telegram", "Telegram"));

        let snapshot = registry.snapshot();
        let json = serde_json::to_string(&snapshot).unwrap();
        let parsed: RegistrySnapshot = serde_json::from_str(&json).unwrap();

        assert_eq!(parsed.channels.len(), 1);
        assert!(parsed.timestamp > 0);
    }
}
