use std::path::Path;

use anyhow::Result;
use serde::{Deserialize, Serialize};
use tracing::debug;

use super::{
    process_environment_fingerprint, sha256_bytes, sha256_json, workspace_fingerprint,
    CachedCheckpointPayload, CheckpointResult, EvaluationResult, CHECKPOINT_SCHEMA_VERSION,
    CONTEXT_CHECK_THROTTLE_WINDOW_SECONDS, HANDOVER_SCHEMA_VERSION,
};
use crate::commands::check;
use crate::db;
use crate::paths::state::StateLayout;
use crate::state::compiled as compiled_state;
use crate::state::consistency;
use crate::state::continuity as runtime_state;
use crate::state::escalation as escalation_state;
use crate::state::session as session_state;
use crate::state::work_stream_decay;

const SCHEMA_VERSION: u32 = 1;
// Keep cache entries at least as long-lived as low-value context-check
// throttling, otherwise suppressed polls still recompute checkpoint state.
const MAX_AGE_SECONDS: u64 = CONTEXT_CHECK_THROTTLE_WINDOW_SECONDS;

#[derive(Clone, Serialize, Deserialize)]
pub(super) struct Metadata {
    schema_version: u32,
    cache_key: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    session_id: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    session_revision: Option<u64>,
    pub(super) projection_digests: compiled_state::ProjectionDigests,
    pub(super) policy_digest: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(super) tracked_session: Option<session_state::SessionStateFile>,
}

#[derive(Clone)]
pub(super) struct Probe {
    pub(super) cache_key: String,
    profile: String,
    session_id: Option<String>,
    session_revision: Option<u64>,
}

struct CheckSurfaceProbe {
    fingerprint: String,
    has_registered_checks: bool,
}

pub(super) fn checkpoint_probe(
    repo_root: &Path,
    layout: &StateLayout,
    profile_name: &str,
    locality_id: &str,
) -> Result<Probe> {
    let (source_fingerprint, raw) =
        quick_runtime_source_fingerprint(repo_root, layout, locality_id)?;
    let shared_db = db::StateDb::try_open_for_layout(layout)?;
    let (tracked_session, tracked_activity, active_session_id) = match shared_db.as_ref() {
        Some(db) => (
            session_state::load_from_db(Some(db))?,
            session_state::load_activity_from_db(Some(db))?,
            session_state::load_session_id_from_db(Some(db))?,
        ),
        None => (
            session_state::load_for_layout(layout)?,
            session_state::load_activity_for_layout(layout)?,
            session_state::load_session_id(layout)?,
        ),
    };
    let work_stream_decay =
        work_stream_decay::load_view_from_db(shared_db.as_ref(), active_session_id.as_deref())?;
    let escalation_entries = if let Some(db) = shared_db.as_ref() {
        escalation_state::load_from_db(Some(db))?
    } else {
        escalation_state::load_for_layout(layout)?
    };
    let latest_projection_observation = consistency::load_latest_projection_observation(layout)?;
    let handoff_config = layout.effective_handoff_config(locality_id)?;
    let workspace_fingerprint = workspace_fingerprint(repo_root, layout)?;
    let lifeloop_renewal = crate::state::lifeloop_renewal::load_view(repo_root);

    let key = serde_json::json!({
        "schema_version": SCHEMA_VERSION,
        "kind": db::radar_cache::KIND_CHECKPOINT,
        "command_schema_version": CHECKPOINT_SCHEMA_VERSION,
        "repo_root": repo_root.display().to_string(),
        "profile": profile_name,
        "locality_id": locality_id,
        "session": tracked_session,
        "session_activity": tracked_activity,
        "active_session_id": active_session_id,
        "source_fingerprint": source_fingerprint,
        "workspace_fingerprint": workspace_fingerprint,
        "recovery": raw.recovery.state,
        "escalation_entries": escalation_entries,
        "work_stream_decay": work_stream_decay,
        "latest_projection_observation": latest_projection_observation,
        "handoff_candidate_refresh_enabled": handoff_config.candidate_refresh_enabled(),
        "lifeloop_renewal": lifeloop_renewal,
    });

    Ok(Probe {
        cache_key: sha256_json(&key)?,
        profile: profile_name.to_owned(),
        session_id: tracked_session
            .as_ref()
            .and_then(|session| session.session_id.clone()),
        session_revision: tracked_session.as_ref().map(|session| session.revision),
    })
}

