use std::sync::Arc;

use tokio::task::JoinHandle;
use tokio_util::sync::CancellationToken;

use super::activation::{
    enqueue_pending_activation, ActivationEnvelope, EnqueueOutcome, ImmediateActivation,
};
use super::spawn_immediate_activation;
use crate::lfd::events::EventHub;
use crate::lfd::executor::WaveExecutor;
use crate::lfd::id::LfdId;
use crate::lfd::scheduler::Scheduler;
use crate::lfd::store::SharedStore;
use crate::lfd::types::{Event, Signal, Trigger, CI_FIX_FLOW};
use time::OffsetDateTime;

#[derive(Debug)]
struct CiFailureActivation {
    wave_id: LfdId,
    pr_number: u32,
    branch: String,
    commit_sha: String,
    check_name: String,
    logs_url: String,
}

pub fn spawn_ci_failure_handler(
    store: SharedStore,
    executor: WaveExecutor,
    scheduler: Arc<Scheduler>,
    event_hub: EventHub,
    cancel: CancellationToken,
) -> JoinHandle<()> {
    let mut rx = event_hub.subscribe();

    tokio::spawn(async move {
        loop {
            tokio::select! {
                _ = cancel.cancelled() => {
                    tracing::info!("ci_failure_handler shutting down");
                    break;
                }
                event = rx.recv() => {
                    match event {
                        Ok(Event::CiFailure {
                            wave_id,
                            pr_number,
                            branch,
                            commit_sha,
                            check_name,
                            logs_url,
                            ..
                        }) => {
                            let activation = CiFailureActivation {
                                wave_id,
                                pr_number,
                                branch,
                                commit_sha,
                                check_name,
                                logs_url,
                            };
                            if let Err(err) =
                                handle_ci_failure_event(&store, &executor, &scheduler, &event_hub, activation).await
                            {
                                tracing::warn!(error = %err, "failed handling CI failure activation");
                            }
                        }
                        Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
                        _ => {}
                    }
                }
            }
        }
    })
}

async fn handle_ci_failure_event(
    store: &SharedStore,
    executor: &WaveExecutor,
    scheduler: &Arc<Scheduler>,
    event_hub: &EventHub,
    activation: CiFailureActivation,
) -> Result<(), String> {
    let trigger = resolve_ci_failure_trigger(store, &activation.wave_id).await?;
    let reason = format!(
        "CI failure for PR #{} on {} ({}): {}",
        activation.pr_number, activation.branch, activation.check_name, activation.logs_url
    );
    let envelope = ActivationEnvelope::new(
        &activation.wave_id,
        Some(&trigger.id),
        reason,
        &activation.commit_sha,
        &activation.commit_sha,
        &activation.branch,
    );

    let wave = store
        .get_wave(&activation.wave_id)
        .await
        .map_err(|err| err.to_string())?
        .ok_or_else(|| format!("wave {} not found", activation.wave_id))?;

    if wave.workers() == 1 {
        let outcome = enqueue_pending_activation(store, event_hub, envelope).await;
        if outcome.is_none() {
            return Err(format!(
                "failed to enqueue CI failure activation for wave {}",
                activation.wave_id
            ));
        }
        if matches!(outcome, Some(EnqueueOutcome::Dropped)) {
            tracing::warn!(
                wave_id = %activation.wave_id,
                trigger_id = %trigger.id,
                "dropped CI failure activation because queue is full"
            );
        }
    } else if let Err(err) = spawn_immediate_activation(
        store,
        executor,
        scheduler,
        event_hub,
        ImmediateActivation {
            wave: &wave,
            flow_override: trigger.flow.clone(),
            roadmap_item: None,
            force_parallel: false,
            envelope,
        },
    )
    .await
    {
        tracing::warn!(wave_id = %wave.id(), trigger_id = %trigger.id, error = %err, "ci failure activation failed");
    }
    Ok(())
}

