use std::collections::HashSet;
use std::path::Path;
use std::sync::Arc;

use axum::extract::State;
use axum::http::{HeaderMap, StatusCode};
use axum::Json;
use bytes::Bytes;
use serde::Deserialize;
use tokio::sync::Mutex;

use crate::lfd::events::EventHub;
use crate::lfd::github::{
    github_repo_from_local, poll_check_runs, verify_webhook_signature, CheckRun,
    GitHubCheckRunEvent, GitHubPullRequestEvent, GitHubPushEvent,
};
use crate::lfd::http::state::HttpState;
use crate::lfd::http::{api_error, ApiMessage, ApiResult};
use crate::lfd::id::LfdId;
use crate::lfd::security::canonicalize_existing_path;
use crate::lfd::store::SharedStore;
use crate::lfd::triggers::{enqueue_pending_activation, ActivationEnvelope};
use crate::lfd::types::{Event, Signal, Trigger, Wave, WaveRun, WaveStatus, CI_FIX_FLOW};
use time::format_description::well_known::Rfc3339;
use time::OffsetDateTime;

#[derive(Deserialize)]
pub struct GitHookRequest {
    #[allow(dead_code)]
    hook: String,
    repo: String,
    branch: Option<String>,
    from_sha: Option<String>,
    to_sha: Option<String>,
}

#[derive(Debug, Clone)]
struct WaveCiTarget {
    wave_id: LfdId,
    wave_run_id: LfdId,
    repo_full_name: String,
    branch: String,
    pr_number: u32,
}

#[derive(Debug, Clone, Copy)]
enum WatchRepoTarget<'a> {
    LocalPath(&'a str),
    GitHubRepo(&'a str),
}

impl WatchRepoTarget<'_> {
    fn label(self) -> &'static str {
        match self {
            Self::LocalPath(_) => "git hook",
            Self::GitHubRepo(_) => "github push",
        }
    }
}

pub async fn git_hook_handler(
    State(state): State<HttpState>,
    Json(payload): Json<GitHookRequest>,
) -> ApiResult<serde_json::Value> {
    let repo = canonical_git_hook_repo(&payload.repo)?;
    let matched = enqueue_watch_for_local_repo(
        &state.store,
        &state.event_hub,
        &repo,
        payload.branch.as_deref(),
        payload.from_sha.as_deref(),
        payload.to_sha.as_deref(),
    )
    .await;
    state
        .event_hub
        .send(Event::worktree_updated(repo.clone(), repo, payload.branch));

    Ok(Json(serde_json::json!({ "ok": true, "matched": matched })))
}

fn canonical_git_hook_repo(
    repo: &str,
) -> Result<String, (StatusCode, Json<crate::lfd::http::dto::ErrorResponse>)> {
    let path = Path::new(repo.trim());
    if !path.is_absolute() {
        return Err(api_error(
            StatusCode::BAD_REQUEST,
            "repo must be an absolute path",
        ));
    }

    let canonical = canonicalize_existing_path(path).map_err(|err| {
        api_error(
            StatusCode::BAD_REQUEST,
            ApiMessage::Untrusted(err.to_string()),
        )
    })?;
    if !canonical.is_dir() {
        return Err(api_error(
            StatusCode::BAD_REQUEST,
            "repo must be an existing directory",
        ));
    }

    Ok(canonical.to_string_lossy().to_string())
}

async fn enqueue_watch_for_local_repo(
    store: &SharedStore,
    event_hub: &EventHub,
    repo_path: &str,
    branch: Option<&str>,
    from_sha: Option<&str>,
    to_sha: Option<&str>,
) -> u32 {
    enqueue_watch_for_repo(
        store,
        event_hub,
        WatchRepoTarget::LocalPath(repo_path),
        branch,
        from_sha,
        to_sha,
    )
    .await
}

