use anyhow::Result;
use rusqlite::Connection;

use crate::analytics;
use crate::change_intel::commit_assoc;
use crate::change_intel::pipeline;
use crate::db;
use crate::github;
use crate::ingest_progress::{
    IngestExecutionPlan, IngestProgress, IngestProgressObserver, ProviderWorkPlan,
};
use crate::providers;

pub fn run(verbose: bool) -> Result<()> {
    let mut db = db::open()?;
    let providers = providers::all_providers();
    println!("Planning ingest...");

    let mut provider_plans = Vec::with_capacity(providers.len());
    for provider in &providers {
        provider_plans.push(ProviderWorkPlan {
            provider_name: provider.name().to_string(),
            session_plan: provider.plan_session_work()?,
            code_change_plan: pipeline::plan_provider_code_changes(provider.name())?,
        });
    }
    let (association_units_estimate, commit_materialization_units_estimate) =
        estimate_execution_units(&db, &provider_plans)?;
    let execution_plan = IngestExecutionPlan::new(
        provider_plans,
        association_units_estimate,
        commit_materialization_units_estimate,
    );
    let mut progress = IngestProgress::new(&execution_plan);

    let mut grand_total = 0usize;
    for provider in &providers {
        let provider_name = provider.name();
        let provider_plan = execution_plan
            .provider_plan(provider_name)
            .ok_or_else(|| anyhow::anyhow!("missing provider plan for {}", provider_name))?;

        if verbose {
            println!("Ingesting {} ...", provider_name);
        }
        let session_result = {
            let result = {
                let mut observer = progress.stage(
                    format_provider_stage(provider_name, "sessions"),
                    provider_plan.session_units(),
                );
                provider.ingest(
                    &db,
                    &provider_plan.session_plan,
                    verbose,
                    Some(&mut observer),
                )
            };
            progress.finish_stage();
            result
        };
        match session_result {
            Ok(n) => {
                if verbose {
                    println!("  {} rows written", n);
                }
                grand_total += n;
            }
            Err(e) => println!("  error: {}", e),
        }

        let change_result = {
            let result = {
                let mut observer = progress.stage(
                    format_provider_stage(provider_name, "changes"),
                    provider_plan.code_change_units(),
                );
                pipeline::ingest_provider_code_changes(
                    &mut db,
                    &provider_plan.code_change_plan,
                    verbose,
                    Some(&mut observer),
                )
            };
            progress.finish_stage();
            result
        };
        match change_result {
            Ok(summary) => {
                let mut line = format!(
                    "  code changes [{}]: sources={} skipped={} calls={} ops={} parse_errors={}",
                    summary.provider,
                    summary.sources_discovered,
                    summary.sources_skipped,
                    summary.tool_calls_inspected,
                    summary.ops_upserted,
                    summary.parse_errors
                );
                if summary.legacy_sessions_considered > 0
                    || summary.legacy_entries_inspected > 0
                    || summary.legacy_diff_rows_found > 0
                    || summary.legacy_ops_upserted > 0
                    || summary.legacy_parse_errors > 0
                {
                    line.push_str(&format!(
                        " legacy[sessions={} entries={} diffs={} ops={} errors={}]",
                        summary.legacy_sessions_considered,
                        summary.legacy_entries_inspected,
                        summary.legacy_diff_rows_found,
                        summary.legacy_ops_upserted,
                        summary.legacy_parse_errors
                    ));
                }
                if verbose {
                    println!("{}", line);
                }
            }
            Err(e) => {
                println!("  code changes error: {}", e);
            }
        }
    }

    {
        {
            let mut observer = progress.stage("Session Events", 1);
            analytics::refresh_session_events(&db)?;
            observer.advance("session events refreshed");
        }
        progress.finish_stage();
    }

    if verbose {
        println!("\nAssociating commits ...");
    }
    let association_plan = commit_assoc::plan(&db)?;
    progress.replace_future_units(
        execution_plan.association_units_estimate,
        association_plan.total_units(),
    );
    let assoc_summary = {
        let result = {
            let mut observer = progress.stage("Commit Association", association_plan.total_units());
            commit_assoc::run_with_plan(&mut db, &association_plan, verbose, Some(&mut observer))
        };
        progress.finish_stage();
        result?
    };
    if verbose {
        for repo in &assoc_summary.repo_summaries {
            if repo.skipped_non_git {
                println!("  {} skipped (not a git repo)", repo.repo_root);
                continue;
            }

            println!(
                "  {} commits={} attributed={} heavy={} errors={}",
                repo.repo_root,
                repo.commits_scanned,
                repo.commits_attributed,
                repo.heavy_commits,
                repo.errors
            );
        }

        println!(
            "\nAssociation summary: repos_considered={} selected={} processed={} non_git_skipped={} commits_scanned={} commits_attributed={} heavy_commits={} errors={}",
            assoc_summary.repos_considered,
            assoc_summary.repos_selected,
            assoc_summary.repos_processed,
            assoc_summary.repos_skipped_non_git,
            assoc_summary.commits_scanned,
            assoc_summary.commits_attributed,
            assoc_summary.heavy_commits,
            assoc_summary.errors
        );
    }

    analytics::refresh_session_events(&db)?;

    let commit_materialization_units = count_commit_materialization_units(&db)?;
    progress.replace_future_units(
        execution_plan.commit_materialization_units_estimate,
        commit_materialization_units,
    );
    let commit_refresh = {
        let result = {
            let mut observer =
                progress.stage("Commit Materialization", commit_materialization_units);
            analytics::refresh_commit_events(&mut db, verbose, Some(&mut observer))
        };
        progress.finish_stage();
        result?
    };
    if verbose {
        println!(
            "Commit events materialized: repos={}/{} commits={}/{} elapsed={:.1}s",
            commit_refresh.repos_processed,
            commit_refresh.repos_total,
            commit_refresh.commits_processed,
            commit_refresh.commits_total,
            commit_refresh.elapsed_ms as f64 / 1000.0
        );
    }

    progress.finish();

    let github_token_configured = github::auth::github_token_source()?.is_some();
    let github_sync_plan = if github_token_configured {
        Some(github::sync::plan_github_pull_requests(&db)?)
    } else {
        None
    };
    let github_summary = match github_sync_plan {
        Some(plan) if plan.total_units() > 0 => {
            let mut github_progress = IngestProgress::new_for_total_units(plan.total_units());
            let result = {
                let result = {
                    let mut observer = github_progress.stage("GitHub Sync", plan.total_units());
                    github::sync::sync_github_pull_requests(&mut db, verbose, Some(&mut observer))
                };
                github_progress.finish_stage();
                result
            };
            result?
        }
        _ => github::sync::sync_github_pull_requests(&mut db, verbose, None)?,
    };
    let github_updated = github_summary.commit_lookups_enqueued > 0
        || github_summary.open_pull_requests_refreshed > 0
        || github_summary.issue_scans_enqueued > 0;
    let compact_github_status = if verbose {
        None
    } else if github_updated {
        Some("GitHub sync: updated")
    } else if github_token_configured {
        (github_summary.repos_considered > 0).then_some("GitHub sync: no updates")
    } else {
        Some("GitHub sync: skipped")
    };
    if verbose {
        if github_updated {
            println!(
                "GitHub PR sync: repos={} lookups={}/{} resolved={} no_pr={} failed={} rate_limited={} prs={} pr_commits={} refreshed_open_prs={} issue_scans={}/{} refreshed_issue_scans={} issues={} issue_fix_prs={} pr_removed_hashes={}",
                github_summary.repos_considered,
                github_summary.commit_lookups_completed,
                github_summary.commit_lookups_enqueued,
                github_summary.resolved_commits,
                github_summary.no_pr_commits,
                github_summary.failed_commits,
                github_summary.rate_limited_commits,
                github_summary.pull_requests_upserted,
                github_summary.pull_request_commits_upserted,
                github_summary.open_pull_requests_refreshed,
                github_summary.issue_scans_completed,
                github_summary.issue_scans_enqueued,
                github_summary.issue_scans_refreshed,
                github_summary.issues_upserted,
                github_summary.issue_fix_pull_requests_upserted,
                github_summary.pull_request_removed_hashes_upserted
            );
        } else if github_summary.repos_considered > 0 {
            if github_token_configured {
                println!("GitHub PR sync: no pending GitHub updates for eligible repos");
            } else {
                println!(
                    "GitHub PR sync: skipped remote fetch (run `paceflow github token` or set PACEFLOW_GITHUB_TOKEN to enable refresh)"
                );
            }
        }
    }
    if github_commit_evidence_changed(&github_summary) {
        if verbose {
            println!("\nRe-materializing commit events with GitHub evidence ...");
        }
        let commit_refresh = analytics::refresh_commit_events(&mut db, verbose, None)?;
        if verbose {
            println!(
                "Commit events materialized: repos={}/{} commits={}/{} elapsed={:.1}s",
                commit_refresh.repos_processed,
                commit_refresh.repos_total,
                commit_refresh.commits_processed,
                commit_refresh.commits_total,
                commit_refresh.elapsed_ms as f64 / 1000.0
            );
        }
    } else if verbose && github_summary.repos_considered > 0 {
        println!("\nSkipping re-materialization: GitHub sync produced no commit evidence changes");
    }
    if let Some(status) = compact_github_status {
        println!("{status}");
    }
    println!("Ingest complete");
    println!("Rows written: {}", grand_total);
    Ok(())
}

