// Copyright 2026 PingCAP, Inc.

use std::{
    fs::OpenOptions,
    io::Write,
    path::{Path, PathBuf},
    sync::{Arc, Mutex},
    time::Duration,
};

use tikv_util::time::Instant;

/// Performs one blocking `write + sync_all` probe on the given path.
pub(crate) fn write_sync_once(path: &Path, payload: &[u8]) -> std::io::Result<Duration> {
    let start = Instant::now();
    let mut file = OpenOptions::new()
        .create(true)
        .write(true)
        .truncate(true)
        .open(path)?;
    file.write_all(payload)?;
    file.sync_all()?;
    Ok(start.saturating_elapsed())
}

#[derive(Debug)]
struct ProbeState {
    current_probe_started_at: Option<Instant>,
    last_success_at: Option<Instant>,
    failure_count_since_last_success: u64,
}

impl ProbeState {
    fn new() -> Self {
        Self {
            current_probe_started_at: None,
            last_success_at: None,
            failure_count_since_last_success: 0,
        }
    }

    fn try_start_probe(&mut self) -> bool {
        if self.current_probe_started_at.is_some() {
            return false;
        }
        self.current_probe_started_at = Some(Instant::now());
        true
    }

    fn finish_probe_success(&mut self) {
        self.current_probe_started_at = None;
        self.last_success_at = Some(Instant::now());
        self.failure_count_since_last_success = 0;
    }

    fn finish_probe_failure(&mut self) {
        self.current_probe_started_at = None;
        self.failure_count_since_last_success += 1;
    }

    fn reset_failure_window(&mut self) {
        // A new enable period must earn fail-fast eligibility with a fresh
        // successful probe.
        let now = Instant::now();
        if self.current_probe_started_at.is_some() {
            self.current_probe_started_at = Some(now);
        }
        self.last_success_at = None;
        self.failure_count_since_last_success = 0;
    }

    fn current_probe_elapsed(&self) -> Option<Duration> {
        self.current_probe_started_at
            .map(|start| start.saturating_elapsed())
    }

    fn time_since_last_success(&self) -> Option<Duration> {
        self.last_success_at.map(|t| t.saturating_elapsed())
    }

    fn failure_count_since_last_success(&self) -> u64 {
        self.failure_count_since_last_success
    }
}

#[derive(Clone, Debug)]
pub(crate) struct ProbeRunner {
    path: PathBuf,
    payload: &'static [u8],
    state: Arc<Mutex<ProbeState>>,
}

impl ProbeRunner {
    pub(crate) fn new(path: PathBuf, payload: &'static [u8]) -> Self {
        Self {
            path,
            payload,
            state: Arc::new(Mutex::new(ProbeState::new())),
        }
    }

    pub(crate) fn path(&self) -> &Path {
        &self.path
    }

    pub(crate) fn probe_once(&self) -> std::io::Result<Duration> {
        write_sync_once(self.path(), self.payload)
    }

    pub(crate) fn try_start_probe(&self) -> bool {
        self.state.lock().unwrap().try_start_probe()
    }

    pub(crate) fn finish_probe_success(&self) {
        self.state.lock().unwrap().finish_probe_success();
    }

    pub(crate) fn finish_probe_failure(&self) {
        self.state.lock().unwrap().finish_probe_failure();
    }

    pub(crate) fn reset_failure_window(&self) {
        self.state.lock().unwrap().reset_failure_window();
    }

    pub(crate) fn current_probe_elapsed(&self) -> Option<Duration> {
        self.state.lock().unwrap().current_probe_elapsed()
    }

    pub(crate) fn time_since_last_success(&self) -> Option<Duration> {
        self.state.lock().unwrap().time_since_last_success()
    }

    pub(crate) fn failure_count_since_last_success(&self) -> u64 {
        self.state
            .lock()
            .unwrap()
            .failure_count_since_last_success()
    }
}

#[cfg(test)]
mod tests {
    use tempfile::tempdir;

    use super::*;

    #[test]
    fn test_probe_runner_probe_once() {
        let dir = tempdir().unwrap();
        let runner = ProbeRunner::new(dir.path().join("probe.tmp"), b"payload");

        let duration = runner.probe_once().unwrap();
        assert!(duration > Duration::ZERO);
        assert!(runner.path().exists());
    }

    #[test]
    fn test_probe_runner_state() {
        let dir = tempdir().unwrap();
        let runner = ProbeRunner::new(dir.path().join("probe.tmp"), b"payload");

        assert!(runner.time_since_last_success().is_none());
        assert!(runner.current_probe_elapsed().is_none());
        assert!(runner.try_start_probe());
        assert!(!runner.try_start_probe());
        assert!(runner.current_probe_elapsed().is_some());
        runner.finish_probe_failure();
        assert!(runner.current_probe_elapsed().is_none());

        assert!(runner.try_start_probe());
        assert!(runner.current_probe_elapsed().is_some());
        runner.finish_probe_success();
        assert!(runner.current_probe_elapsed().is_none());
        assert!(runner.time_since_last_success().is_some());
        assert_eq!(runner.failure_count_since_last_success(), 0);

        assert!(runner.try_start_probe());
        runner.finish_probe_failure();
        assert!(runner.time_since_last_success().is_some());
        assert_eq!(runner.failure_count_since_last_success(), 1);

        runner.reset_failure_window();
        assert!(runner.time_since_last_success().is_none());
        assert_eq!(runner.failure_count_since_last_success(), 0);
    }
}