async fn enqueue_watch_for_github_repo(
    store: &SharedStore,
    event_hub: &EventHub,
    repo_full_name: &str,
    git_ref: Option<&str>,
    from_sha: Option<&str>,
    to_sha: Option<&str>,
) -> u32 {
    enqueue_watch_for_repo(
        store,
        event_hub,
        WatchRepoTarget::GitHubRepo(repo_full_name),
        git_ref,
        from_sha,
        to_sha,
    )
    .await
}

async fn enqueue_watch_for_repo(
    store: &SharedStore,
    event_hub: &EventHub,
    repo_target: WatchRepoTarget<'_>,
    git_ref: Option<&str>,
    from_sha: Option<&str>,
    to_sha: Option<&str>,
) -> u32 {
    if !is_main_ref(git_ref) {
        return 0;
    }

    let triggers = match store.list_triggers_by_signal(Signal::Repo.as_i32()).await {
        Ok(triggers) => triggers,
        Err(err) => {
            tracing::warn!(
                error = %err,
                source = repo_target.label(),
                "failed to list repo triggers"
            );
            return 0;
        }
    };

    let reason = build_push_reason(git_ref, from_sha, to_sha);
    let from_sha = from_sha.unwrap_or("");
    let to_sha = to_sha.unwrap_or("");

    let mut matched = 0_u32;
    for mut trigger in triggers {
        if !trigger.enabled {
            continue;
        }

        let wave = match store.get_wave(&trigger.wave_id).await {
            Ok(Some(wave)) => wave,
            Ok(None) => continue,
            Err(err) => {
                tracing::warn!(
                    trigger_id = %trigger.id,
                    source = repo_target.label(),
                    error = %err,
                    "failed loading wave for repo trigger"
                );
                continue;
            }
        };
        if wave.status() == WaveStatus::Paused || !wave_matches_watch_repo(&wave, repo_target) {
            continue;
        }

        update_watch_main_sha(store, &mut trigger, to_sha).await;

        let outcome = enqueue_pending_activation(
            store,
            event_hub,
            ActivationEnvelope::new(
                &trigger.wave_id,
                Some(&trigger.id),
                reason.clone(),
                from_sha,
                to_sha,
                "main",
            ),
        )
        .await;
        if outcome.is_some() {
            matched += 1;
        }
    }

    matched
}

fn wave_matches_watch_repo(wave: &Wave, repo_target: WatchRepoTarget<'_>) -> bool {
    match repo_target {
        WatchRepoTarget::LocalPath(repo_path) => canonicalize_existing_path(Path::new(wave.repo()))
            .ok()
            .is_some_and(|path| path.to_string_lossy() == repo_path),
        WatchRepoTarget::GitHubRepo(repo_full_name) => {
            github_repo_from_local(Path::new(wave.repo()))
                .as_deref()
                .is_some_and(|wave_repo_full_name| wave_repo_full_name == repo_full_name)
        }
    }
}

async fn update_watch_main_sha(store: &SharedStore, trigger: &mut Trigger, to_sha: &str) {
    if to_sha.is_empty() {
        return;
    }

    trigger.last_main_sha = Some(to_sha.to_string());
    let _ = store.update_trigger(trigger).await;
}

fn is_main_ref(value: Option<&str>) -> bool {
    value.is_none_or(|ref_name| {
        ref_name == "main" || ref_name == "refs/heads/main" || ref_name.ends_with("/main")
    })
}

fn build_push_reason(branch: Option<&str>, from_sha: Option<&str>, to_sha: Option<&str>) -> String {
    let branch = branch.unwrap_or("main");
    let from_sha = from_sha.unwrap_or("");
    let to_sha = to_sha.unwrap_or("");
    if from_sha.is_empty() || to_sha.is_empty() {
        return format!("{branch} updated");
    }
    format!("{branch} advanced {from_sha}..{to_sha}")
}

