use std::collections::{HashMap, HashSet};
use std::fs;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Instant;

use anyhow::{Context, Result};
use axum::body::Bytes;
use tokio::sync::{Mutex, watch};

use crate::catalog::{self, Repository};
use crate::generate::{self, Config as GenerateConfig, STATE_FILE};
use crate::render::{catalog_index, user_index};

use super::{AppState, Inner};

#[derive(Default)]
pub(super) struct Snapshot {
    pub(super) catalog_html: Bytes,
    pub(super) user_html: HashMap<String, Bytes>,
    pub(super) repos: HashMap<String, RepoEntry>,
}

#[derive(Clone)]
pub(super) struct RepoEntry {
    pub(super) repository: Repository,
    build_gate: Arc<Mutex<()>>,
}

#[derive(Clone, Copy, Default)]
struct BuildQueueState {
    running: bool,
    priority_waiting: usize,
}

/// Serializes site builds: each already fans out over every core with
/// rayon, so concurrent builds only thrash. Human-triggered builds claim
/// the slot ahead of background rebuilds, and cancel the background build
/// currently holding it; the scanner re-queues whatever it abandons.
pub(super) struct BuildQueue {
    state: watch::Sender<BuildQueueState>,
    background: std::sync::Mutex<Option<Arc<AtomicBool>>>,
}

pub(super) struct BuildPermit<'a> {
    queue: &'a BuildQueue,
}

impl BuildQueue {
    pub(super) fn new() -> Self {
        Self {
            state: watch::channel(BuildQueueState::default()).0,
            background: std::sync::Mutex::new(None),
        }
    }

    fn register_background(&self) -> Arc<AtomicBool> {
        let flag = Arc::new(AtomicBool::new(false));
        *self.background.lock().unwrap_or_else(|e| e.into_inner()) = Some(flag.clone());
        flag
    }

    fn clear_background(&self, flag: &Arc<AtomicBool>) {
        let mut background = self.background.lock().unwrap_or_else(|e| e.into_inner());
        if background.as_ref().is_some_and(|active| Arc::ptr_eq(active, flag)) {
            *background = None;
        }
    }

    async fn acquire(&self, priority: bool) -> BuildPermit<'_> {
        if priority {
            self.state
                .send_modify(|state| state.priority_waiting += 1);
            let background = self.background.lock().unwrap_or_else(|e| e.into_inner());
            if let Some(flag) = background.as_ref() {
                flag.store(true, Ordering::Relaxed);
            }
        }
        let mut seen = self.state.subscribe();
        loop {
            let mut claimed = false;
            self.state.send_modify(|state| {
                if !state.running && (priority || state.priority_waiting == 0) {
                    state.running = true;
                    if priority {
                        state.priority_waiting -= 1;
                    }
                    claimed = true;
                }
            });
            if claimed {
                return BuildPermit { queue: self };
            }
            seen.borrow_and_update();
            let _ = seen.changed().await;
        }
    }
}

impl Drop for BuildPermit<'_> {
    fn drop(&mut self) {
        self.queue
            .state
            .send_modify(|state| state.running = false);
    }
}

pub(super) async fn scan_periodically(state: AppState) {
    let mut tick = tokio::time::interval(state.config.check_interval);
    tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
    tick.tick().await; // the first tick is immediate; run() already scanned
    loop {
        tokio::select! {
            _ = tick.tick() => {},
            _ = state.rescan_requested.notified() => {},
        }
        if let Err(error) = state.rescan().await {
            eprintln!("sorceryd: rescan: {error}");
        }
    }
}

