use std::sync::Arc;

use arc_swap::ArcSwap;
use subtle::ConstantTimeEq;

use crate::config::{RuntimeConfig, ServerConfig, SharedConfig};
use crate::error::{Error, Result};
use crate::types::{ApiKey, NamespaceId, Permission};

use super::permissions::PermissionSet;
use super::{AuthSession, Authenticator};

/// A namespace and its granted permissions, stored alongside the raw key bytes.
struct KeyEntry {
    namespace: NamespaceId,
    permissions: PermissionSet,
    key_bytes: Vec<u8>,
}

/// In-memory API key store with hot-reload support.
///
/// Backed by `ArcSwap<Vec<KeyEntry>>` for lock-free reads with atomic swaps
/// on config reload. Uses linear scan with constant-time key comparison to
/// prevent timing attacks.
pub struct ApiKeyStore {
    entries: ArcSwap<Vec<KeyEntry>>,
}

impl ApiKeyStore {
    /// Build the key store from server configuration.
    pub fn from_config(config: &ServerConfig) -> Self {
        let entries = Self::build_entries_from_config(config);
        Self {
            entries: ArcSwap::from_pointee(entries),
        }
    }

    /// Build entries from a ServerConfig.
    fn build_entries_from_config(config: &ServerConfig) -> Vec<KeyEntry> {
        config
            .namespaces
            .iter()
            .map(|(ns_id, ns_config)| KeyEntry {
                namespace: ns_id.clone(),
                permissions: PermissionSet::from_slice(&ns_config.permissions),
                key_bytes: ns_config.api_key.0.as_bytes().to_vec(),
            })
            .collect()
    }

    /// Rebuild entries from a RuntimeConfig snapshot (used during hot-reload).
    fn build_entries_from_runtime_config(config: &RuntimeConfig) -> Vec<KeyEntry> {
        config
            .namespace_auth
            .iter()
            .map(|(ns_id, auth)| KeyEntry {
                namespace: ns_id.clone(),
                permissions: PermissionSet::from_slice(&auth.permissions),
                key_bytes: auth.api_key.0.as_bytes().to_vec(),
            })
            .collect()
    }

    /// Rebuild the auth entries from the current SharedConfig.
    /// Called when auth changes are detected during config reload.
    pub fn reload_from_config(&self, shared_config: &SharedConfig) {
        let cfg = shared_config.load();
        let entries = Self::build_entries_from_runtime_config(&cfg);
        let count = entries.len();
        self.entries.store(Arc::new(entries));
        tracing::info!(
            namespaces = count,
            "auth store rebuilt from reloaded config"
        );
    }

    /// Spawn a background task that watches for reload notifications and
    /// rebuilds the auth store when API keys or permissions change.
    pub fn spawn_reload_watcher(
        self: &Arc<Self>,
        shared_config: SharedConfig,
        mut reload_rx: tokio::sync::watch::Receiver<Option<crate::config::ReloadNotification>>,
    ) -> tokio::task::JoinHandle<()> {
        let store = Arc::clone(self);
        tokio::spawn(async move {
            while reload_rx.changed().await.is_ok() {
                let notification = reload_rx.borrow().clone();
                if let Some(ref notif) = notification
                    && notif.auth_changed
                {
                    store.reload_from_config(&shared_config);
                }
            }
        })
    }

    /// Find and return session info for the matching key.
    fn find_session(&self, key: &ApiKey) -> Option<(NamespaceId, PermissionSet)> {
        let entries = self.entries.load_full();
        let input = key.0.as_bytes();
        entries
            .iter()
            .find(|entry| input.ct_eq(&entry.key_bytes).into())
            .map(|entry| (entry.namespace.clone(), entry.permissions))
    }
}

impl Authenticator for ApiKeyStore {
    fn authenticate(&self, key: &ApiKey) -> Result<AuthSession> {
        match self.find_session(key) {
            Some((namespace, permissions)) => {
                tracing::debug!(
                    namespace = %namespace,
                    "authentication successful"
                );
                Ok(AuthSession {
                    namespace,
                    permissions,
                })
            }
            None => {
                tracing::debug!("authentication failed: unknown API key");
                metrics::counter!("fafb_auth_failures_total", "reason" => "unknown_key")
                    .increment(1);
                Err(Error::AuthFailed)
            }
        }
    }

