use std::collections::BTreeMap;
use std::sync::{Arc, Mutex};
use std::time::Duration as StdDuration;

use async_trait::async_trait;
use harn_clock::{Clock, RealClock};
use serde::{Deserialize, Serialize};
use serde_json::{json, Value as JsonValue};
use time::OffsetDateTime;
use tokio::task::JoinHandle;

use crate::connectors::{
    ActivationHandle, ClientError, Connector, ConnectorClient, ConnectorCtx, ConnectorError,
    ProviderPayloadSchema, RawInbound, TriggerBinding, TriggerKind,
};
use crate::event_log::{AnyEventLog, EventLog, LogEvent, Topic};
use crate::triggers::event::KnownProviderPayload;
use crate::triggers::{
    CronEventPayload, ProviderId, ProviderPayload, SignatureStatus, TriggerEvent,
    DEFAULT_INBOX_RETENTION_DAYS,
};

use self::scheduler::{run_tick_loop, CronSchedule, ShutdownSignal, TickHandler};
use self::state::{CronStateStore, PersistedCronState};

pub(crate) mod scheduler;
pub(crate) mod state;

#[cfg(test)]
mod tests;

pub const CRON_TICK_TOPIC: &str = "connectors.cron.tick";
const TEST_FAIL_AFTER_EMIT_ENV: &str = "HARN_TEST_CRON_FAIL_AFTER_EMIT";
const TEST_SINGLE_TICK_AT_ENV: &str = "HARN_TEST_CRON_SINGLE_TICK_AT";

#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CatchupMode {
    #[default]
    Skip,
    All,
    Latest,
}

pub struct CronConnector {
    provider_id: ProviderId,
    kinds: Vec<TriggerKind>,
    client: Arc<CronClient>,
    ctx: Mutex<Option<ConnectorCtx>>,
    clock: Arc<dyn Clock>,
    sink_override: Option<Arc<dyn CronEventSink>>,
    tasks: Mutex<Vec<JoinHandle<()>>>,
    shutdown: Mutex<Arc<ShutdownSignal>>,
}

impl CronConnector {
    pub fn new() -> Self {
        Self::with_clock(Arc::new(RealClock::new()))
    }

    /// Build a connector backed by an injected clock. Production callers use
    /// [`harn_clock::RealClock`]; tests use [`harn_clock::PausedClock`].
    pub fn with_clock(clock: Arc<dyn Clock>) -> Self {
        Self {
            provider_id: ProviderId::from("cron"),
            kinds: vec![TriggerKind::from("cron")],
            client: Arc::new(CronClient),
            ctx: Mutex::new(None),
            clock,
            sink_override: None,
            tasks: Mutex::new(Vec::new()),
            shutdown: Mutex::new(Arc::new(ShutdownSignal::default())),
        }
    }

    pub(crate) fn with_clock_and_sink(clock: Arc<dyn Clock>, sink: Arc<dyn CronEventSink>) -> Self {
        let mut connector = Self::with_clock(clock);
        connector.sink_override = Some(sink);
        connector
    }

    fn context(&self) -> Result<ConnectorCtx, ConnectorError> {
        self.ctx
            .lock()
            .expect("cron connector context mutex poisoned")
            .clone()
            .ok_or_else(|| {
                ConnectorError::Activation(
                    "cron connector must be initialized before activation".to_string(),
                )
            })
    }

    fn request_stop(&self) {
        self.shutdown
            .lock()
            .expect("cron connector shutdown mutex poisoned")
            .request_stop();
    }

    fn take_tasks(&self) -> Vec<JoinHandle<()>> {
        self.tasks
            .lock()
            .expect("cron connector tasks mutex poisoned")
            .drain(..)
            .collect()
    }
}

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

impl Drop for CronConnector {
    fn drop(&mut self) {
        self.request_stop();
        let mut tasks = self
            .tasks
            .lock()
            .expect("cron connector tasks mutex poisoned");
        for task in tasks.drain(..) {
            task.abort();
        }
    }
}

