// Copyright 2026 PingCAP, Inc.

use std::{
    path::{Path, PathBuf},
    sync::atomic::{AtomicU64, Ordering},
};

use native_br::common::TempLocalObject;
use tikv_util::{box_err, warn};

use crate::Result;

const WAL_CACHE_SUBDIR: &str = "cache";

static WAL_SPOOL_TMP_ID: AtomicU64 = AtomicU64::new(0);

#[inline]
pub(crate) fn spool_dir_for_store(working_dir: &Path, store_id: u64) -> PathBuf {
    working_dir
        .join(store_id.to_string())
        .join(WAL_CACHE_SUBDIR)
}

/// Best-effort cleanup for stale WAL fetch cache directories on startup.
///
/// Both cached WAL chunks and spool files are temporary and should not persist
/// across restarts.
pub(crate) fn cleanup_stale_wal_spool_dirs(worker_dir: &Path) {
    let entries = match std::fs::read_dir(worker_dir) {
        Ok(entries) => entries,
        Err(err) => {
            warn!("cleanup wal_spool: read_dir failed"; "dir" => ?worker_dir, "err" => ?err);
            return;
        }
    };
    for entry in entries.flatten() {
        let Ok(file_type) = entry.file_type() else {
            continue;
        };
        if !file_type.is_dir() {
            continue;
        }
        let spool_dir = entry.path().join(WAL_CACHE_SUBDIR);
        if let Err(err) = std::fs::remove_dir_all(&spool_dir) {
            if err.kind() != std::io::ErrorKind::NotFound {
                warn!("cleanup wal_spool: remove_dir_all failed"; "dir" => ?spool_dir, "err" => ?err);
            }
        }
    }
}

pub(crate) fn create_temp_wal_spool_object(
    spool_dir: &Path,
    store_id: u64,
    epoch_id: u32,
    start_off: u64,
) -> Result<TempLocalObject> {
    let tmp_id = WAL_SPOOL_TMP_ID.fetch_add(1, Ordering::Relaxed);
    let path = spool_dir.join(format!(
        "{store_id}_{epoch_id}_{start_off}.{tmp_id}.wal.tmp"
    ));
    TempLocalObject::create(path)
        .map_err(|e| box_err!("create temp wal spool file failed: {:?}", e))
}