// Copyright 2021 TiKV Project Authors. Licensed under Apache-2.0.
// Modifications Copyright 2022 PingCAP, Inc.

use crate::store::Config;

pub(crate) struct TickSchedule {
    run_at: u64,
    interval: u64,
}

impl TickSchedule {
    fn new(interval: u64) -> Self {
        Self {
            run_at: 0,
            interval,
        }
    }
}

#[inline]
fn interval_to_tick_count(interval: u64, base_interval: u64) -> u64 {
    if interval == 0 {
        return 0;
    }
    if base_interval == 0 {
        return 1;
    }
    interval.div_ceil(base_interval)
}

pub(crate) struct Ticker {
    tick: u64,
    pub(crate) tick_flag: u64,
    schedules: Vec<TickSchedule>,
}

impl Ticker {
    pub(crate) fn new(config: &Config) -> Self {
        let base_interval = config.raft_base_tick_interval.as_millis();
        let schedules = vec![
            TickSchedule::new(1),
            TickSchedule::new(interval_to_tick_count(
                config.split_region_check_tick_interval.as_millis(),
                base_interval,
            )),
            TickSchedule::new(interval_to_tick_count(
                config.pd_heartbeat_tick_interval.as_millis(),
                base_interval,
            )),
            TickSchedule::new(interval_to_tick_count(
                config.maintenance_tick_interval.as_millis(),
                base_interval,
            )),
            TickSchedule::new(interval_to_tick_count(
                config.raft_log_gc_tick_interval.as_millis(),
                base_interval,
            )),
            TickSchedule::new(interval_to_tick_count(
                config.peer_long_check_interval.as_millis(),
                base_interval,
            )),
            TickSchedule::new(interval_to_tick_count(
                config.reactive_memory_lock_tick_interval.as_millis(),
                base_interval,
            )),
            TickSchedule::new(interval_to_tick_count(
                config.file_gc_compaction_check_tick_interval.as_millis(),
                base_interval,
            )),
        ];
        Self {
            tick: 1,
            tick_flag: 0,
            schedules,
        }
    }

    pub(crate) fn update_peer_ticker(&mut self, config: &Config, region_id: u64) {
        let base_interval = config.raft_base_tick_interval.as_millis();
        self.update_peer_ticker_interval(
            PEER_TICK_SPLIT_CHECK,
            interval_to_tick_count(
                config.split_region_check_tick_interval.as_millis(),
                base_interval,
            ),
        );
        self.update_peer_ticker_interval_staggered(
            PEER_TICK_PD_HEARTBEAT,
            interval_to_tick_count(config.pd_heartbeat_tick_interval.as_millis(), base_interval),
            region_id,
        );
        self.update_peer_ticker_interval_staggered(
            PEER_TICK_MAINTENANCE,
            interval_to_tick_count(config.maintenance_tick_interval.as_millis(), base_interval),
            region_id,
        );
        self.update_peer_ticker_interval(
            PEER_TICK_RAFT_LOG_GC,
            interval_to_tick_count(config.raft_log_gc_tick_interval.as_millis(), base_interval),
        );
        self.update_peer_ticker_interval_staggered(
            PEER_TICK_CHECK_LONG,
            interval_to_tick_count(config.peer_long_check_interval.as_millis(), base_interval),
            region_id,
        );
        self.update_peer_ticker_interval(
            PEER_TICK_REACTIVATE_MEMORY_LOCK,
            interval_to_tick_count(
                config.reactive_memory_lock_tick_interval.as_millis(),
                base_interval,
            ),
        );
        self.update_peer_ticker_interval(
            PEER_TICK_FILE_GC_COMPACTION_CHECK,
            interval_to_tick_count(
                config.file_gc_compaction_check_tick_interval.as_millis(),
                base_interval,
            ),
        );
    }

    pub(crate) fn new_store(config: &Config) -> Self {
        let base_interval = config.raft_base_tick_interval.as_millis();
        let schedules = vec![
            TickSchedule::new(interval_to_tick_count(
                config.pd_store_heartbeat_tick_interval.as_millis(),
                base_interval,
            )),
            TickSchedule::new(interval_to_tick_count(
                config.update_gc_safe_point_interval.as_millis(),
                base_interval,
            )),
            TickSchedule::new(interval_to_tick_count(
                config.local_file_gc_tick_interval.as_millis(),
                base_interval,
            )),
        ];
        Self {
            tick: 0,
            tick_flag: 0,
            schedules,
        }
    }