pub async fn github_webhook_handler(
    State(state): State<HttpState>,
    headers: HeaderMap,
    body: Bytes,
) -> ApiResult<serde_json::Value> {
    let secret = state.github.webhook_secret.clone();
    if secret.trim().is_empty() {
        return Err(api_error(
            StatusCode::SERVICE_UNAVAILABLE,
            "github webhook secret is not configured",
        ));
    }

    let signature = headers
        .get("X-Hub-Signature-256")
        .and_then(|value| value.to_str().ok())
        .ok_or_else(|| api_error(StatusCode::UNAUTHORIZED, "missing X-Hub-Signature-256"))?;
    if !verify_webhook_signature(&secret, body.as_ref(), signature) {
        return Err(api_error(
            StatusCode::UNAUTHORIZED,
            "invalid webhook signature",
        ));
    }

    let event_kind = headers
        .get("X-GitHub-Event")
        .and_then(|value| value.to_str().ok())
        .unwrap_or_default()
        .to_ascii_lowercase();

    match event_kind.as_str() {
        "push" => {
            let event = serde_json::from_slice::<GitHubPushEvent>(&body).map_err(|err| {
                api_error(
                    StatusCode::BAD_REQUEST,
                    ApiMessage::Untrusted(err.to_string()),
                )
            })?;

            let matched = enqueue_watch_for_github_repo(
                &state.store,
                &state.event_hub,
                &event.repository.full_name,
                Some(&event.git_ref),
                Some(&event.before),
                Some(&event.after),
            )
            .await;

            Ok(Json(serde_json::json!({ "ok": true, "matched": matched })))
        }
        "check_run" => {
            let event = serde_json::from_slice::<GitHubCheckRunEvent>(&body).map_err(|err| {
                api_error(
                    StatusCode::BAD_REQUEST,
                    ApiMessage::Untrusted(err.to_string()),
                )
            })?;

            if event.action != "completed"
                || !is_failed_check_run(
                    &event.check_run.status,
                    event.check_run.conclusion.as_deref(),
                )
            {
                return Ok(Json(
                    serde_json::json!({ "ok": true, "matched": 0, "skipped": true }),
                ));
            }

            let mut matched = 0_u32;
            for pr in &event.check_run.pull_requests {
                let targets = find_wave_ci_targets(
                    &state.store,
                    &event.repository.full_name,
                    &pr.head.branch,
                    Some(pr.number),
                )
                .await
                .map_err(|err| {
                    api_error(
                        StatusCode::INTERNAL_SERVER_ERROR,
                        ApiMessage::Untrusted(err),
                    )
                })?;

                for target in targets {
                    let emitted = emit_ci_failure(
                        &state.event_hub,
                        &state.ci_failure_cache,
                        build_ci_failure_event(&target, &event.check_run),
                    )
                    .await;
                    if emitted {
                        matched += 1;
                        tracing::info!(
                            wave_id = %target.wave_id,
                            wave_run_id = %target.wave_run_id,
                            repo = %event.repository.full_name,
                            branch = %target.branch,
                            commit_sha = %event.check_run.head_sha,
                            check_id = event.check_run.id,
                            check_name = %event.check_run.name,
                            "matched GitHub CI failure to wave"
                        );
                    }
                }
            }

            Ok(Json(serde_json::json!({ "ok": true, "matched": matched })))
        }
        "pull_request" => {
            let event = serde_json::from_slice::<GitHubPullRequestEvent>(&body).map_err(|err| {
                api_error(
                    StatusCode::BAD_REQUEST,
                    ApiMessage::Untrusted(err.to_string()),
                )
            })?;
            if event.action != "closed" || !event.pull_request.merged {
                return Ok(Json(
                    serde_json::json!({ "ok": true, "processed": 0, "skipped": true }),
                ));
            }
            let merged_at = event
                .pull_request
                .merged_at
                .as_deref()
                .and_then(|value| OffsetDateTime::parse(value, &Rfc3339).ok())
                .unwrap_or_else(OffsetDateTime::now_utc);
            let wave_ids = find_waves_for_pr(
                &state.store,
                &event.repository.full_name,
                event.pull_request.number,
            )
            .await
            .map_err(|err| {
                api_error(
                    StatusCode::INTERNAL_SERVER_ERROR,
                    ApiMessage::Untrusted(err),
                )
            })?;
            let mut processed = 0_u32;
            for wave_id in wave_ids {
                let handled = crate::lfd::queue::handle_pr_merged_with_events(
                    &state.store,
                    &state.github,
                    &wave_id,
                    event.pull_request.number,
                    merged_at,
                    Some(&state.event_hub),
                )
                .await
                .map_err(|err| {
                    api_error(
                        StatusCode::INTERNAL_SERVER_ERROR,
                        ApiMessage::Untrusted(err),
                    )
                })?;
                if handled {
                    processed += 1;
                }
            }
            Ok(Json(
                serde_json::json!({ "ok": true, "processed": processed }),
            ))
        }
        _ => Ok(Json(
            serde_json::json!({ "ok": true, "matched": 0, "skipped": true }),
        )),
    }
}