#[async_trait]
impl Connector for CronConnector {
    fn provider_id(&self) -> &ProviderId {
        &self.provider_id
    }

    fn kinds(&self) -> &[TriggerKind] {
        &self.kinds
    }

    async fn init(&mut self, ctx: ConnectorCtx) -> Result<(), ConnectorError> {
        *self
            .ctx
            .lock()
            .expect("cron connector context mutex poisoned") = Some(ctx);
        Ok(())
    }

    async fn activate(
        &self,
        bindings: &[TriggerBinding],
    ) -> Result<ActivationHandle, ConnectorError> {
        let ctx = self.context()?;
        let shutdown = Arc::new(ShutdownSignal::default());
        *self
            .shutdown
            .lock()
            .expect("cron connector shutdown mutex poisoned") = shutdown.clone();
        let state_store = Arc::new(CronStateStore::new(ctx.event_log.clone()));
        let sink: Arc<dyn CronEventSink> = self.sink_override.clone().unwrap_or_else(|| {
            Arc::new(EventLogCronEventSink::new(
                ctx.event_log.clone(),
                ctx.inbox.clone(),
            ))
        });
        {
            let mut tasks = self
                .tasks
                .lock()
                .expect("cron connector tasks mutex poisoned");
            for task in tasks.drain(..) {
                task.abort();
            }
        }

        for binding in bindings {
            let trigger = CronTrigger::from_binding(binding)?;
            let clock = self.clock.clone();
            let sink = sink.clone();
            let state_store = state_store.clone();
            let last_fired = state_store
                .load(&trigger.trigger_id)
                .await?
                .map(|state| state.last_fired_at);
            let now = clock.now_utc();
            let catchup_ticks = match trigger.catchup_mode {
                CatchupMode::Skip => Vec::new(),
                CatchupMode::All => trigger.schedule.due_ticks_between(last_fired, now)?,
                CatchupMode::Latest => trigger
                    .schedule
                    .due_ticks_between(last_fired, now)?
                    .into_iter()
                    .last()
                    .into_iter()
                    .collect(),
            };
            let cursor = match trigger.catchup_mode {
                CatchupMode::Skip => now,
                _ => last_fired.unwrap_or(now),
            };
            let handler = Arc::new(CronTaskHandler {
                trigger,
                sink,
                state_store,
            });
            if let Some(tick_at) = test_single_tick_at()? {
                let task = tokio::spawn(async move {
                    let _ = handler.on_tick(tick_at, false).await;
                });
                self.tasks
                    .lock()
                    .expect("cron connector tasks mutex poisoned")
                    .push(task);
                continue;
            }
            let shutdown = shutdown.clone();
            let task = tokio::spawn(async move {
                let _ = run_tick_loop(
                    handler.trigger.schedule.clone(),
                    clock,
                    cursor,
                    catchup_ticks,
                    handler,
                    shutdown,
                )
                .await;
            });
            self.tasks
                .lock()
                .expect("cron connector tasks mutex poisoned")
                .push(task);
        }

        Ok(ActivationHandle::new(
            self.provider_id.clone(),
            bindings.len(),
        ))
    }

    async fn shutdown(&self, deadline: StdDuration) -> Result<(), ConnectorError> {
        self.request_stop();
        let pending = self.take_tasks();

        if pending.is_empty() {
            return Ok(());
        }

        let wait_all = async {
            for task in pending {
                let _ = task.await;
            }
        };
        tokio::time::timeout(deadline, wait_all)
            .await
            .map_err(|_| {
                ConnectorError::Activation(format!(
                    "cron connector shutdown exceeded {}s",
                    deadline.as_secs()
                ))
            })?;
        Ok(())
    }

    async fn normalize_inbound(&self, _raw: RawInbound) -> Result<TriggerEvent, ConnectorError> {
        Err(ConnectorError::Unsupported(
            "cron is an in-process scheduler and does not accept inbound payloads".to_string(),
        ))
    }

