pub(crate) mod cli;
pub(crate) mod extractors;
#[cfg(feature = "source_http_polling")]
pub(crate) mod http_polling;
#[cfg(feature = "source_kafka")]
pub(crate) mod kafka;
#[cfg(feature = "source_nats")]
pub(crate) mod nats;
#[cfg(feature = "source_opendal")]
pub(crate) mod opendal;
pub(crate) mod parsers;
pub(crate) mod send_cdevents;
#[cfg(feature = "source_sse")]
pub(crate) mod sse;
pub(crate) mod subprocess;
pub(crate) mod webhook;

use crate::cdevent_utils::sanitize_id;
use crate::errors::{Error, IntoDiagnostic, Result};
use crate::transformers;
use crate::{Message, Sender};
use axum::Router;
use cdevents_sdk::CDEvent;
use serde::Deserialize;
use std::collections::HashMap;
use std::sync::Arc;
use tokio::task::JoinHandle;
use tokio_util::sync::CancellationToken;

type SourceHandlesAndRoutes = (Vec<JoinHandle<Result<()>>>, Vec<Router>);

// TODO support name/reference for extractor / transformer
#[derive(Clone, Debug, Deserialize, Default)]
pub(crate) struct Config {
    #[serde(default)]
    enabled: bool,
    #[serde(default)]
    extractor: extractors::Config,
    #[serde(flatten)]
    pub(crate) chain: transformers::TransformerChainConfig,
}

impl Config {
    pub fn is_enabled(&self) -> bool {
        self.enabled
    }

    pub fn resolve_transformers(
        &mut self,
        configs: &HashMap<String, transformers::Config>,
    ) -> Result<()> {
        self.chain.resolve(configs)
    }

    pub(crate) fn append_transformers(&mut self, extra: &[transformers::Config]) {
        self.chain.append(extra);
    }

    pub(crate) fn inject_subprocess_exit_code_out(
        &mut self,
        exit_code_out: Arc<std::sync::atomic::AtomicI32>,
    ) {
        if let extractors::Config::Subprocess(sub) = &mut self.extractor {
            sub.exit_code_out = exit_code_out;
        }
    }
}

pub(crate) fn make(
    name: &str,
    config: &Config,
    tx: Sender<Message>,
    cancel_token: CancellationToken,
    state_config: Option<&crate::state::Config>,
    root_url: &url::Url,
) -> Result<extractors::Extractor> {
    let mut default_source = root_url.clone();
    default_source.query_pairs_mut().append_pair("source", name);
    let terminal: EventSourcePipe =
        Box::new(send_cdevents::Processor::new(tx, default_source.to_string()));
    let pipe = transformers::build_transformer_chain(&config.chain.transformers, terminal)?;
    config.extractor.make_extractor(name, pipe, cancel_token, state_config)
}

pub use crate::event::Event as EventSource;
pub use crate::event::EventPipe as EventSourcePipe;

impl TryFrom<EventSource> for CDEvent {
    type Error = miette::Error;

    fn try_from(value: EventSource) -> std::result::Result<Self, Self::Error> {
        let mut body = value.body;
        set_timestamp_if_missing(&mut body);
        set_source_if_missing(&mut body, &value.metadata);
        set_id_zero_or_missing_to_cid(&mut body)?;
        let mut cdevent: CDEvent = serde_json::from_value(body.clone()).map_err(|cause0| {
            // to provide located / contextualized error, body is converted into string, to be then parsed with error
            // TODO find an alternatives that avoid string conversion and reparse
            if let Ok(body_str) = serde_json::to_string_pretty(&body)
                && let Err(cause) = serde_json::from_str::<CDEvent>(&body_str)
            {
                Error::from_serde_error(&body_str, cause)
            } else {
                Error::from_serde_error("", cause0)
            }
        })?;
        let sanitized_id = sanitize_id(cdevent.id())?;
        cdevent = cdevent.with_id(sanitized_id);
        Ok(cdevent)
    }
}