impl Inner {
    /// Discover repositories and publish a fresh immutable catalog snapshot.
    pub(super) async fn rescan(&self) -> Result<()> {
        let _gate = self.rescan_gate.lock().await;
        let config = self.config.clone();
        let (repositories, freshness) = tokio::task::spawn_blocking(move || -> Result<_> {
            let repositories = catalog::discover(&config.repositories)?;
            let freshness = repositories
                .iter()
                .map(|repository| is_current(&generate_config(&config, repository)))
                .collect::<Vec<_>>();
            Ok((repositories, freshness))
        })
        .await??;

        let old = self.snapshot.load_full();
        let mut repos = HashMap::with_capacity(repositories.len());
        let mut stale = Vec::new();
        for (repository, current) in repositories.iter().zip(freshness) {
            let current = match current {
                Ok(current) => current,
                Err(error) => {
                    eprintln!(
                        "sorceryd: skipping {}/{}: {error}",
                        repository.user, repository.name
                    );
                    continue;
                }
            };
            let key = format!("{}/{}", repository.user, repository.name);
            let build_gate = old
                .repos
                .get(&key)
                .map(|entry| entry.build_gate.clone())
                .unwrap_or_else(|| Arc::new(Mutex::new(())));
            let entry = RepoEntry {
                repository: repository.clone(),
                build_gate,
            };
            if !current {
                stale.push(entry.clone());
            }
            repos.insert(key, entry);
        }

        let instance_name = &self.config.instance_name;
        let user_html = repositories
            .chunk_by(|a, b| a.user == b.user)
            .map(|repos| {
                let user = &repos[0].user;
                (user.clone(), user_index(instance_name, user, repos).into())
            })
            .collect();
        self.snapshot.store(Arc::new(Snapshot {
            catalog_html: catalog_index(instance_name, &repositories).into(),
            user_html,
            repos,
        }));
        self.background.send_replace(stale);
        Ok(())
    }

    pub(super) async fn build(&self, entry: &RepoEntry, priority: bool) -> Result<bool> {
        let background = (!priority).then(|| self.build_queue.register_background());
        let permit = self.build_queue.acquire(priority).await;
        let _gate = entry.build_gate.lock().await;
        let _cache = self.cache_lock.read().await;
        let config = generate_config(&self.config, &entry.repository);
        let highlights = self.highlights.clone();
        let cancel = background.clone();
        let result =
            tokio::task::spawn_blocking(move || ensure_current(&config, highlights, cancel)).await;
        if let Some(flag) = &background {
            self.build_queue.clear_background(flag);
        }
        drop(permit);
        result?
    }
}

pub(super) async fn background_worker(state: AppState, mut stale: watch::Receiver<Vec<RepoEntry>>) {
    while stale.changed().await.is_ok() {
        let entries = stale.borrow_and_update().clone();
        for entry in entries {
            let started = Instant::now();
            let name = format!("{}/{}", entry.repository.user, entry.repository.name);
            match state.build(&entry, false).await {
                Ok(true) => eprintln!(
                    "sorceryd: rebuilt {name} in {:.2}s",
                    started.elapsed().as_secs_f64(),
                ),
                Ok(false) => {}
                Err(error) => eprintln!("sorceryd: background rebuild of {name}: {error}"),
            }
        }
        match garbage_collect(&state).await {
            Ok(None | Some(0)) => {}
            Ok(Some(count)) => {
                eprintln!("sorceryd: removed {count} stale cache directories")
            }
            Err(error) => eprintln!("sorceryd: cache garbage collection: {error}"),
        }
    }
}

async fn garbage_collect(state: &Inner) -> Result<Option<usize>> {
    let Ok(_cache) = state.cache_lock.try_write() else {
        return Ok(None);
    };
    let cache = state.config.cache.clone();
    let live = state
        .snapshot
        .load()
        .repos
        .values()
        .map(|entry| state.config.repo_cache(&entry.repository))
        .collect::<HashSet<_>>();
    tokio::task::spawn_blocking(move || garbage_collect_cache(&cache, &live))
        .await?
        .map(Some)
}

fn garbage_collect_cache(cache: &Path, live: &HashSet<PathBuf>) -> Result<usize> {
    let mut removed = 0;
    for user in fs::read_dir(cache)? {
        let user = user?;
        if !user.file_type()?.is_dir() {
            continue;
        }
        for repo in fs::read_dir(user.path())? {
            let repo = repo?;
            if !repo.file_type()?.is_dir() {
                continue;
            }
            let path = repo.path();
            if path.join(STATE_FILE).is_file() && !live.contains(&path) {
                fs::remove_dir_all(path)?;
                removed += 1;
            }
        }
    }
    Ok(removed)
}