pub(super) fn evaluation_probe(checkpoint: &CheckpointResult) -> Result<Probe> {
    let repo_root = Path::new(&checkpoint.path);
    let workspace_fingerprint = workspace_fingerprint(repo_root, &checkpoint.layout)?;
    let check_surface = check_surface_probe(repo_root)?;
    let check_environment_fingerprint = check_surface
        .has_registered_checks
        .then(process_environment_fingerprint);
    let checkpoint_digest = stable_checkpoint_digest(checkpoint)?;
    let key = serde_json::json!({
        "schema_version": SCHEMA_VERSION,
        "kind": db::radar_cache::KIND_EVALUATION,
        "command_schema_version": HANDOVER_SCHEMA_VERSION,
        "checkpoint_digest": checkpoint_digest,
        "workspace_fingerprint": workspace_fingerprint,
        "check_surface_fingerprint": check_surface.fingerprint,
        "check_environment_fingerprint": check_environment_fingerprint,
        "profile": checkpoint.profile,
        "locality_id": checkpoint.locality_id,
        "session_id": checkpoint.tracked_session.as_ref().and_then(|session| session.session_id.clone()),
        "session_revision": checkpoint.tracked_session.as_ref().map(|session| session.revision),
    });

    Ok(Probe {
        cache_key: sha256_json(&key)?,
        profile: checkpoint.profile.clone(),
        session_id: checkpoint
            .tracked_session
            .as_ref()
            .and_then(|session| session.session_id.clone()),
        session_revision: checkpoint
            .tracked_session
            .as_ref()
            .map(|session| session.revision),
    })
}

pub(super) fn read_checkpoint(
    layout: &StateLayout,
    probe: &Probe,
) -> Result<Option<CheckpointResult>> {
    let Some(db) = db::StateDb::try_open_for_layout(layout)? else {
        return Ok(None);
    };
    let Some(entry) = db::radar_cache::read(
        db.conn(),
        db::radar_cache::KIND_CHECKPOINT,
        &probe.cache_key,
    )?
    else {
        return Ok(None);
    };
    if !entry_is_fresh(&entry) {
        let _ = db::radar_cache::delete(db.conn(), entry.kind, &entry.cache_key);
        return Ok(None);
    }
    let metadata = match read_metadata(&entry, probe) {
        Some(metadata) => metadata,
        None => {
            let _ = db::radar_cache::delete(db.conn(), entry.kind, &entry.cache_key);
            return Ok(None);
        }
    };
    let payload: CachedCheckpointPayload = match serde_json::from_str(&entry.payload_json) {
        Ok(payload) => payload,
        Err(error) => {
            debug!(kind = entry.kind, cache_key = %entry.cache_key, %error, "ignoring unreadable radar cache payload");
            let _ = db::radar_cache::delete(db.conn(), entry.kind, &entry.cache_key);
            return Ok(None);
        }
    };
    Ok(Some(payload.into_checkpoint(metadata, layout.clone())))
}

pub(super) fn write_checkpoint(
    layout: &StateLayout,
    probe: &Probe,
    checkpoint: &CheckpointResult,
) -> Result<()> {
    let db = db::StateDb::open(&layout.state_db_path())?;
    let observed_at_epoch_s = session_state::now_epoch_s()?;
    db::radar_cache::delete_expired(
        db.conn(),
        observed_at_epoch_s.saturating_sub(MAX_AGE_SECONDS),
    )?;
    let metadata = Metadata {
        schema_version: SCHEMA_VERSION,
        cache_key: probe.cache_key.clone(),
        session_id: probe.session_id.clone(),
        session_revision: probe.session_revision,
        projection_digests: checkpoint.projection_digests.clone(),
        policy_digest: checkpoint.policy_digest.clone(),
        tracked_session: checkpoint.tracked_session.clone(),
    };
    db::radar_cache::upsert(
        db.conn(),
        &db::radar_cache::RadarCacheEntry {
            kind: db::radar_cache::KIND_CHECKPOINT,
            cache_key: probe.cache_key.clone(),
            profile: probe.profile.clone(),
            session_id: probe.session_id.clone(),
            session_revision: probe.session_revision,
            observed_at_epoch_s,
            payload_json: serde_json::to_string(&CachedCheckpointPayload::from_checkpoint(
                checkpoint,
            ))?,
            metadata_json: serde_json::to_string(&metadata)?,
        },
    )
}

pub(super) fn read_evaluation(
    layout: &StateLayout,
    probe: &Probe,
) -> Result<Option<EvaluationResult>> {
    let Some(db) = db::StateDb::try_open_for_layout(layout)? else {
        return Ok(None);
    };
    let Some(entry) = db::radar_cache::read(
        db.conn(),
        db::radar_cache::KIND_EVALUATION,
        &probe.cache_key,
    )?
    else {
        return Ok(None);
    };
    if !entry_is_fresh(&entry) {
        let _ = db::radar_cache::delete(db.conn(), entry.kind, &entry.cache_key);
        return Ok(None);
    }
    if read_metadata(&entry, probe).is_none() {
        let _ = db::radar_cache::delete(db.conn(), entry.kind, &entry.cache_key);
        return Ok(None);
    }
    match serde_json::from_str(&entry.payload_json) {
        Ok(payload) => Ok(Some(payload)),
        Err(error) => {
            debug!(kind = entry.kind, cache_key = %entry.cache_key, %error, "ignoring unreadable radar cache payload");
            let _ = db::radar_cache::delete(db.conn(), entry.kind, &entry.cache_key);
            Ok(None)
        }
    }
}