async fn resolve_ci_failure_trigger(
    store: &SharedStore,
    wave_id: &LfdId,
) -> Result<Trigger, String> {
    let triggers = store
        .list_triggers(Some(wave_id))
        .await
        .map_err(|err| err.to_string())?;
    if let Some(existing) = triggers
        .into_iter()
        .find(|trigger| trigger.signal == Signal::CiFailure)
    {
        return Ok(existing);
    }

    let trigger = Trigger {
        id: LfdId::new(),
        wave_id: wave_id.clone(),
        source_wave_id: None,
        signal: Signal::CiFailure,
        flow: Some(CI_FIX_FLOW.to_string()),
        last_main_sha: None,
        last_triggered_at: Some(OffsetDateTime::now_utc().unix_timestamp()),
        created_at: Some(OffsetDateTime::now_utc()),
        enabled: true,
        max_iterations: None,
    };
    store
        .create_trigger(&trigger)
        .await
        .map_err(|err| err.to_string())?;
    Ok(trigger)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::lfd::scheduler::Scheduler;
    use crate::lfd::store::{open_store, StorageConfig};
    use crate::lfd::types::{Wave, WaveMode, WaveRun, WaveRunStatus, WaveStatus};
    use std::sync::Arc;

    async fn create_store() -> SharedStore {
        let path = std::env::temp_dir().join(format!("lfd-ci-failure-test-{}.db", LfdId::new()));
        open_store(&StorageConfig::sqlite(path))
            .await
            .map(Arc::new)
            .expect("store")
    }

    async fn create_wave(store: &SharedStore) -> Wave {
        let wave = Wave {
            id: LfdId::new(),
            name: "ci-wave".to_string(),
            repo: ".".to_string(),
            mode: WaveMode::Loop,
            primary_flow: "ship-roadmap".to_string(),
            crons: Vec::new(),
            direction: Vec::new(),
            area: Vec::new(),
            status: WaveStatus::Idle,
            iteration: 0,
            cycle_start_iteration: 0,
            created_at: Some(OffsetDateTime::now_utc()),
            workers: 1,
        };
        store.create_wave(&wave).await.expect("create wave");
        wave
    }

    async fn create_ci_failure_trigger(store: &SharedStore, wave: &Wave) -> Trigger {
        let trigger = Trigger {
            id: LfdId::new(),
            wave_id: wave.id.clone(),
            source_wave_id: None,
            signal: Signal::CiFailure,
            flow: Some(CI_FIX_FLOW.to_string()),
            last_main_sha: None,
            last_triggered_at: None,
            created_at: Some(OffsetDateTime::now_utc()),
            enabled: true,
            max_iterations: None,
        };
        store
            .create_trigger(&trigger)
            .await
            .expect("create trigger");
        trigger
    }

    #[tokio::test]
    async fn resolve_ci_failure_trigger_reuses_existing() {
        let store = create_store().await;
        let wave = create_wave(&store).await;
        let existing = create_ci_failure_trigger(&store, &wave).await;

        let trigger = resolve_ci_failure_trigger(&store, &wave.id)
            .await
            .expect("resolve trigger");
        assert_eq!(trigger.id, existing.id);

        let triggers = store
            .list_triggers(Some(&wave.id))
            .await
            .expect("list triggers");
        assert_eq!(triggers.len(), 1);
    }

    #[tokio::test]
    async fn resolve_ci_failure_trigger_creates_default_ci_fix_flow() {
        let store = create_store().await;
        let wave = create_wave(&store).await;

        let trigger = resolve_ci_failure_trigger(&store, &wave.id)
            .await
            .expect("resolve trigger");

        assert_eq!(trigger.signal, Signal::CiFailure);
        assert_eq!(trigger.flow.as_deref(), Some("ci-fix"));
        assert!(trigger.enabled);
    }

    async fn create_serialized_wave(store: &SharedStore) -> Wave {
        let wave = Wave {
            id: LfdId::new(),
            name: "ci-wave-serialized".to_string(),
            repo: ".".to_string(),
            mode: WaveMode::Loop,
            primary_flow: "ship-roadmap".to_string(),
            crons: Vec::new(),
            direction: Vec::new(),
            area: Vec::new(),
            status: WaveStatus::Idle,
            iteration: 0,
            cycle_start_iteration: 0,
            created_at: Some(OffsetDateTime::now_utc()),
            workers: 1,
        };
        store.create_wave(&wave).await.expect("create wave");
        wave
    }

    fn stub_executor(store: SharedStore, event_hub: EventHub) -> (WaveExecutor, Arc<Scheduler>) {
        use crate::lfd::executor::{AgentExecutor, AgentRunContext};
        use crate::lfd::output::OutputHub;
        use async_trait::async_trait;
        use std::path::Path;

        struct StubRunner;

        #[async_trait]
        impl AgentExecutor for StubRunner {
            async fn run(
                &self,
                _cmd: Vec<String>,
                _cwd: &Path,
                _ctx: AgentRunContext,
            ) -> anyhow::Result<i32> {
                Ok(0)
            }
            async fn terminate(&self, _agent_id: &str) -> anyhow::Result<()> {
                Ok(())
            }
        }

        let scheduler = Arc::new(Scheduler::new(1));
        let output_dir = std::env::temp_dir().join(format!("lfd-ci-test-output-{}", LfdId::new()));
        let output = OutputHub::new(16, output_dir);
        let executor = WaveExecutor::with_runner(
            store,
            scheduler.clone(),
            output,
            event_hub,
            Arc::new(StubRunner),
        );
        (executor, scheduler)
    }

    #[tokio::test]
    async fn handle_ci_failure_event_enqueues_push_activation_for_serialized_wave() {
        let store = create_store().await;
        let event_hub = EventHub::new(16);
        let wave = create_serialized_wave(&store).await;
        let trigger = create_ci_failure_trigger(&store, &wave).await;
        let mut active_run = WaveRun::new(LfdId::new(), wave.id.clone());
        active_run.status = WaveRunStatus::Running;
        active_run.snapshot.repo = wave.repo.clone();
        active_run.snapshot.flow = wave.primary_flow.clone();
        store
            .create_wave_run(&active_run)
            .await
            .expect("seed active run");
        let activation = CiFailureActivation {
            wave_id: wave.id.clone(),
            pr_number: 42,
            branch: "feature/test".to_string(),
            commit_sha: "abc123".to_string(),
            check_name: "rust-test".to_string(),
            logs_url: "https://example.com/logs".to_string(),
        };

        let (executor, scheduler) = stub_executor(store.clone(), event_hub.clone());

        handle_ci_failure_event(&store, &executor, &scheduler, &event_hub, activation)
            .await
            .expect("enqueue activation");

        let pending = store
            .list_pending_activations(&wave.id)
            .await
            .expect("list pending");
        assert_eq!(pending.len(), 1);
        assert_eq!(pending[0].trigger_id, Some(trigger.id));
        assert_eq!(pending[0].from_sha, "abc123");
        assert_eq!(pending[0].to_sha, "abc123");
        assert!(pending[0].reason.contains("PR #42"));
        assert!(pending[0].reason.contains("rust-test"));
    }
}