//#[allow(clippy::indexing_slicing)]
fn set_id_zero_or_missing_to_cid(body: &mut serde_json::Value) -> Result<()> {
    use cid::Cid;
    use multihash::Multihash;
    use serde_json::json;
    use sha2::{Digest, Sha256};

    const RAW: u64 = 0x55;
    const SHA2_256: u64 = 0x12;

    let compute_id = body
        .get("context")
        .and_then(|context| context.get("id"))
        .is_none_or(|s| s.is_null() || s.as_str().is_none_or(|id| id.is_empty() || id == "0"));
    if compute_id {
        // Do not use multihash-codetable because one of it's transitive dependency raise
        // an alert "unmaintained advisory detected" about `proc-macro-error`
        // https://rustsec.org/advisories/RUSTSEC-2024-0370
        // let hash = Code::Sha2_256.digest(serde_json::to_string(&input.body)?.as_bytes());
        let mut hasher = Sha256::new();
        // remove the field before computation to create the same hash when set to "0" or missing
        body["context"].as_object_mut().map(|obj| obj.remove("id"));
        hasher.update(serde_json::to_string(&body).into_diagnostic()?.as_bytes());
        let hash = hasher.finalize();
        let mhash = Multihash::<64>::wrap(SHA2_256, &hash).into_diagnostic()?;
        let cid = Cid::new_v1(RAW, mhash);
        body["context"]["id"] = json!(cid.to_string());
    }
    Ok(())
}

fn set_timestamp_if_missing(body: &mut serde_json::Value) {
    use serde_json::json;
    let compute_timestamp = body.get("context").is_none_or(|context| {
        context.get("timestamp").is_none_or(|s| s.is_null() || s.as_str().is_none_or(str::is_empty))
    });
    if compute_timestamp {
        let timestamp = jiff::Timestamp::now().to_string(); //rfc3339
        body["context"]["timestamp"] = json!(timestamp);
    }
}

fn set_source_if_missing(body: &mut serde_json::Value, metadata: &serde_json::Value) {
    let source_missing = body.get("context").is_none_or(|context| {
        context.get("source").is_none_or(|s| s.is_null() || s.as_str().is_none_or(str::is_empty))
    });
    if source_missing
        && let Some(source) = metadata
            .get("context")
            .and_then(|c| c.get("source"))
            .and_then(serde_json::Value::as_str)
            .filter(|s| !s.is_empty())
    {
        body["context"]["source"] = serde_json::json!(source);
    }
}

pub(crate) fn create_sources_and_routes(
    source_configs: impl IntoIterator<Item = (String, Config)>,
    tx: &tokio::sync::broadcast::Sender<Message>,
    cancel_token: &CancellationToken,
    state_config: Option<&crate::state::Config>,
    root_url: &url::Url,
) -> Result<SourceHandlesAndRoutes> {
    let sources = source_configs
        .into_iter()
        .filter(|(_name, config)| config.is_enabled())
        .inspect(|(name, _config)| tracing::info!(kind = "source", name, "starting"))
        .map(|(name, config)| {
            make(&name, &config, tx.clone(), cancel_token.clone(), state_config, root_url)
        })
        .collect::<Result<Vec<_>>>()?;

    let mut join_handles = vec![];
    let mut routes = vec![];
    for source in sources {
        match source {
            extractors::Extractor::Task(task) => join_handles.push(task),
            extractors::Extractor::Webhook(route) => routes.push(route),
        }
    }
    Ok((join_handles, routes))
}

#[cfg(test)]
mod tests {
    use super::*;
    use assert2::assert;
    use serde_json::json;

    #[test]
    fn test_set_id_preserve_non_zero_id() {
        let mut body = json!({
            "context": {
                "id": "my-id",
                "source": "/event/source/123",
                "type": "dev.cdevents.service.deployed.0.1.1",
                "timestamp": "2023-03-20T14:27:05.315384Z"
            },
            "subject": {
                "id": "mySubject123",
                "source": "/event/source/123",
                "type": "service",
                "content": {
                    "environment": {
                        "id": "test123"
                    },
                    "artifactId": "pkg:oci/myapp@sha256%3A0b31b1c02ff458ad9b7b81cbdf8f028bd54699fa151f221d1e8de6817db93427"
                }
            }
        });

        set_id_zero_or_missing_to_cid(&mut body).unwrap();

        assert_eq!(body["context"]["id"], json!("my-id"));
    }

