use std::collections::HashMap;

use common_utils::types::TenantConfig;
use error_stack::ResultExt;
use events::{EventsError, Message, MessagingInterface};
use hyperswitch_interfaces::events as events_interfaces;
use hyperswitch_masking::ErasedMaskSerialize;
use router_env::logger;
use serde::{Deserialize, Serialize};
use storage_impl::errors::{ApplicationError, StorageError, StorageResult};
use time::PrimitiveDateTime;

use crate::{
    db::KafkaProducer,
    services::kafka::{KafkaMessage, KafkaSettings},
};

pub mod api_logs;
pub mod audit_events;
pub mod connector_api_logs;
pub mod event_logger;
pub mod external_service_call;
pub mod outgoing_webhook_logs;
pub mod routing_api_logs;
#[derive(Debug, Serialize, Clone, Copy)]
#[serde(rename_all = "snake_case")]
pub enum EventType {
    PaymentIntent,
    FraudCheck,
    PaymentAttempt,
    Refund,
    ApiLogs,
    ConnectorApiLogs,
    OutgoingWebhookLogs,
    Dispute,
    AuditEvent,
    #[cfg(feature = "payouts")]
    Payout,
    Consolidated,
    Authentication,
    RoutingApiLogs,
    RevenueRecovery,
    ExternalServiceCall,
}

#[derive(Debug, Default, Deserialize, Clone)]
pub struct EventsConfig {
    #[serde(flatten)]
    pub source: EventsSource,
    #[serde(default)]
    pub emit_external_service_call_events: bool,
}

#[derive(Debug, Default, Deserialize, Clone)]
#[serde(tag = "source")]
#[serde(rename_all = "lowercase")]
pub enum EventsSource {
    Kafka {
        kafka: Box<KafkaSettings>,
    },
    #[default]
    Logs,
}

#[allow(clippy::large_enum_variant)]
#[derive(Debug, Clone)]
pub enum EventsHandler {
    Kafka(KafkaProducer),
    Logs(event_logger::EventLogger),
}

impl Default for EventsHandler {
    fn default() -> Self {
        Self::Logs(event_logger::EventLogger {})
    }
}

impl events_interfaces::EventHandlerInterface for EventsHandler {
    fn log_connector_event(&self, event: &events_interfaces::connector_api_logs::ConnectorEvent) {
        self.log_event(event);
    }
}

impl common_utils::external_service::ExternalServiceEventEmitter for EventsHandler {
    fn emit_external_service_call(
        &self,
        event: common_utils::external_service::ExternalServiceCall,
    ) {
        self.log_event(&external_service_call::KafkaExternalServiceCall { event: &event });
    }
}

impl EventsConfig {
    pub async fn get_event_handler(&self) -> StorageResult<EventsHandler> {
        Ok(match &self.source {
            EventsSource::Kafka { kafka } => EventsHandler::Kafka(
                KafkaProducer::create(kafka)
                    .await
                    .change_context(StorageError::InitializationError)?,
            ),
            EventsSource::Logs => EventsHandler::Logs(event_logger::EventLogger::default()),
        })
    }

    pub fn validate(&self) -> Result<(), ApplicationError> {
        match &self.source {
            EventsSource::Kafka { kafka } => {
                kafka.validate()?;
                if self.emit_external_service_call_events {
                    kafka.validate_external_service_call_topic()?;
                }
                Ok(())
            }
            EventsSource::Logs => Ok(()),
        }
    }
}

impl EventsHandler {
    pub fn log_event<T: KafkaMessage>(&self, event: &T) {
        match self {
            Self::Kafka(kafka) => kafka.log_event(event).unwrap_or_else(|e| {
                logger::error!("Failed to log event: {:?}", e);
            }),
            Self::Logs(logger) => logger.log_event(event),
        };
    }
    pub fn add_tenant(&mut self, tenant_config: &dyn TenantConfig) {
        if let Self::Kafka(kafka_producer) = self {
            kafka_producer.set_tenancy(tenant_config);
        }
    }
}

impl MessagingInterface for EventsHandler {
    type MessageClass = EventType;

    fn send_message<T>(
        &self,
        data: T,
        metadata: HashMap<String, String>,
        timestamp: PrimitiveDateTime,
    ) -> error_stack::Result<(), EventsError>
    where
        T: Message<Class = Self::MessageClass> + ErasedMaskSerialize,
    {
        match self {
            Self::Kafka(a) => a.send_message(data, metadata, timestamp),
            Self::Logs(a) => a.send_message(data, metadata, timestamp),
        }
    }
}
