// Copyright 2026 PingCAP, Inc.

use std::{io, path::Path};

use tikv_util::{box_try, info, warn};

use crate::Result;

/// Resets the existing merged rfengine directory for the bootstrap path.
///
/// When the manifest is empty, merged_engine will bootstrap by calling
/// `recover_from_backup`. In this mode, any leftover data in
/// `<local_dir>/merged` is invalid input: stale peers can show up in
/// `raft.get_region_peer_map()` but have no corresponding `RegionProgress`,
/// leading to a panic during startup.
///
/// Design (issue #4818):
/// - If `<local_dir>/merged` exists and `<local_dir>/merged.stale` does not,
///   rename `merged` -> `merged.stale` for troubleshooting.
/// - If `merged.stale` already exists, keep it and remove `merged`.
/// - Recreate an empty `merged` directory for the new rfengine instance.
///
/// The caller must ensure `merged_dir` already exists.
pub(crate) fn reset_merged_dir_on_bootstrap(merged_dir: &Path, stale_dir: &Path) -> Result<()> {
    if box_try!(stale_dir.try_exists()) {
        info!(
            "bootstrap: merged.stale already exists, remove merged dir";
            "merged" => ?merged_dir,
            "stale" => ?stale_dir,
        );
        remove_path_recursively(merged_dir)?;
    } else {
        info!(
            "bootstrap: rename merged dir to merged.stale";
            "merged" => ?merged_dir,
            "stale" => ?stale_dir,
        );
        match std::fs::rename(merged_dir, stale_dir) {
            Ok(()) => {}
            Err(err) if should_keep_existing_stale_dir(&err) => {
                // Best-effort: keep the existing stale dir and drop the current merged dir.
                warn!(
                    "bootstrap: merged.stale appeared during rename, remove merged dir";
                    "merged" => ?merged_dir,
                    "stale" => ?stale_dir,
                    "err" => ?err,
                );
                remove_path_recursively(merged_dir)?;
            }
            Err(err) => return Err(err.into()),
        }
    }

    // Recreate an empty dir for the new rfengine instance.
    std::fs::create_dir_all(merged_dir)?;
    Ok(())
}

fn should_keep_existing_stale_dir(err: &io::Error) -> bool {
    matches!(
        err.kind(),
        io::ErrorKind::AlreadyExists | io::ErrorKind::DirectoryNotEmpty
    )
}

fn remove_path_recursively(path: &Path) -> Result<()> {
    match std::fs::symlink_metadata(path) {
        Ok(meta) => {
            if meta.is_dir() {
                std::fs::remove_dir_all(path)?;
            } else {
                std::fs::remove_file(path)?;
            }
            Ok(())
        }
        Err(err) if err.kind() == io::ErrorKind::NotFound => Ok(()),
        Err(err) => Err(err.into()),
    }
}

#[cfg(test)]
mod tests {
    use std::fs;

    use tempfile::tempdir;

    use super::*;

    #[test]
    fn test_reset_merged_dir_on_bootstrap_renames_first() {
        let dir = tempdir().unwrap();
        let local_dir = dir.path();
        let merged_dir = local_dir.join("merged");
        let stale_dir = local_dir.join("merged.stale");

        fs::create_dir_all(&merged_dir).unwrap();
        fs::write(merged_dir.join("marker"), b"x").unwrap();

        reset_merged_dir_on_bootstrap(&merged_dir, &stale_dir).unwrap();

        assert!(merged_dir.is_dir());
        assert!(stale_dir.is_dir());
        assert!(!merged_dir.join("marker").exists());
        assert!(stale_dir.join("marker").exists());
    }

    #[test]
    fn test_reset_merged_dir_on_bootstrap_keeps_first_stale() {
        let dir = tempdir().unwrap();
        let local_dir = dir.path();
        let merged_dir = local_dir.join("merged");
        let stale_dir = local_dir.join("merged.stale");

        fs::create_dir_all(&stale_dir).unwrap();
        fs::write(stale_dir.join("first"), b"a").unwrap();

        fs::create_dir_all(&merged_dir).unwrap();
        fs::write(merged_dir.join("second"), b"b").unwrap();

        reset_merged_dir_on_bootstrap(&merged_dir, &stale_dir).unwrap();

        assert!(merged_dir.is_dir());
        assert!(stale_dir.is_dir());
        assert!(stale_dir.join("first").exists());
        assert!(!stale_dir.join("second").exists());
        assert!(!merged_dir.join("second").exists());
    }

    #[test]
    fn test_should_keep_existing_stale_dir_errors() {
        assert!(should_keep_existing_stale_dir(&io::Error::from(
            io::ErrorKind::AlreadyExists,
        )));
        assert!(should_keep_existing_stale_dir(&io::Error::from(
            io::ErrorKind::DirectoryNotEmpty,
        )));
        assert!(!should_keep_existing_stale_dir(&io::Error::from(
            io::ErrorKind::PermissionDenied,
        )));
    }
}