// r8r: rust port of n8n.io
// Initial author: Michael Cochran <mcochran@cochranblock.org>
// Built for: Tanner Woodrum, intended cofounder
// License: Unlicense (public domain dedication, see UNLICENSE)
//
// `cron` — trigger entry node. Blocks until the next firing of a cron
// expression, emits a single item with the firing timestamp, then exits.
// v0.1: one-shot. Long-lived recurring listener is v0.2.

use async_trait::async_trait;
use crate::core::error::NodeError;
use crate::core::types::*;
use crate::engine::{ExecuteCtx, Node, NodeFactory};
use serde_json::json;
use std::sync::Arc;
use tokio::sync::{oneshot, Mutex};
use tokio_cron_scheduler::{Job, JobScheduler};

pub struct CronNode;

static META: NodeMetadata = NodeMetadata {
    name: "cron",
    display_name: "Cron",
    version: &[1],
    group: &["trigger"],
    description: "Triggers the workflow on the next firing of a cron expression",
    inputs: 0,
    outputs: 1,
    properties: &[],
    credentials: &[],
};

#[async_trait]
impl Node for CronNode {
    fn metadata(&self) -> &'static NodeMetadata { &META }

    async fn execute(&self, ctx: &mut ExecuteCtx<'_>) -> Result<NodeOutput, NodeError> {
        let cron_expr = ctx.param_str("cron_expr")?;

        let (tx, rx) = oneshot::channel::<chrono::DateTime<chrono::Utc>>();
        let shared = Arc::new(Mutex::new(Some(tx)));

        let mut sched = JobScheduler::new().await
            .map_err(|e| NodeError::Other(format!("cron init: {}", e)))?;

        let job_shared = shared.clone();
        let job = Job::new_async(cron_expr.as_str(), move |_uuid, _l| {
            let shared = job_shared.clone();
            Box::pin(async move {
                if let Some(tx) = shared.lock().await.take() {
                    let _ = tx.send(chrono::Utc::now());
                }
            })
        }).map_err(|e| NodeError::Other(format!("cron parse `{}`: {}", cron_expr, e)))?;

        sched.add(job).await.map_err(|e| NodeError::Other(format!("cron add: {}", e)))?;
        sched.start().await.map_err(|e| NodeError::Other(format!("cron start: {}", e)))?;

        tracing::info!(cron = %cron_expr, "cron waiting for next firing");

        let fired_at = rx.await
            .map_err(|e| NodeError::Other(format!("cron recv: {}", e)))?;
        let _ = sched.shutdown().await;

        let item = NodeItem {
            json: json!({ "fired_at": fired_at.to_rfc3339() }),
            ..Default::default()
        };
        Ok(vec![vec![item]])
    }
}

inventory::submit! { NodeFactory::new("cron", || Arc::new(CronNode)) }