    fn check_permission(
        &self,
        namespace: &NamespaceId,
        key: &ApiKey,
        perm: Permission,
    ) -> Result<()> {
        let (entry_ns, entry_perms) = self.find_session(key).ok_or_else(|| {
            tracing::debug!(namespace = %namespace, "permission check failed: unknown API key");
            metrics::counter!("fafb_auth_failures_total", "reason" => "unknown_key").increment(1);
            Error::AuthFailed
        })?;
        if entry_ns != *namespace {
            tracing::debug!(
                requested_namespace = %namespace,
                key_namespace = %entry_ns,
                "permission check failed: namespace mismatch"
            );
            metrics::counter!("fafb_auth_failures_total", "reason" => "namespace_mismatch")
                .increment(1);
            return Err(Error::AuthFailed);
        }
        entry_perms.check(perm)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::config::NamespaceConfig;
    use std::collections::HashMap;

    /// Minimal ServerConfig with no namespaces (all fields use serde defaults).
    fn empty_config() -> ServerConfig {
        toml::from_str("").unwrap()
    }

    /// Config with two namespaces for testing.
    fn test_config() -> ServerConfig {
        let mut namespaces = HashMap::new();
        namespaces.insert(
            NamespaceId("production".into()),
            NamespaceConfig {
                api_key: ApiKey("prod-key-abcdefghijklmnop".into()),
                permissions: vec![
                    Permission::Subscribe,
                    Permission::Publish,
                    Permission::Query,
                ],
                max_redelivery_attempts: 5,
                dlq_topic_suffix: "__dlq".to_owned(),
                max_total_messages: None,
                max_topics: None,
                topic_max_depth: None,
                topic_max_bytes: None,
                backpressure_policy: None,
                max_connections: None,
                max_disk_bytes: None,
                retention: Default::default(),
                default_ttl_secs: 0,
                max_ttl_secs: 0,
                min_ttl_secs: 0,
            },
        );
        namespaces.insert(
            NamespaceId("analytics".into()),
            NamespaceConfig {
                api_key: ApiKey("analytics-key-123456789012".into()),
                permissions: vec![Permission::Publish, Permission::Query],
                max_redelivery_attempts: 5,
                dlq_topic_suffix: "__dlq".to_owned(),
                max_total_messages: None,
                max_topics: None,
                topic_max_depth: None,
                topic_max_bytes: None,
                backpressure_policy: None,
                max_connections: None,
                max_disk_bytes: None,
                retention: Default::default(),
                default_ttl_secs: 0,
                max_ttl_secs: 0,
                min_ttl_secs: 0,
            },
        );
        ServerConfig {
            namespaces,
            ..empty_config()
        }
    }

    // -- authenticate tests --

    #[test]
    fn authenticate_valid_key_returns_session() {
        let store = ApiKeyStore::from_config(&test_config());
        let session = store
            .authenticate(&ApiKey("prod-key-abcdefghijklmnop".into()))
            .unwrap();
        assert_eq!(session.namespace, NamespaceId("production".into()));
        assert!(session.permissions.contains(Permission::Subscribe));
        assert!(session.permissions.contains(Permission::Publish));
        assert!(session.permissions.contains(Permission::Query));
    }

    #[test]
    fn authenticate_returns_correct_namespace_and_permissions() {
        let store = ApiKeyStore::from_config(&test_config());
        let session = store
            .authenticate(&ApiKey("analytics-key-123456789012".into()))
            .unwrap();
        assert_eq!(session.namespace, NamespaceId("analytics".into()));
        assert!(!session.permissions.contains(Permission::Subscribe));
        assert!(session.permissions.contains(Permission::Publish));
        assert!(session.permissions.contains(Permission::Query));
    }

    #[test]
    fn authenticate_invalid_key() {
        let store = ApiKeyStore::from_config(&test_config());
        let err = store
            .authenticate(&ApiKey("wrong-key-xxxxxxxxxxxxxxxx".into()))
            .unwrap_err();
        assert!(matches!(err, Error::AuthFailed));
    }

    #[test]
    fn authenticate_wrong_length_key() {
        let store = ApiKeyStore::from_config(&test_config());
        // Shorter than any configured key
        assert!(
            store
                .authenticate(&ApiKey("short-24-chars-xxxxxxxxx".into()))
                .is_err()
        );
        // Longer than any configured key
        assert!(
            store
                .authenticate(&ApiKey(
                    "much-longer-key-that-exceeds-all-configured-keys-in-store".into()
                ))
                .is_err()
        );
    }

    // -- check_permission tests --

    #[test]
    fn check_permission_granted() {
        let store = ApiKeyStore::from_config(&test_config());
        let ns = NamespaceId("production".into());
        let key = ApiKey("prod-key-abcdefghijklmnop".into());
        assert!(
            store
                .check_permission(&ns, &key, Permission::Subscribe)
                .is_ok()
        );
        assert!(
            store
                .check_permission(&ns, &key, Permission::Publish)
                .is_ok()
        );
        assert!(store.check_permission(&ns, &key, Permission::Query).is_ok());
    }

    #[test]
    fn check_permission_denied() {
        let store = ApiKeyStore::from_config(&test_config());
        let ns = NamespaceId("analytics".into());
        let key = ApiKey("analytics-key-123456789012".into());
        let err = store
            .check_permission(&ns, &key, Permission::Subscribe)
            .unwrap_err();
        assert!(matches!(err, Error::PermissionDenied));
    }

    #[test]
    fn check_permission_invalid_key() {
        let store = ApiKeyStore::from_config(&test_config());
        let ns = NamespaceId("production".into());
        let key = ApiKey("wrong-key-xxxxxxxxxxxxxxxx".into());
        let err = store
            .check_permission(&ns, &key, Permission::Subscribe)
            .unwrap_err();
        assert!(matches!(err, Error::AuthFailed));
    }

    #[test]
    fn check_permission_key_namespace_mismatch() {
        let store = ApiKeyStore::from_config(&test_config());
        // Valid key for "production", but claiming "analytics"
        let ns = NamespaceId("analytics".into());
        let key = ApiKey("prod-key-abcdefghijklmnop".into());
        let err = store
            .check_permission(&ns, &key, Permission::Publish)
            .unwrap_err();
        assert!(matches!(err, Error::AuthFailed));
    }

    // -- edge cases --

    #[test]
    fn empty_config_rejects_all_keys() {
        let store = ApiKeyStore::from_config(&empty_config());
        let err = store
            .authenticate(&ApiKey("anything-at-all-123456789".into()))
            .unwrap_err();
        assert!(matches!(err, Error::AuthFailed));
    }

    #[test]
    fn api_key_store_is_send_and_sync() {
        fn assert_authenticator<T: Authenticator>() {}
        assert_authenticator::<ApiKeyStore>();
    }
}