    fn payload_schema(&self) -> ProviderPayloadSchema {
        ProviderPayloadSchema::named("CronEventPayload")
    }

    fn client(&self) -> Arc<dyn ConnectorClient> {
        self.client.clone()
    }
}

#[derive(Debug)]
struct CronClient;

#[async_trait]
impl ConnectorClient for CronClient {
    async fn call(&self, method: &str, _args: JsonValue) -> Result<JsonValue, ClientError> {
        Err(ClientError::MethodNotFound(format!(
            "cron connector does not implement outbound method '{method}'"
        )))
    }
}

#[async_trait]
pub(crate) trait CronEventSink: Send + Sync {
    async fn emit(
        &self,
        binding_id: &str,
        retention: StdDuration,
        event: TriggerEvent,
    ) -> Result<(), ConnectorError>;
}

struct EventLogCronEventSink {
    event_log: Arc<AnyEventLog>,
    inbox: Arc<crate::triggers::InboxIndex>,
    topic: Topic,
}

impl EventLogCronEventSink {
    fn new(event_log: Arc<AnyEventLog>, inbox: Arc<crate::triggers::InboxIndex>) -> Self {
        Self {
            event_log,
            inbox,
            topic: Topic::new(CRON_TICK_TOPIC).expect("cron tick topic is valid"),
        }
    }
}

#[async_trait]
impl CronEventSink for EventLogCronEventSink {
    async fn emit(
        &self,
        binding_id: &str,
        retention: StdDuration,
        mut event: TriggerEvent,
    ) -> Result<(), ConnectorError> {
        if !self
            .inbox
            .insert_if_new(binding_id, &event.dedupe_key, retention)
            .await?
        {
            return Ok(());
        }
        let dedupe_key = event.dedupe_key.clone();
        event.mark_dedupe_claimed();
        let event = match crate::connectors::postprocess_normalized_event(
            self.inbox.as_ref(),
            binding_id,
            true,
            retention,
            event,
        )
        .await?
        {
            crate::connectors::PostNormalizeOutcome::Ready(event) => *event,
            crate::connectors::PostNormalizeOutcome::DuplicateDropped => {
                return Err(ConnectorError::Activation(format!(
                    "cron event `{dedupe_key}` for binding `{binding_id}` was deduped twice"
                )));
            }
        };
        let payload = serde_json::to_value(&event).map_err(ConnectorError::from)?;
        self.event_log
            .append(&self.topic, LogEvent::new("trigger_event", payload))
            .await
            .map_err(ConnectorError::from)?;
        Ok(())
    }
}

#[derive(Clone)]
struct CronTaskHandler {
    trigger: CronTrigger,
    sink: Arc<dyn CronEventSink>,
    state_store: Arc<CronStateStore>,
}

#[async_trait]
impl TickHandler for CronTaskHandler {
    async fn on_tick(&self, tick_at: OffsetDateTime, catchup: bool) -> Result<(), ConnectorError> {
        let event = self.trigger.to_event(tick_at, catchup);
        self.sink
            .emit(
                &self.trigger.trigger_id,
                self.trigger.dedupe_retention,
                event,
            )
            .await?;
        maybe_fail_after_emit();
        self.state_store
            .persist(PersistedCronState {
                trigger_id: self.trigger.trigger_id.clone(),
                last_fired_at: tick_at,
            })
            .await?;
        Ok(())
    }
}

#[derive(Clone)]
struct CronTrigger {
    trigger_id: String,
    schedule_raw: String,
    timezone_raw: String,
    schedule: CronSchedule,
    catchup_mode: CatchupMode,
    dedupe_retention: StdDuration,
}

