//! timer:* — periodic tick consumer.
//!
//! URI: `timer:<name>?period_millis=1000` (default 1000ms).
//!
//! Consumer-only. On `start()`, spawns a tokio task that emits an Exchange
//! (with header `CamelTimerFiredTime`) every `period`. `stop()` aborts the
//! task.

use async_trait::async_trait;
use coreon_core::{
    uri::CamelUri, CamelError, Component, Consumer, Endpoint, Exchange, Message, Processor,
    Producer, Result,
};
use dashmap::DashMap;
use std::{
    sync::{
        atomic::{AtomicU64, Ordering},
        Arc,
    },
    time::Duration,
};
use tokio::{sync::Mutex, task::JoinHandle};
use tracing::debug;

pub struct TimerComponent {
    endpoints: DashMap<String, Arc<TimerEndpoint>>,
}

impl TimerComponent {
    pub fn new() -> Arc<Self> {
        Arc::new(Self {
            endpoints: DashMap::new(),
        })
    }
}

impl Default for TimerComponent {
    fn default() -> Self {
        Self {
            endpoints: DashMap::new(),
        }
    }
}

#[async_trait]
impl Component for TimerComponent {
    fn scheme(&self) -> &str {
        "timer"
    }

    async fn create_endpoint(&self, uri: &CamelUri) -> Result<Arc<dyn Endpoint>> {
        let key = uri.as_str().to_owned();
        if let Some(ep) = self.endpoints.get(&key) {
            return Ok(ep.clone());
        }
        let period_ms = uri
            .get_param("period_millis")
            .map(str::parse::<u64>)
            .transpose()
            .map_err(|e| CamelError::InvalidUri {
                uri: key.clone(),
                reason: format!("period_millis: {e}"),
            })?
            .unwrap_or(1000);
        let ep = Arc::new(TimerEndpoint {
            uri: key.clone(),
            name: uri.path.clone(),
            period: Duration::from_millis(period_ms),
        });
        self.endpoints.insert(key, ep.clone());
        Ok(ep)
    }
}

pub struct TimerEndpoint {
    uri: String,
    name: String,
    period: Duration,
}

#[async_trait]
impl Endpoint for TimerEndpoint {
    fn uri(&self) -> &str {
        &self.uri
    }

    async fn create_producer(&self) -> Result<Arc<dyn Producer>> {
        Err(CamelError::Endpoint(
            "timer: does not support producers (source-only endpoint)".into(),
        ))
    }

    async fn create_consumer(&self, pipeline: Arc<dyn Processor>) -> Result<Arc<dyn Consumer>> {
        Ok(Arc::new(TimerConsumer {
            name: self.name.clone(),
            period: self.period,
            pipeline,
            task: Mutex::new(None),
        }))
    }
}

pub struct TimerConsumer {
    name: String,
    period: Duration,
    pipeline: Arc<dyn Processor>,
    task: Mutex<Option<JoinHandle<()>>>,
}

#[async_trait]
impl Consumer for TimerConsumer {
    async fn start(&self) -> Result<()> {
        let pipeline = self.pipeline.clone();
        let period = self.period;
        let name = self.name.clone();
        let counter = Arc::new(AtomicU64::new(0));
        let handle = tokio::spawn(async move {
            let mut ticker = tokio::time::interval(period);
            // First tick fires immediately; skip so behavior matches Camel's
            // "fireCount starts at 1 after first period".
            ticker.tick().await;
            loop {
                ticker.tick().await;
                let n = counter.fetch_add(1, Ordering::Relaxed) + 1;
                let msg = Message::default()
                    .with_header("CamelTimerName", &name)
                    .with_header("CamelTimerCounter", n.to_string());
                let mut ex = Exchange {
                    r#in: msg,
                    ..Exchange::default()
                };
                if let Err(e) = pipeline.process(&mut ex).await {
                    debug!(timer = %name, error = %e, "timer tick pipeline failed");
                }
            }
        });
        *self.task.lock().await = Some(handle);
        Ok(())
    }

    async fn stop(&self) -> Result<()> {
        if let Some(h) = self.task.lock().await.take() {
            h.abort();
        }
        Ok(())
    }
}