pub async fn poll_all_waves_ci(
    store: &SharedStore,
    event_hub: &EventHub,
    token: &str,
    cache: &Arc<Mutex<HashSet<String>>>,
) -> Result<u32, String> {
    let targets = list_wave_ci_targets(store, None).await?;
    emit_ci_failures_for_targets(event_hub, cache, token, targets).await
}

pub async fn poll_wave_ci(
    store: &SharedStore,
    event_hub: &EventHub,
    cache: &Arc<Mutex<HashSet<String>>>,
    wave_id: &LfdId,
    token: &str,
) -> Result<u32, String> {
    let targets = list_wave_ci_targets(store, Some(wave_id.clone())).await?;
    emit_ci_failures_for_targets(event_hub, cache, token, targets).await
}

async fn emit_ci_failures_for_targets(
    event_hub: &EventHub,
    cache: &Arc<Mutex<HashSet<String>>>,
    token: &str,
    targets: Vec<WaveCiTarget>,
) -> Result<u32, String> {
    let mut emitted = 0_u32;

    for target in targets {
        let check_runs = poll_check_runs(&target.repo_full_name, &target.branch, token).await?;
        for check_run in check_runs {
            if !is_failed_check_run(&check_run.status, check_run.conclusion.as_deref()) {
                continue;
            }

            let event = build_ci_failure_event(&target, &check_run);
            if emit_ci_failure(event_hub, cache, event).await {
                emitted += 1;
            }
        }
    }

    Ok(emitted)
}

async fn list_wave_ci_targets(
    store: &SharedStore,
    wave_filter: Option<LfdId>,
) -> Result<Vec<WaveCiTarget>, String> {
    let waves = if let Some(wave_id) = wave_filter {
        store
            .get_wave(&wave_id)
            .await
            .map_err(|err| err.to_string())?
            .map(|wave| vec![wave])
            .unwrap_or_default()
    } else {
        store
            .list_waves(None)
            .await
            .map_err(|err| err.to_string())?
    };
    collect_wave_ci_targets(store, waves, None, None, None).await
}

async fn find_wave_ci_targets(
    store: &SharedStore,
    repo_full_name: &str,
    branch: &str,
    pr_number: Option<u32>,
) -> Result<Vec<WaveCiTarget>, String> {
    let waves = store
        .list_waves(None)
        .await
        .map_err(|err| err.to_string())?;
    collect_wave_ci_targets(store, waves, Some(repo_full_name), Some(branch), pr_number).await
}

async fn find_waves_for_pr(
    store: &SharedStore,
    repo_full_name: &str,
    pr_number: u32,
) -> Result<Vec<LfdId>, String> {
    let waves = store
        .list_waves(None)
        .await
        .map_err(|err| err.to_string())?;
    let mut matches = Vec::new();
    for wave in waves {
        let Some(repo_name) = github_repo_from_local(Path::new(wave.repo())) else {
            continue;
        };
        if repo_name != repo_full_name {
            continue;
        }
        let has_pr = store
            .list_stack_runs(wave.id())
            .await
            .map_err(|err| err.to_string())?
            .into_iter()
            .any(|run| run.pr.and_then(|pr| pr.number) == Some(pr_number));
        if has_pr {
            matches.push(wave.id().clone());
        }
    }
    Ok(matches)
}