    pub(crate) fn update_store_ticker(&mut self, config: &Config) {
        let base_interval = config.raft_base_tick_interval.as_millis();
        self.update_store_ticker_interval(
            STORE_TICK_PD_HEARTBEAT,
            interval_to_tick_count(
                config.pd_store_heartbeat_tick_interval.as_millis(),
                base_interval,
            ),
        );
        self.update_store_ticker_interval(
            STORE_TICK_UPDATE_GC_SAFE_POINT,
            interval_to_tick_count(
                config.update_gc_safe_point_interval.as_millis(),
                base_interval,
            ),
        );
        self.update_store_ticker_interval(
            STORE_TICK_LOCAL_FILE_GC,
            interval_to_tick_count(
                config.local_file_gc_tick_interval.as_millis(),
                base_interval,
            ),
        );
    }

    pub(crate) fn tick_clock(&mut self) {
        self.tick += 1;
        self.tick_flag = 0;
    }

    pub(crate) fn schedule(&mut self, tick: PeerTick) {
        let sched = &mut self.schedules[tick.idx];
        if sched.interval == 0 {
            sched.run_at = 0;
            return;
        }
        self.tick_flag |= 1 << tick.idx;
        sched.run_at = self.tick + sched.interval;
    }

    pub(crate) fn reschedule_elapsed_peer_ticks(&mut self) {
        for sched in &mut self.schedules {
            if sched.interval != 0 && sched.run_at != 0 && sched.run_at <= self.tick {
                sched.run_at = self.tick + sched.interval;
            }
        }
    }

    /// Schedules the first run at a deterministic offset within the interval.
    /// Peers are spread by region ID while later runs keep the regular
    /// interval.
    pub(crate) fn schedule_staggered(&mut self, tick: PeerTick, region_id: u64) {
        let sched = &mut self.schedules[tick.idx];
        if sched.interval == 0 {
            sched.run_at = 0;
            return;
        }
        self.tick_flag |= 1 << tick.idx;
        sched.run_at = self.tick + staggered_delay(region_id, sched.interval);
    }

    fn is_on_schedule(&self, idx: usize) -> bool {
        let sched = &self.schedules[idx];
        sched.run_at == self.tick
    }

    pub(crate) fn is_on_tick(&self, tick: PeerTick) -> bool {
        self.is_on_schedule(tick.idx)
    }

    pub(crate) fn schedule_store(&mut self, tick: StoreTick) {
        let sched = &mut self.schedules[tick.idx];
        if sched.interval == 0 {
            sched.run_at = 0;
            return;
        }
        sched.run_at = self.tick + sched.interval;
    }

    pub(crate) fn is_on_store_tick(&self, tick: StoreTick) -> bool {
        self.is_on_schedule(tick.idx)
    }

    #[inline]
    pub(crate) fn update_peer_ticker_interval(&mut self, tick: PeerTick, interval: u64) {
        self.update_ticker_interval(tick.idx, interval);
    }

    fn update_peer_ticker_interval_staggered(
        &mut self,
        tick: PeerTick,
        interval: u64,
        region_id: u64,
    ) {
        let sched = &mut self.schedules[tick.idx];
        if sched.interval == interval {
            return;
        }
        let existing_run_at = sched.run_at;
        sched.interval = interval;
        if interval == 0 {
            sched.run_at = 0;
            return;
        }
        let staggered_run_at = self.tick + staggered_delay(region_id, interval);
        sched.run_at = if existing_run_at > self.tick {
            std::cmp::min(existing_run_at, staggered_run_at)
        } else {
            staggered_run_at
        };
    }

    #[inline]
    pub(crate) fn update_store_ticker_interval(&mut self, tick: StoreTick, interval: u64) {
        self.update_ticker_interval(tick.idx, interval);
    }

    fn update_ticker_interval(&mut self, idx: usize, interval: u64) {
        let sched = &mut self.schedules[idx];
        if sched.interval == interval {
            return;
        }
        sched.interval = interval;
        if interval == 0 {
            sched.run_at = 0;
            return;
        }
        // reset next run_at if the interval decreases.
        let new_run_at = if sched.run_at <= self.tick {
            self.tick + interval
        } else {
            std::cmp::min(sched.run_at, self.tick + interval)
        };

        sched.run_at = new_run_at;
    }
}

#[inline]
fn staggered_delay(region_id: u64, interval: u64) -> u64 {
    debug_assert!(interval > 0);
    u64::from(crc32c::crc32c(&region_id.to_le_bytes())) % interval + 1
}

#[derive(Copy, Clone, PartialEq, Eq)]
pub(crate) struct PeerTick {
    idx: usize,
}