    #[test]
    fn test_set_id_zero_to_cid() {
        let mut body = json!({
            "context": {
                "id": "0",
                "source": "/event/source/123",
                "type": "dev.cdevents.service.deployed.0.1.1",
                "timestamp": "2023-03-20T14:27:05.315384Z"
            },
            "subject": {
                "id": "mySubject123",
                "source": "/event/source/123",
                "type": "service",
                "content": {
                    "environment": {
                        "id": "test123"
                    },
                    "artifactId": "pkg:oci/myapp@sha256%3A0b31b1c02ff458ad9b7b81cbdf8f028bd54699fa151f221d1e8de6817db93427"
                }
            }
        });

        set_id_zero_or_missing_to_cid(&mut body).unwrap();

        assert_eq!(
            body["context"]["id"],
            json!("bafkreicuuogf3pc3lkdewdnngcnxgfrnontlfvtiihmxvcxqqxphxson3m")
        );
    }

    #[test]
    fn test_set_id_missing_to_cid() {
        let mut body = json!({
            "context": {
                "source": "/event/source/123",
                "type": "dev.cdevents.service.deployed.0.1.1",
                "timestamp": "2023-03-20T14:27:05.315384Z"
            },
            "subject": {
                "id": "mySubject123",
                "source": "/event/source/123",
                "type": "service",
                "content": {
                    "environment": {
                        "id": "test123"
                    },
                    "artifactId": "pkg:oci/myapp@sha256%3A0b31b1c02ff458ad9b7b81cbdf8f028bd54699fa151f221d1e8de6817db93427"
                }
            }
        });

        set_id_zero_or_missing_to_cid(&mut body).unwrap();

        assert_eq!(
            body["context"]["id"],
            json!("bafkreicuuogf3pc3lkdewdnngcnxgfrnontlfvtiihmxvcxqqxphxson3m")
        );
    }

    #[test]
    fn test_set_timestamp_if_missing() {
        let mut body = json!({
            "context": {
                "source": "/event/source/123",
                "type": "dev.cdevents.service.deployed.0.1.1",
            },
            "subject": {
                "id": "mySubject123",
                "source": "/event/source/123",
                "type": "service",
                "content": {
                    "environment": {
                        "id": "test123"
                    },
                    "artifactId": "pkg:oci/myapp@sha256%3A0b31b1c02ff458ad9b7b81cbdf8f028bd54699fa151f221d1e8de6817db93427"
                }
            }
        });

        assert!(!body["context"]["timestamp"].is_string());
        set_timestamp_if_missing(&mut body);
        assert2::assert!(let Some(datetime) = body["context"]["timestamp"].as_str());
        assert2::assert!(let Ok(_) = datetime.parse::<jiff::Timestamp>());
    }

    #[test]
    fn test_set_source_if_missing() {
        let metadata = json!({
            "context": {
                "source": "https://example.com?source=my-source"
            }
        });

        // Case 1: source absent → filled from metadata
        let mut body = json!({ "context": {} });
        set_source_if_missing(&mut body, &metadata);
        assert_eq!(
            body["context"]["source"].as_str(),
            Some("https://example.com?source=my-source")
        );

        // Case 2: source null → filled from metadata
        let mut body = json!({ "context": { "source": null } });
        set_source_if_missing(&mut body, &metadata);
        assert_eq!(
            body["context"]["source"].as_str(),
            Some("https://example.com?source=my-source")
        );

        // Case 3: source empty string → filled from metadata
        let mut body = json!({ "context": { "source": "" } });
        set_source_if_missing(&mut body, &metadata);
        assert_eq!(
            body["context"]["source"].as_str(),
            Some("https://example.com?source=my-source")
        );

        // Case 4: source already set → not overridden
        let mut body = json!({ "context": { "source": "/existing/source" } });
        set_source_if_missing(&mut body, &metadata);
        assert_eq!(body["context"]["source"].as_str(), Some("/existing/source"));

        // Case 5: metadata has no context.source → body unchanged
        let mut body = json!({ "context": {} });
        set_source_if_missing(&mut body, &json!({}));
        assert!(body["context"]["source"].is_null());
    }
}