pub(super) fn write_evaluation(
    layout: &StateLayout,
    probe: &Probe,
    checkpoint: &CheckpointResult,
    evaluation: &EvaluationResult,
) -> Result<()> {
    let db = db::StateDb::open(&layout.state_db_path())?;
    let observed_at_epoch_s = session_state::now_epoch_s()?;
    db::radar_cache::delete_expired(
        db.conn(),
        observed_at_epoch_s.saturating_sub(MAX_AGE_SECONDS),
    )?;
    let metadata = Metadata {
        schema_version: SCHEMA_VERSION,
        cache_key: probe.cache_key.clone(),
        session_id: probe.session_id.clone(),
        session_revision: probe.session_revision,
        projection_digests: checkpoint.projection_digests.clone(),
        policy_digest: checkpoint.policy_digest.clone(),
        tracked_session: checkpoint.tracked_session.clone(),
    };
    db::radar_cache::upsert(
        db.conn(),
        &db::radar_cache::RadarCacheEntry {
            kind: db::radar_cache::KIND_EVALUATION,
            cache_key: probe.cache_key.clone(),
            profile: probe.profile.clone(),
            session_id: probe.session_id.clone(),
            session_revision: probe.session_revision,
            observed_at_epoch_s,
            payload_json: serde_json::to_string(evaluation)?,
            metadata_json: serde_json::to_string(&metadata)?,
        },
    )
}

fn quick_runtime_source_fingerprint(
    repo_root: &Path,
    layout: &StateLayout,
    locality_id: &str,
) -> Result<(String, runtime_state::RawRuntimeSources)> {
    let raw = runtime_state::load_raw_start_runtime_sources(repo_root, layout, locality_id)?;
    let surfaces = runtime_state::RuntimeSourceSurfaces {
        execution_gates: raw.execution_gates.source.clone(),
        handoff: raw.handoff.clone(),
    };
    let source_fingerprint = compiled_state::fingerprint_for_inputs(&surfaces)?;
    Ok((source_fingerprint, raw))
}

fn stable_checkpoint_digest(checkpoint: &CheckpointResult) -> Result<String> {
    let mut value = serde_json::to_value(CachedCheckpointPayload::from_checkpoint(checkpoint))?;
    if let Some(object) = value.as_object_mut() {
        if let Some(projection_telemetry) = object
            .get_mut("projection_telemetry")
            .and_then(|value| value.as_object_mut())
        {
            projection_telemetry.remove("cadence");
        }
    }
    sha256_json(&value)
}

fn entry_is_fresh(entry: &db::radar_cache::RadarCacheEntry) -> bool {
    let Ok(now) = session_state::now_epoch_s() else {
        return false;
    };
    now.saturating_sub(entry.observed_at_epoch_s) <= MAX_AGE_SECONDS
}

fn read_metadata(entry: &db::radar_cache::RadarCacheEntry, probe: &Probe) -> Option<Metadata> {
    let metadata: Metadata = match serde_json::from_str(&entry.metadata_json) {
        Ok(metadata) => metadata,
        Err(error) => {
            debug!(kind = entry.kind, cache_key = %entry.cache_key, %error, "ignoring unreadable radar cache metadata");
            return None;
        }
    };
    if metadata.schema_version != SCHEMA_VERSION || metadata.cache_key != probe.cache_key {
        return None;
    }
    Some(metadata)
}

fn check_surface_probe(repo_root: &Path) -> Result<CheckSurfaceProbe> {
    let surface_path = repo_root.join(check::COMMANDS_DIR);
    let mut entries = Vec::new();
    let mut has_registered_checks = false;
    if surface_path.is_dir() {
        for entry in std::fs::read_dir(&surface_path)? {
            let entry = entry?;
            let path = entry.path();
            let Some(name) = path.file_name().and_then(|name| name.to_str()) else {
                continue;
            };
            if !name.starts_with(check::CHECK_PREFIX) {
                continue;
            }
            let metadata = std::fs::symlink_metadata(&path)?;
            let file_type = metadata.file_type();
            let entry_kind = if file_type.is_file() {
                has_registered_checks = true;
                "file"
            } else if file_type.is_dir() {
                "dir"
            } else if file_type.is_symlink() {
                "symlink"
            } else {
                "other"
            };
            let sha256 = if file_type.is_file() {
                Some(sha256_bytes(&std::fs::read(&path)?))
            } else {
                None
            };
            entries.push(serde_json::json!({
                "name": name,
                "kind": entry_kind,
                "mode": file_mode_fingerprint(&metadata),
                "len": metadata.len(),
                "sha256": sha256,
            }));
        }
    }
    entries.sort_by(|a, b| {
        a.get("name")
            .and_then(|value| value.as_str())
            .cmp(&b.get("name").and_then(|value| value.as_str()))
    });
    Ok(CheckSurfaceProbe {
        fingerprint: sha256_json(&entries)?,
        has_registered_checks,
    })
}

#[cfg(unix)]
fn file_mode_fingerprint(metadata: &std::fs::Metadata) -> u32 {
    use std::os::unix::fs::PermissionsExt;

    metadata.permissions().mode() & 0o777
}

#[cfg(not(unix))]
fn file_mode_fingerprint(metadata: &std::fs::Metadata) -> bool {
    metadata.permissions().readonly()
}