async fn collect_wave_ci_targets(
    store: &SharedStore,
    waves: Vec<Wave>,
    repo_filter: Option<&str>,
    branch: Option<&str>,
    pr_number: Option<u32>,
) -> Result<Vec<WaveCiTarget>, String> {
    let mut targets = Vec::new();

    for wave in waves {
        let Some(repo_full_name) = github_repo_from_local(Path::new(wave.repo())) else {
            continue;
        };
        if repo_filter.is_some_and(|repo| repo != repo_full_name) {
            continue;
        }

        let Some(target) =
            find_wave_ci_target(store, &wave, &repo_full_name, branch, pr_number).await?
        else {
            continue;
        };
        targets.push(target);
    }

    Ok(targets)
}

async fn find_wave_ci_target(
    store: &SharedStore,
    wave: &Wave,
    repo_full_name: &str,
    branch: Option<&str>,
    pr_number: Option<u32>,
) -> Result<Option<WaveCiTarget>, String> {
    let runs = store
        .list_wave_runs(Some(wave.id()), None)
        .await
        .map_err(|err| err.to_string())?;
    let run = runs
        .into_iter()
        .find(|run| run_matches_ci_target(run, branch, pr_number));

    Ok(run.and_then(|run| wave_ci_target(wave.id(), repo_full_name, &run)))
}

fn run_matches_ci_target(run: &WaveRun, branch: Option<&str>, pr_number: Option<u32>) -> bool {
    if run.snapshot.flow == CI_FIX_FLOW {
        return false;
    }

    let Some(pr) = run.pr.as_ref() else {
        return false;
    };
    super::is_open_pr_state(pr.state.as_deref())
        && branch.is_none_or(|branch| pr.branch.as_deref() == Some(branch))
        && pr_number.is_none_or(|number| pr.number == Some(number))
}

fn is_failed_check_run(status: &str, conclusion: Option<&str>) -> bool {
    status == "completed" && conclusion == Some("failure")
}

fn wave_ci_target(wave_id: &LfdId, repo_full_name: &str, run: &WaveRun) -> Option<WaveCiTarget> {
    let pr = run.pr.as_ref()?;
    Some(WaveCiTarget {
        wave_id: wave_id.clone(),
        wave_run_id: run.id.clone(),
        repo_full_name: repo_full_name.to_string(),
        branch: pr.branch.clone()?,
        pr_number: pr.number?,
    })
}

fn build_ci_failure_event(target: &WaveCiTarget, check_run: &CheckRun) -> Event {
    Event::ci_failure(
        target.wave_id.clone(),
        target.wave_run_id.clone(),
        target.pr_number,
        target.branch.clone(),
        check_run.head_sha.clone(),
        check_run.name.clone(),
        check_run.html_url.clone(),
    )
}