impl CronTrigger {
    fn from_binding(binding: &TriggerBinding) -> Result<Self, ConnectorError> {
        let config: CronBindingConfig =
            serde_json::from_value(binding.config.clone()).map_err(ConnectorError::from)?;
        let timezone = parse_iana_timezone(&config.timezone)?;
        Ok(Self {
            trigger_id: binding.binding_id.clone(),
            schedule_raw: config.schedule.clone(),
            timezone_raw: config.timezone.clone(),
            schedule: CronSchedule::parse(config.schedule, timezone)?,
            catchup_mode: config.catchup_mode,
            dedupe_retention: StdDuration::from_secs(
                u64::from(config.retention_days.max(1)) * 24 * 60 * 60,
            ),
        })
    }

    fn to_event(&self, tick_at: OffsetDateTime, catchup: bool) -> TriggerEvent {
        let payload = ProviderPayload::Known(KnownProviderPayload::Cron(CronEventPayload {
            cron_id: Some(self.trigger_id.clone()),
            schedule: Some(self.schedule_raw.clone()),
            tick_at,
            raw: json!({
                "catchup": catchup,
                "timezone": self.timezone_raw,
            }),
        }));
        TriggerEvent::new(
            ProviderId::from("cron"),
            "tick",
            Some(tick_at),
            format!("cron:{}:{}", self.trigger_id, tick_at.unix_timestamp()),
            None,
            BTreeMap::new(),
            payload,
            SignatureStatus::Unsigned,
        )
    }
}

#[derive(Clone, Debug, Deserialize)]
struct CronBindingConfig {
    schedule: String,
    timezone: String,
    #[serde(default, alias = "catch_up", alias = "catchup")]
    catchup_mode: CatchupMode,
    #[serde(default = "default_retention_days")]
    retention_days: u32,
}

fn default_retention_days() -> u32 {
    DEFAULT_INBOX_RETENTION_DAYS
}

fn parse_iana_timezone(raw: &str) -> Result<chrono_tz::Tz, ConnectorError> {
    if looks_like_utc_offset(raw) {
        return Err(ConnectorError::Activation(format!(
            "invalid cron timezone '{raw}': use an IANA timezone name like 'America/New_York', not a UTC offset"
        )));
    }
    raw.parse::<chrono_tz::Tz>().map_err(|error| {
        ConnectorError::Activation(format!("invalid cron timezone '{raw}': {error}"))
    })
}

pub(crate) fn looks_like_utc_offset(raw: &str) -> bool {
    let value = raw.trim();
    if let Some(rest) = value
        .strip_prefix("UTC")
        .or_else(|| value.strip_prefix("utc"))
        .or_else(|| value.strip_prefix("GMT"))
        .or_else(|| value.strip_prefix("gmt"))
    {
        return rest.starts_with('+') || rest.starts_with('-');
    }
    let chars: Vec<char> = value.chars().collect();
    if chars.len() < 3 || !matches!(chars[0], '+' | '-') {
        return false;
    }
    chars[1..]
        .iter()
        .all(|ch| ch.is_ascii_digit() || *ch == ':')
}

fn maybe_fail_after_emit() {
    if std::env::var_os(TEST_FAIL_AFTER_EMIT_ENV).is_some() {
        std::process::exit(86);
    }
}

fn test_single_tick_at() -> Result<Option<OffsetDateTime>, ConnectorError> {
    let Some(raw) = std::env::var_os(TEST_SINGLE_TICK_AT_ENV) else {
        return Ok(None);
    };
    let raw = raw.to_string_lossy();
    let timestamp = raw.parse::<i64>().map_err(|error| {
        ConnectorError::Activation(format!(
            "invalid {TEST_SINGLE_TICK_AT_ENV} unix timestamp '{raw}': {error}"
        ))
    })?;
    OffsetDateTime::from_unix_timestamp(timestamp)
        .map(Some)
        .map_err(|error| {
            ConnectorError::Activation(format!(
                "invalid {TEST_SINGLE_TICK_AT_ENV} unix timestamp '{raw}': {error}"
            ))
        })
}
