// Copyright 2020 TiKV Project Authors. Licensed under Apache-2.0.
// Modifications Copyright 2026 PingCAP, Inc.

use std::sync::mpsc;

use api_version::{KvFormat, test_kv_format_impl};
use cdc::{Task, Validate};
use kvproto::{cdcpb::*, kvrpcpb::*};

use crate::{TestSuite, get_ready_region, new_event_feed, subscribe_until_initialized};

const CDC_FLUSH_OLD_VALUE_METRICS: &str = "cdc_flush_old_value_metrics";

fn with_old_value_metrics_flush_disabled(test: impl FnOnce()) {
    fail::cfg(CDC_FLUSH_OLD_VALUE_METRICS, "return").unwrap();
    tikv_util::defer!(fail::remove(CDC_FLUSH_OLD_VALUE_METRICS));

    test();
}

#[test]
fn test_old_value_cache_hit() {
    with_old_value_metrics_flush_disabled(|| {
        test_kv_format_impl!(test_old_value_cache_hit_impl<ApiV2>);
    });
}

fn test_old_value_cache_hit_impl<F: KvFormat>() {
    let mut suite = TestSuite::new(1, F::TAG);
    let scheduler = suite.endpoints.values().next().unwrap().clone();
    let region = get_ready_region(&mut suite, b"xk");
    let region_id = region.id;
    let mut req = suite.new_changedata_request(region_id);
    req.set_extra_op(ExtraOp::ReadOldValue);
    let (mut req_tx, event_feed_wrap, receive_event) =
        new_event_feed(suite.get_region_cdc_client(region_id));
    subscribe_until_initialized(
        &mut req_tx,
        receive_event.as_ref(),
        &req,
        "test_old_value_cache_hit initial subscribe",
    );
    let (tx, rx) = mpsc::channel();

    // Insert value, simulate INSERT INTO.
    let mut m1 = Mutation::default();
    let k1 = b"xk1".to_vec();
    m1.set_op(Op::Insert);
    m1.key = k1.clone();
    m1.value = b"v1".to_vec();
    suite.must_kv_prewrite(region_id, vec![m1], k1.clone(), 10.into());
    #[cfg(NO_NEXT_GEN_COMPATIBLE)]
    {
        let mut events = receive_event(false).events.to_vec();
        match events.remove(0).event.unwrap() {
            Event_oneof_event::Entries(mut es) => {
                let row = &es.take_entries().to_vec()[0];
                assert_eq!(row.get_value(), b"v1");
                assert_eq!(row.get_old_value(), b"");
                assert_eq!(row.get_type(), EventLogType::Prewrite);
                assert_eq!(row.get_start_ts(), 10);
            }
            other => panic!("unknown event {:?}", other),
        }
    }

    suite.must_kv_commit(region_id, vec![k1], 10.into(), 15.into());
    let mut events = receive_event(false).events.to_vec();
    match events.remove(0).event.unwrap() {
        Event_oneof_event::Entries(mut es) => {
            let row = &es.take_entries().to_vec()[0];
            assert_eq!(row.get_type(), EventLogType::Committed);
            assert_eq!(row.get_commit_ts(), 15);
        }
        other => panic!("unknown event {:?}", other),
    }
    // k1 old value must be cached.
    let tx_ = tx.clone();
    scheduler
        .schedule(Task::Validate(Validate::OldValueCache(Box::new(
            move |old_value_cache| {
                tx_.send((old_value_cache.access_count(), old_value_cache.miss_count()))
                    .unwrap();
            },
        ))))
        .unwrap();
    let (access_count, miss_count) = rx.recv().unwrap();
    assert_eq!(access_count, 1);
    assert_eq!(miss_count, 0);

    // Update a noexist value, simulate INSERT IGNORE INTO.
    let mut m2 = Mutation::default();
    let k2 = b"xk2".to_vec();
    m2.set_op(Op::Put);
    m2.key = k2.clone();
    m2.value = b"v2".to_vec();
    suite.must_kv_prewrite(region_id, vec![m2], k2.clone(), 10.into());
    #[cfg(NO_NEXT_GEN_COMPATIBLE)]
    {
        let mut events = receive_event(false).events.to_vec();
        match events.remove(0).event.unwrap() {
            Event_oneof_event::Entries(mut es) => {
                let row = &es.take_entries().to_vec()[0];
                assert_eq!(row.get_value(), b"v2");
                assert_eq!(row.get_old_value(), b"");
                assert_eq!(row.get_type(), EventLogType::Prewrite);
                assert_eq!(row.get_start_ts(), 10);
            }
            other => panic!("unknown event {:?}", other),
        }
    }

    suite.must_kv_commit(region_id, vec![k2], 10.into(), 15.into());
    let mut events = receive_event(false).events.to_vec();
    match events.remove(0).event.unwrap() {
        Event_oneof_event::Entries(mut es) => {
            let row = &es.take_entries().to_vec()[0];
            assert_eq!(row.get_type(), EventLogType::Committed);
            assert_eq!(row.get_commit_ts(), 15);
        }
        other => panic!("unknown event {:?}", other),
    }

    // k2 old value must be cached.
    let tx_ = tx.clone();
    scheduler
        .schedule(Task::Validate(Validate::OldValueCache(Box::new(
            move |old_value_cache| {
                tx_.send((old_value_cache.access_count(), old_value_cache.miss_count()))
                    .unwrap();
            },
        ))))
        .unwrap();
    let (access_count, miss_count) = rx.recv().unwrap();
    assert_eq!(access_count, 2);
    assert_eq!(miss_count, 0);

    // Update an exist value, simulate UPDATE.
    let mut m2 = Mutation::default();
    let k2 = b"xk2".to_vec();
    m2.set_op(Op::Put);
    m2.key = k2.clone();
    m2.value = b"v3".to_vec();
    suite.must_kv_prewrite(region_id, vec![m2], k2.clone(), 20.into());

    #[cfg(NO_NEXT_GEN_COMPATIBLE)]
    {
        let mut events = receive_event(false).events.to_vec();
        match events.remove(0).event.unwrap() {
            Event_oneof_event::Entries(mut es) => {
                let row = &es.take_entries().to_vec()[0];
                assert_eq!(row.get_value(), b"v3");
                assert_eq!(row.get_old_value(), b"v2");
                assert_eq!(row.get_type(), EventLogType::Prewrite);
                assert_eq!(row.get_start_ts(), 20);
            }
            other => panic!("unknown event {:?}", other),
        }
    }

    suite.must_kv_commit(region_id, vec![k2], 20.into(), 25.into());
    let mut events = receive_event(false).events.to_vec();
    match events.remove(0).event.unwrap() {
        Event_oneof_event::Entries(mut es) => {
            let row = &es.take_entries().to_vec()[0];
            assert_eq!(row.get_type(), EventLogType::Committed);
            assert_eq!(row.get_commit_ts(), 25);
        }
        other => panic!("unknown event {:?}", other),
    }

    // k2 old value must be cached.
    let tx_ = tx;
    scheduler
        .schedule(Task::Validate(Validate::OldValueCache(Box::new(
            move |old_value_cache| {
                tx_.send((old_value_cache.access_count(), old_value_cache.miss_count()))
                    .unwrap();
            },
        ))))
        .unwrap();
    let (access_count, miss_count) = rx.recv().unwrap();
    assert_eq!(access_count, 3);
    assert_eq!(miss_count, 0);

    event_feed_wrap.replace(None);
    suite.stop();
}

