// Copyright 2026 TiKV Project Authors. Licensed under Apache-2.0.

// #[PerformanceCriticalPath]
use collections::HashMap;
use txn_types::TimeStamp;

use crate::storage::{
    Snapshot, Statistics,
    txn::{
        ProcessResult, Result,
        async_batch_resolve_lock::AsyncResolveLockTask,
        commands::{Command, CommandExt, ReadCommand, TypedCommand},
    },
};

command! {
    /// Dispatch a region-level resolve lock request to the async batch worker.
    ResolveLockAsync:
        cmd_ty => (),
        display => "kv::resolve_lock_async", (),
        content => {
            /// Maps lock_ts to commit_ts. See ./resolve_lock.rs for details.
            txn_status: HashMap<TimeStamp, TimeStamp>,
        }
}

impl CommandExt for ResolveLockAsync {
    ctx!();
    tag!(resolve_lock_async);
    request_type!(KvResolveLock);
    property!(readonly);
    property!(is_sys_cmd);

    fn write_bytes(&self) -> usize {
        0
    }

    gen_lock!(empty);
}

#[maybe_async::async_trait]
impl<S: Snapshot + 'static> ReadCommand<S> for ResolveLockAsync {
    #[maybe_async]
    async fn process_read(
        self,
        _snapshot: S,
        _statistics: &mut Statistics,
    ) -> Result<ProcessResult> {
        if self.txn_status.is_empty() {
            return Ok(ProcessResult::Res);
        }
        // A successful foreground response for this command only means the
        // region-level resolve signal has been accepted for best-effort
        // background processing. The actual scan and resolve run later and can
        // still fail due to stale leader/epoch or other region errors.
        Ok(ProcessResult::AsyncResolveLockTask {
            task: AsyncResolveLockTask::new(self.ctx, self.txn_status),
        })
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::storage::{Engine, TestEngineBuilder};

    #[test]
    fn test_process_read_returns_res_for_empty_txn_status() {
        let mut engine = TestEngineBuilder::new().build().unwrap();
        let snapshot = engine.snapshot(Default::default()).unwrap();
        let mut statistics = Statistics::default();
        let command = match ResolveLockAsync::new(HashMap::default(), Default::default()).cmd {
            Command::ResolveLockAsync(command) => command,
            _ => unreachable!(),
        };

        let result = command.process_read(snapshot, &mut statistics).unwrap();
        assert!(matches!(result, ProcessResult::Res));
    }
}