fn format_provider_stage(provider_name: &str, phase: &str) -> String {
    format!("{provider_name} {phase}")
}

fn estimate_execution_units(
    db: &Connection,
    provider_plans: &[ProviderWorkPlan],
) -> Result<(usize, usize)> {
    let existing_repo_count: i64 = db.query_row(
        "SELECT COUNT(DISTINCT repo_root)
         FROM fact_session_code_change
         WHERE repo_root IS NOT NULL AND TRIM(repo_root) != ''",
        [],
        |row| row.get(0),
    )?;
    let existing_commit_count: i64 =
        db.query_row("SELECT COUNT(*) FROM fact_commit", [], |row| row.get(0))?;
    let existing_repo_count = existing_repo_count.max(0) as usize;
    let existing_commit_count = existing_commit_count.max(0) as usize;

    let planned_work_units = provider_plans
        .iter()
        .map(|plan| plan.session_units() + plan.code_change_units())
        .sum::<usize>();
    let estimated_repo_count = existing_repo_count.max(usize::from(planned_work_units > 0));
    let estimated_commit_count = existing_commit_count
        .max(planned_work_units)
        .max(estimated_repo_count * 25);
    let association_units_estimate = (estimated_repo_count * 2) + estimated_commit_count;
    let commit_materialization_units_estimate = estimated_repo_count + estimated_commit_count;

    Ok((
        association_units_estimate,
        commit_materialization_units_estimate,
    ))
}

fn count_commit_materialization_units(db: &Connection) -> Result<usize> {
    let commit_count: i64 =
        db.query_row("SELECT COUNT(*) FROM fact_commit", [], |row| row.get(0))?;
    let repo_count: i64 = db.query_row(
        "SELECT COUNT(DISTINCT repo_root)
         FROM fact_commit
         WHERE repo_root IS NOT NULL AND TRIM(repo_root) != ''",
        [],
        |row| row.get(0),
    )?;
    Ok((commit_count.max(0) + repo_count.max(0)) as usize)
}

fn github_commit_evidence_changed(summary: &github::types::GitHubSyncSummary) -> bool {
    summary.commit_lookups_completed > 0
        || summary.open_pull_requests_refreshed > 0
        || summary.issue_scans_refreshed > 0
        || summary.issue_fix_pull_requests_upserted > 0
        || summary.pull_request_removed_hashes_upserted > 0
}