fn generate_config(config: &super::Config, repository: &Repository) -> GenerateConfig {
    GenerateConfig {
        repo: repository.path.clone(),
        out: config.repo_cache(repository),
        instance_name: config.instance_name.clone(),
        name: Some(format!("{}/{}", repository.user, repository.name)),
        clone_url: config.site_url(repository),
    }
}

fn is_current(config: &GenerateConfig) -> Result<bool> {
    let expected = generate::expected_state(config)?;
    Ok(fs::read_to_string(config.out.join(STATE_FILE))
        .ok()
        .as_deref()
        == Some(&expected))
}

/// Regenerate the cache directory if its recorded state differs from the
/// repository's, building into a staging sibling and swapping atomically.
fn ensure_current(
    config: &GenerateConfig,
    highlights: Arc<crate::highlight::Cache>,
    cancel: Option<Arc<AtomicBool>>,
) -> Result<bool> {
    if is_current(config)? {
        return Ok(false);
    }

    let parent = config
        .out
        .parent()
        .context("cache repository has no parent")?;
    let name = config
        .out
        .file_name()
        .context("cache repository has no name")?;
    fs::create_dir_all(parent)?;
    let staging = parent.join(format!(
        ".{}.new-{}",
        name.to_string_lossy(),
        std::process::id()
    ));
    if staging.exists() {
        fs::remove_dir_all(&staging)?;
    }
    let mut staging_config = config.clone();
    staging_config.out = staging.clone();
    if let Err(error) = generate::full_cancellable(
        &staging_config,
        Some(&config.out),
        highlights,
        cancel.clone(),
    ) {
        // A human-triggered build wanted the machine; the scanner will
        // re-queue this repository for a later background attempt.
        if cancel
            .as_ref()
            .is_some_and(|flag| flag.load(Ordering::Relaxed))
        {
            return Ok(false);
        }
        return Err(error);
    }
    // keep info/refs + objects/info/packs current for static git-dir access
    let status = std::process::Command::new("git")
        .arg("-C")
        .arg(&config.repo)
        .arg("update-server-info")
        .status()?;
    if !status.success() {
        eprintln!(
            "sorceryd: update-server-info failed for {}",
            config.repo.display()
        );
    }
    generate::swap_dir(&staging, &config.out)?;
    Ok(true)
}

#[cfg(test)]
mod tests {
    use std::collections::HashSet;
    use std::fs;
    use std::sync::Arc;
    use std::sync::atomic::Ordering;

    use anyhow::Result;

    use crate::generate::STATE_FILE;
    use crate::testutil::TempDir;

    use super::{BuildQueue, garbage_collect_cache};

    #[tokio::test]
    async fn priority_builds_claim_the_slot_ahead_of_background_builds() {
        let queue = Arc::new(BuildQueue::new());
        let first = queue.acquire(false).await;
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        for (priority, label) in [(false, "background"), (true, "priority")] {
            let queue = queue.clone();
            let tx = tx.clone();
            tokio::spawn(async move {
                let _permit = queue.acquire(priority).await;
                tx.send(label).unwrap();
            });
            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
        }
        drop(first);
        assert_eq!(rx.recv().await, Some("priority"));
        assert_eq!(rx.recv().await, Some("background"));
    }

    #[tokio::test]
    async fn priority_builds_cancel_the_active_background_build() {
        let queue = BuildQueue::new();
        let flag = queue.register_background();
        let _permit = queue.acquire(true).await;
        assert!(flag.load(Ordering::Relaxed));
        queue.clear_background(&flag);
    }

    #[test]
    fn garbage_collects_only_inactive_site_caches() -> Result<()> {
        let root = TempDir::new("cache-gc");
        let live = root.join("alice/live");
        let stale = root.join("alice/stale");
        let staging = root.join("alice/.live.new-1");
        let asset = root.join("fonts/inter");
        for path in [&live, &stale, &staging] {
            fs::create_dir_all(path)?;
            fs::write(path.join(STATE_FILE), "state")?;
        }
        fs::create_dir_all(&asset)?;

        assert_eq!(
            garbage_collect_cache(&root, &HashSet::from([live.clone()]))?,
            2
        );
        assert!(live.is_dir());
        assert!(!stale.exists());
        assert!(!staging.exists());
        assert!(asset.is_dir());
        Ok(())
    }
}