pub(crate) const PEER_TICK_RAFT: PeerTick = PeerTick { idx: 0 };
pub(crate) const PEER_TICK_SPLIT_CHECK: PeerTick = PeerTick { idx: 1 };
pub(crate) const PEER_TICK_PD_HEARTBEAT: PeerTick = PeerTick { idx: 2 };
pub(crate) const PEER_TICK_MAINTENANCE: PeerTick = PeerTick { idx: 3 };
pub(crate) const PEER_TICK_RAFT_LOG_GC: PeerTick = PeerTick { idx: 4 };
pub(crate) const PEER_TICK_CHECK_LONG: PeerTick = PeerTick { idx: 5 }; // Default is 5 minutes.
pub(crate) const PEER_TICK_REACTIVATE_MEMORY_LOCK: PeerTick = PeerTick { idx: 6 };
pub(crate) const PEER_TICK_FILE_GC_COMPACTION_CHECK: PeerTick = PeerTick { idx: 7 };

#[derive(Eq, PartialEq)]
pub struct StoreTick {
    idx: usize,
}

pub(crate) const STORE_TICK_PD_HEARTBEAT: StoreTick = StoreTick { idx: 0 };
pub(crate) const STORE_TICK_UPDATE_GC_SAFE_POINT: StoreTick = StoreTick { idx: 1 };
pub(crate) const STORE_TICK_LOCAL_FILE_GC: StoreTick = StoreTick { idx: 2 };

#[cfg(test)]
mod tests {
    use tikv_util::config::ReadableDuration;

    use super::*;

    #[test]
    fn test_peer_ticks_round_up_to_one_base_tick() {
        let mut cfg = Config::new();
        cfg.raft_base_tick_interval = ReadableDuration::secs(5);
        cfg.split_region_check_tick_interval = ReadableDuration::secs(1);
        cfg.pd_heartbeat_tick_interval = ReadableDuration::secs(2);
        cfg.maintenance_tick_interval = ReadableDuration::secs(3);
        cfg.raft_log_gc_tick_interval = ReadableDuration::secs(4);
        cfg.peer_long_check_interval = ReadableDuration::secs(1);

        let mut ticker = Ticker::new(&cfg);
        for tick in [
            PEER_TICK_RAFT,
            PEER_TICK_SPLIT_CHECK,
            PEER_TICK_PD_HEARTBEAT,
            PEER_TICK_MAINTENANCE,
            PEER_TICK_RAFT_LOG_GC,
            PEER_TICK_CHECK_LONG,
        ] {
            ticker.schedule(tick);
        }

        ticker.tick_clock();

        for tick in [
            PEER_TICK_RAFT,
            PEER_TICK_SPLIT_CHECK,
            PEER_TICK_PD_HEARTBEAT,
            PEER_TICK_MAINTENANCE,
            PEER_TICK_RAFT_LOG_GC,
            PEER_TICK_CHECK_LONG,
        ] {
            assert!(ticker.is_on_tick(tick));
        }
    }

    #[test]
    fn test_store_ticks_round_up_to_one_base_tick() {
        let mut cfg = Config::new();
        cfg.raft_base_tick_interval = ReadableDuration::secs(5);
        cfg.pd_store_heartbeat_tick_interval = ReadableDuration::secs(1);
        cfg.update_gc_safe_point_interval = ReadableDuration::secs(2);
        cfg.local_file_gc_tick_interval = ReadableDuration::secs(3);

        let mut ticker = Ticker::new_store(&cfg);
        for tick in [
            STORE_TICK_PD_HEARTBEAT,
            STORE_TICK_UPDATE_GC_SAFE_POINT,
            STORE_TICK_LOCAL_FILE_GC,
        ] {
            ticker.schedule_store(tick);
        }

        ticker.tick_clock();

        for tick in [
            STORE_TICK_PD_HEARTBEAT,
            STORE_TICK_UPDATE_GC_SAFE_POINT,
            STORE_TICK_LOCAL_FILE_GC,
        ] {
            assert!(ticker.is_on_store_tick(tick));
        }
    }

    #[test]
    fn test_zero_interval_stays_disabled() {
        let mut cfg = Config::new();
        cfg.update_gc_safe_point_interval = ReadableDuration::ZERO;

        let mut ticker = Ticker::new_store(&cfg);
        ticker.schedule_store(STORE_TICK_UPDATE_GC_SAFE_POINT);
        ticker.tick_clock();

        assert!(!ticker.is_on_store_tick(STORE_TICK_UPDATE_GC_SAFE_POINT));
    }

    #[test]
    fn test_peer_tick_updates_round_up_to_one_base_tick() {
        let mut cfg = Config::new();
        cfg.raft_base_tick_interval = ReadableDuration::secs(5);
        cfg.split_region_check_tick_interval = ReadableDuration::secs(10);

        let mut ticker = Ticker::new(&cfg);
        ticker.schedule(PEER_TICK_SPLIT_CHECK);

        cfg.split_region_check_tick_interval = ReadableDuration::secs(1);
        ticker.update_peer_ticker(&cfg, 1);
        ticker.tick_clock();

        assert!(ticker.is_on_tick(PEER_TICK_SPLIT_CHECK));
    }