#[test]
fn test_old_value_cache_hit_pessimistic() {
    with_old_value_metrics_flush_disabled(|| {
        test_kv_format_impl!(test_old_value_cache_hit_pessimistic_impl<ApiV2>);
    });
}

fn test_old_value_cache_hit_pessimistic_impl<F: KvFormat>() {
    let mut suite = TestSuite::new(1, F::TAG);
    let scheduler = suite.endpoints.values().next().unwrap().clone();
    let region = get_ready_region(&mut suite, b"xkey0");
    let region_id = region.id;
    let mut req = suite.new_changedata_request(region_id);
    req.set_extra_op(ExtraOp::ReadOldValue);
    let (mut req_tx, event_feed_wrap, receive_event) =
        new_event_feed(suite.get_region_cdc_client(region_id));
    subscribe_until_initialized(
        &mut req_tx,
        receive_event.as_ref(),
        &req,
        "test_old_value_cache_hit_pessimistic initial subscribe",
    );
    let (tx, rx) = mpsc::channel();

    // Insert a value in pessimistic txn.
    let mut m3 = Mutation::default();
    let k3 = b"xkeyxk3".to_vec();
    m3.set_op(Op::PessimisticLock);
    m3.key = k3.clone();
    suite.must_acquire_pessimistic_lock(
        region_id,
        vec![m3.clone()],
        k3.clone(),
        10.into(),
        10.into(),
    );
    // CDC does not outputs PessimisticLock.
    // No cache access.
    let tx_ = tx.clone();
    scheduler
        .schedule(Task::Validate(Validate::OldValueCache(Box::new(
            move |old_value_cache| {
                tx_.send((old_value_cache.access_count(), old_value_cache.miss_count()))
                    .unwrap();
            },
        ))))
        .unwrap();
    let (access_count, miss_count) = rx.recv().unwrap();
    assert_eq!(access_count, 0);
    assert_eq!(miss_count, 0);
    m3.set_op(Op::Put);
    m3.value = b"v1".to_vec();
    suite.must_kv_pessimistic_prewrite(region_id, vec![m3], k3.clone(), 10.into(), 10.into());
    #[cfg(NO_NEXT_GEN_COMPATIBLE)]
    {
        // Current implementation emits an empty entry when encountering a `pessimistic`
        // lock...
        let _events = receive_event(false).events.to_vec();
        let mut events = receive_event(false).events.to_vec();
        match events.remove(0).event.unwrap() {
            Event_oneof_event::Entries(mut es) => {
                let row = &es.take_entries().to_vec()[0];
                assert_eq!(row.get_value(), b"v1");
                assert_eq!(row.get_old_value(), b"");
                assert_eq!(row.get_type(), EventLogType::Prewrite);
                assert_eq!(row.get_start_ts(), 10);
            }
            other => panic!("unknown event {:?}", other),
        }
    }

    suite.must_kv_commit(region_id, vec![k3], 10.into(), 15.into());
    let mut events = receive_event(false).events.to_vec();
    match events.remove(0).event.unwrap() {
        Event_oneof_event::Entries(mut es) => {
            let row = &es.take_entries().to_vec()[0];
            assert_eq!(row.get_type(), EventLogType::Committed);
            assert_eq!(row.get_commit_ts(), 15);
        }
        other => panic!("unknown event {:?}", other),
    }
    // k3 old value must be cached.
    let tx_ = tx.clone();
    scheduler
        .schedule(Task::Validate(Validate::OldValueCache(Box::new(
            move |old_value_cache| {
                tx_.send((old_value_cache.access_count(), old_value_cache.miss_count()))
                    .unwrap();
            },
        ))))
        .unwrap();
    let (access_count, miss_count) = rx.recv().unwrap();
    assert_eq!(access_count, 1);
    assert_eq!(miss_count, 0);

    // Update a value in pessimistic txn.
    let mut m3 = Mutation::default();
    let k3 = b"xkeyxk3".to_vec();
    m3.set_op(Op::PessimisticLock);
    m3.key = k3.clone();
    suite.must_acquire_pessimistic_lock(
        region_id,
        vec![m3.clone()],
        k3.clone(),
        20.into(),
        20.into(),
    );
    // CDC does not outputs PessimisticLock.
    // No cache access.
    let tx_ = tx.clone();
    scheduler
        .schedule(Task::Validate(Validate::OldValueCache(Box::new(
            move |old_value_cache| {
                tx_.send((old_value_cache.access_count(), old_value_cache.miss_count()))
                    .unwrap();
            },
        ))))
        .unwrap();
    let (access_count, miss_count) = rx.recv().unwrap();
    assert_eq!(access_count, 1);
    assert_eq!(miss_count, 0);
    m3.set_op(Op::Put);
    m3.value = b"v2".to_vec();
    suite.must_kv_pessimistic_prewrite(region_id, vec![m3], k3.clone(), 20.into(), 20.into());
    #[cfg(NO_NEXT_GEN_COMPATIBLE)]
    {
        // Current implementation emits an empty entry when encountering a `pessimistic`
        // lock...
        let _events = receive_event(false).events.to_vec();
        let mut events = receive_event(false).events.to_vec();
        match events.remove(0).event.unwrap() {
            Event_oneof_event::Entries(mut es) => {
                let row = &es.take_entries().to_vec()[0];
                assert_eq!(row.get_value(), b"v2");
                assert_eq!(row.get_old_value(), b"v1");
                assert_eq!(row.get_type(), EventLogType::Prewrite);
                assert_eq!(row.get_start_ts(), 20);
            }
            other => panic!("unknown event {:?}", other),
        }
    }

    suite.must_kv_commit(region_id, vec![k3], 20.into(), 25.into());
    let mut events = receive_event(false).events.to_vec();
    match events.remove(0).event.unwrap() {
        Event_oneof_event::Entries(mut es) => {
            let row = &es.take_entries().to_vec()[0];
            assert_eq!(row.get_type(), EventLogType::Committed);
            assert_eq!(row.get_commit_ts(), 25);
        }
        other => panic!("unknown event {:?}", other),
    }

    // k3 old value must be cached.
    let tx_ = tx;
    scheduler
        .schedule(Task::Validate(Validate::OldValueCache(Box::new(
            move |old_value_cache| {
                tx_.send((old_value_cache.access_count(), old_value_cache.miss_count()))
                    .unwrap();
            },
        ))))
        .unwrap();
    let (access_count, miss_count) = rx.recv().unwrap();
    assert_eq!(access_count, 2);
    assert_eq!(miss_count, 0);

    event_feed_wrap.replace(None);
    suite.stop();
}