async fn emit_ci_failure(
    event_hub: &EventHub,
    cache: &Arc<Mutex<HashSet<String>>>,
    event: Event,
) -> bool {
    let (wave_id, commit_sha) = match &event {
        Event::CiFailure {
            wave_id,
            commit_sha,
            ..
        } => (wave_id.clone(), commit_sha.clone()),
        _ => return false,
    };

    let key = format!("{wave_id}:{commit_sha}");
    let mut cache = cache.lock().await;
    if !cache.insert(key) {
        return false;
    }
    event_hub.send(event);
    true
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::lfd::auth::{AuthFailureThrottle, AuthProvider};
    use crate::lfd::config::{ExecutorConfig, GitHubConfig, HttpSecurityConfig};
    use crate::lfd::events::EventHub;
    use crate::lfd::executor::WaveExecutor;
    use crate::lfd::http::state::HttpState;
    use crate::lfd::output::OutputHub;
    use crate::lfd::provider_auth::ProviderAuthService;
    use crate::lfd::scheduler::Scheduler;
    use crate::lfd::sessions::SessionManager;
    use crate::lfd::store::{open_store, SharedStore, StorageConfig};
    use crate::lfd::types::{
        PullRequest, Signal, Trigger, Wave, WaveMode, WaveRunSnapshot, WaveRunStatus, WaveStatus,
    };
    use std::sync::Arc;
    use tempfile::tempdir;
    use time::OffsetDateTime;
    use tokio::sync::Mutex;

    async fn test_http_state() -> HttpState {
        let tmp = tempdir().expect("tempdir");
        let db_path = tmp.path().join("lfd.db");
        let store: SharedStore = Arc::new(
            open_store(&StorageConfig::sqlite(db_path))
                .await
                .expect("open sqlite store"),
        );
        let scheduler = Arc::new(Scheduler::new(1));
        let output_hub = OutputHub::new(128, tmp.path().join("output"));
        let event_hub = EventHub::new(128);
        let sessions = SessionManager::new(store.clone());
        let executor = Arc::new(
            WaveExecutor::new(
                store.clone(),
                scheduler.clone(),
                output_hub.clone(),
                event_hub.clone(),
                ExecutorConfig::default(),
                GitHubConfig::default(),
            )
            .expect("build executor"),
        );
        HttpState {
            store: store.clone(),
            scheduler,
            executor,
            event_hub,
            output_hub,
            provider_auth: ProviderAuthService::new(store.clone()),
            auth: AuthProvider::Bearer {
                session_token: secrecy::SecretString::from("test-token".to_string()),
            },
            started_at: OffsetDateTime::now_utc(),
            github: GitHubConfig::default(),
            http_security: HttpSecurityConfig::default(),
            auth_failure_throttle: AuthFailureThrottle::new(),
            ci_failure_cache: Arc::new(Mutex::new(std::collections::HashSet::new())),
            sessions,
        }
    }

    #[tokio::test]
    async fn git_hook_handler_rejects_relative_repo_paths() {
        let state = test_http_state().await;
        let result = git_hook_handler(
            State(state),
            Json(GitHookRequest {
                hook: "post-commit".to_string(),
                repo: "../repo".to_string(),
                branch: None,
                from_sha: None,
                to_sha: None,
            }),
        )
        .await;
        assert!(matches!(result, Err((StatusCode::BAD_REQUEST, _))));
    }

    #[tokio::test]
    async fn git_hook_handler_canonicalizes_repo_path() {
        let state = test_http_state().await;
        let repo_dir = tempdir().expect("repo tempdir");
        let mut rx = state.event_hub.subscribe();
        let alias_path = repo_dir.path().join("..").join(
            repo_dir
                .path()
                .file_name()
                .expect("tempdir should have file name"),
        );
        let result = git_hook_handler(
            State(state),
            Json(GitHookRequest {
                hook: "post-commit".to_string(),
                repo: alias_path.to_string_lossy().to_string(),
                branch: Some("main".to_string()),
                from_sha: None,
                to_sha: None,
            }),
        )
        .await;
        assert!(result.is_ok());

        let event = rx.try_recv().expect("event emitted");
        match event {
            Event::WorktreeUpdated { repo, .. } => {
                assert_eq!(
                    repo,
                    repo_dir
                        .path()
                        .canonicalize()
                        .expect("canonical repo path")
                        .to_string_lossy()
                );
            }
            _ => panic!("expected worktree updated event"),
        }
    }

    #[tokio::test]
    async fn github_push_enqueues_watch_activation() {
        let state = test_http_state().await;
        let repo_dir = tempdir().expect("repo tempdir");
        std::process::Command::new("git")
            .args(["-C", repo_dir.path().to_string_lossy().as_ref(), "init"])
            .status()
            .expect("git init should run");
        std::process::Command::new("git")
            .args([
                "-C",
                repo_dir.path().to_string_lossy().as_ref(),
                "remote",
                "add",
                "origin",
                "git@github.com:loopflowstudio/loopflow.git",
            ])
            .status()
            .expect("git remote add should run");

        let wave = Wave {
            id: LfdId::new(),
            name: "watch-wave".to_string(),
            repo: repo_dir.path().to_string_lossy().to_string(),
            mode: WaveMode::Loop,
            primary_flow: "ship-roadmap".to_string(),
            crons: Vec::new(),
            direction: vec![],
            area: vec![],
            status: WaveStatus::Idle,
            iteration: 0,
            cycle_start_iteration: 0,
            created_at: Some(OffsetDateTime::now_utc()),
            workers: 1,
        };
        state.store.create_wave(&wave).await.expect("create wave");
        let trigger = Trigger {
            id: LfdId::new(),
            wave_id: wave.id.clone(),
            source_wave_id: None,
            signal: Signal::Repo,
            flow: None,
            last_main_sha: None,
            last_triggered_at: None,
            created_at: Some(OffsetDateTime::now_utc()),
            enabled: true,
            max_iterations: None,
        };
        state
            .store
            .create_trigger(&trigger)
            .await
            .expect("create repo trigger");

        let matched = enqueue_watch_for_github_repo(
            &state.store,
            &state.event_hub,
            "loopflowstudio/loopflow",
            Some("refs/heads/main"),
            Some("abc"),
            Some("def"),
        )
        .await;
        assert_eq!(matched, 1);

        let pending = state
            .store
            .list_pending_activations(&wave.id)
            .await
            .expect("pending activations");
        assert_eq!(pending.len(), 1);
        assert_eq!(pending[0].to_sha, "def");
    }

    fn wave_run_with_pr(flow: &str, pr_state: Option<&str>, branch: Option<&str>) -> WaveRun {
        WaveRun {
            id: LfdId::new(),
            wave_id: LfdId::new(),
            snapshot: WaveRunSnapshot {
                repo: ".".to_string(),
                flow: flow.to_string(),
                direction: Vec::new(),
                area: Vec::new(),
            },
            iteration: 0,
            step_index: 0,
            status: WaveRunStatus::Running,
            worktree: "/tmp/worktree".to_string(),
            branch: "feature".to_string(),
            started_at: None,
            ended_at: None,
            error: None,
            flow_parents: Vec::new(),
            execution_cursor: None,
            activation_log_id: None,
            parent_run_id: None,
            parent_pr_number: None,
            stack_position: 0,
            stack_group_id: "wave-group".to_string(),
            stack_status: crate::lfd::types::WaveRunStackStatus::Active,
            lineage_inferred: false,
            target_branch: "main".to_string(),
            repair_of: None,
            pr: Some(PullRequest {
                url: "https://example.test/pr/1".to_string(),
                number: Some(1),
                state: pr_state.map(ToString::to_string),
                title: Some("test".to_string()),
                branch: branch.map(ToString::to_string),
            }),
        }
    }

    #[test]
    fn run_matches_ci_target_only_matches_open_main_prs() {
        let run = wave_run_with_pr("build", Some("open"), Some("feature"));
        assert!(run_matches_ci_target(&run, Some("feature"), Some(1)));

        let closed = wave_run_with_pr("build", Some("closed"), Some("feature"));
        assert!(!run_matches_ci_target(&closed, Some("feature"), Some(1)));

        let unknown_state = wave_run_with_pr("build", None, Some("feature"));
        assert!(!run_matches_ci_target(
            &unknown_state,
            Some("feature"),
            Some(1)
        ));

        let ci_fix = wave_run_with_pr("ci-fix", Some("open"), Some("feature"));
        assert!(!run_matches_ci_target(&ci_fix, Some("feature"), Some(1)));
    }

    #[tokio::test]
    async fn emit_ci_failure_deduplicates_by_wave_and_commit_sha() {
        let event_hub = EventHub::new(8);
        let cache = Arc::new(Mutex::new(HashSet::new()));
        let event = Event::ci_failure(
            LfdId::new(),
            LfdId::new(),
            1,
            "feature".to_string(),
            "abc123".to_string(),
            "test-check".to_string(),
            "https://example.test/logs".to_string(),
        );

        assert!(emit_ci_failure(&event_hub, &cache, event.clone()).await);
        assert!(!emit_ci_failure(&event_hub, &cache, event).await);
    }
}