    #[test]
    fn test_reschedule_elapsed_peer_ticks_rearms_missed_ticks() {
        let mut cfg = Config::new();
        cfg.raft_base_tick_interval = ReadableDuration::secs(1);
        cfg.split_region_check_tick_interval = ReadableDuration::secs(1);
        cfg.file_gc_compaction_check_tick_interval = ReadableDuration::ZERO;

        let mut ticker = Ticker::new(&cfg);
        ticker.schedule(PEER_TICK_RAFT);
        ticker.schedule(PEER_TICK_SPLIT_CHECK);
        ticker.schedule(PEER_TICK_FILE_GC_COMPACTION_CHECK);

        ticker.tick_clock();
        assert!(ticker.is_on_tick(PEER_TICK_RAFT));
        assert!(ticker.is_on_tick(PEER_TICK_SPLIT_CHECK));
        assert!(!ticker.is_on_tick(PEER_TICK_FILE_GC_COMPACTION_CHECK));

        ticker.reschedule_elapsed_peer_ticks();
        assert!(!ticker.is_on_tick(PEER_TICK_RAFT));
        assert!(!ticker.is_on_tick(PEER_TICK_SPLIT_CHECK));
        assert!(!ticker.is_on_tick(PEER_TICK_FILE_GC_COMPACTION_CHECK));

        ticker.tick_clock();
        assert!(ticker.is_on_tick(PEER_TICK_RAFT));
        assert!(ticker.is_on_tick(PEER_TICK_SPLIT_CHECK));
        assert!(!ticker.is_on_tick(PEER_TICK_FILE_GC_COMPACTION_CHECK));
    }

    #[test]
    fn test_staggered_peer_tick_uses_region_id_offset() {
        let mut cfg = Config::new();
        cfg.raft_base_tick_interval = ReadableDuration::secs(1);
        cfg.pd_heartbeat_tick_interval = ReadableDuration::secs(60);
        let region_id = 42;
        let mut ticker = Ticker::new(&cfg);

        ticker.schedule_staggered(PEER_TICK_PD_HEARTBEAT, region_id);

        let schedule = &ticker.schedules[PEER_TICK_PD_HEARTBEAT.idx];
        assert_eq!(
            schedule.run_at,
            ticker.tick + staggered_delay(region_id, schedule.interval)
        );
        assert!(schedule.run_at > ticker.tick);
        assert!(schedule.run_at <= ticker.tick + schedule.interval);
    }

    #[test]
    fn test_staggered_peer_tick_rephases_after_interval_change() {
        let mut cfg = Config::new();
        cfg.raft_base_tick_interval = ReadableDuration::secs(1);
        cfg.maintenance_tick_interval = ReadableDuration::secs(60);
        let region_id = 42;
        let mut ticker = Ticker::new(&cfg);
        ticker.schedule_staggered(PEER_TICK_MAINTENANCE, region_id);
        let previous_run_at = ticker.schedules[PEER_TICK_MAINTENANCE.idx].run_at;

        cfg.maintenance_tick_interval = ReadableDuration::secs(30);
        ticker.update_peer_ticker(&cfg, region_id);

        let schedule = &ticker.schedules[PEER_TICK_MAINTENANCE.idx];
        assert_eq!(schedule.interval, 30);
        assert_eq!(
            schedule.run_at,
            std::cmp::min(
                previous_run_at,
                ticker.tick + staggered_delay(region_id, schedule.interval)
            )
        );
    }

    #[test]
    fn test_staggered_peer_tick_update_does_not_postpone_existing_deadline() {
        let mut cfg = Config::new();
        cfg.raft_base_tick_interval = ReadableDuration::secs(1);
        cfg.pd_heartbeat_tick_interval = ReadableDuration::secs(60);
        let region_id = 42;
        let mut ticker = Ticker::new(&cfg);
        ticker.schedule_staggered(PEER_TICK_PD_HEARTBEAT, region_id);
        let existing_run_at = ticker.tick + 1;
        ticker.schedules[PEER_TICK_PD_HEARTBEAT.idx].run_at = existing_run_at;

        cfg.pd_heartbeat_tick_interval = ReadableDuration::secs(120);
        ticker.update_peer_ticker(&cfg, region_id);

        assert_eq!(
            ticker.schedules[PEER_TICK_PD_HEARTBEAT.idx].run_at,
            existing_run_at
        );
    }
}