// Copyright 2025 PingCAP, Inc.
//! ServiceSafepointManager manages service safepoints for changefeeds.
//!
//! Ref: https://github.com/pingcap/tiflow/blob/release-8.5/pkg/txnutil/gc/doc.go

use std::{fmt, sync::Arc, time::Duration};

use collections::{HashMap, HashMapEntry};
use pd_client::PdClient;
use security::HttpClient;
use tikv_util::{
    box_try, box_try_join, debug, error, info, warn,
    worker::{Runnable, RunnableWithTimer, Scheduler, Worker},
};
use txn_types::TimeStamp;

use crate::{
    Error, KeyspaceStates, ReplicationWorkerConfig, Result, SafepointConfig,
    metrics::{REP_KEYSPACE_SERVICE_SAFEPOINT, REP_SAFEPOINT_EVENTS_COUNTER},
    ticdc_util::{ReplicationTaskItem, list_changefeeds_from_ticdc},
};

const GC_WORKER_SERVICE_SAFEPOINT_ID: &str = "gc_worker";

pub(crate) struct ServiceSafepointManager {
    config: SafepointConfig,
    merged_store_id: u64,

    pd: Arc<dyn PdClient>,
    runtime: tokio::runtime::Handle,

    worker: Worker,
    scheduler: Scheduler<ServiceSafepointTask>,
}

impl ServiceSafepointManager {
    pub(crate) fn new(
        merged_store_id: u64,
        pd: Arc<dyn PdClient>,
        rep_config: &ReplicationWorkerConfig,
        runtime: tokio::runtime::Handle,
        keyspaces: HashMap<u32 /* keyspace_id */, KeyspaceContext>,
    ) -> Result<Self> {
        info!("ServiceSafepointManager start";
            "store" => merged_store_id, "keyspaces" => ?keyspaces);

        let http_client = box_try!(pd.get_security_mgr().http_client(hyper::Client::builder()));
        let config = rep_config.safepoint.clone();
        let runner = ServiceSafepointRunner {
            config: config.clone(),
            merged_store_id,
            pd: pd.clone(),
            runtime: runtime.clone(),
            http_client,
            keyspaces,
        };

        let worker = Worker::new("ServiceSafepointWorker");
        let scheduler = worker.start_with_timer("ServiceSafepointRunner", runner);
        if let Err(err) = scheduler.schedule(ServiceSafepointTask::SyncAllKeyspaces) {
            warn!(
                "ServiceSafepointManager: initial sync schedule failed: {:?}",
                err
            );
        }
        Ok(Self {
            config,
            merged_store_id,
            pd,
            runtime,
            worker,
            scheduler,
        })
    }

    pub(crate) fn shutdown(self) {
        self.scheduler.stop();
        self.worker.stop();
    }

    pub(crate) fn ensure_changefeed_start_ts_safety(
        &mut self,
        keyspace_id: u32,
        feed_id: &str,
        start_ts: u64,
    ) -> Result<()> {
        let keyspace_str = keyspace_id.to_string();
        let service_id = self.get_ensure_start_ts_service_id(feed_id);
        match self
            .runtime
            .block_on(self.pd.update_keyspace_service_safe_point(
                keyspace_id,
                service_id,
                start_ts.into(),
                self.config.create_changefeed_gc_ttl.0,
            )) {
            Ok(new_safepoint) => {
                info!("ServiceSafepointManager: ensure_changefeed_start_ts_safety ok";
                    "keyspace" => keyspace_id, "feed" => feed_id,
                    "start_ts" => start_ts, "new_safepoint" => new_safepoint);
                REP_SAFEPOINT_EVENTS_COUNTER
                    .with_label_values(&["ok", &keyspace_str, "ensure_start_ts_safepoint"])
                    .inc();
                Ok(())
            }
            Err(pd_client::Error::UnsafeServiceGcSafePoint {
                requested,
                current_minimal,
            }) => {
                debug_assert_eq!(requested.into_inner(), start_ts);
                warn!("ServiceSafepointManager: start_ts before safepoint";
                    "keyspace" => keyspace_id, "feed" => feed_id,
                    "start_ts" => start_ts, "current" => current_minimal);
                REP_SAFEPOINT_EVENTS_COUNTER
                    .with_label_values(&["warn", &keyspace_str, "start_ts_before_safepoint"])
                    .inc();
                Err(Error::StartTsBeforeSafepoint {
                    start_ts,
                    gc_safe_point: current_minimal.into_inner(),
                })
            }
            Err(err) => {
                error!("ServiceSafepointManager: ensure_changefeed_start_ts_safety failed: {:?}", err;
                    "keyspace" => keyspace_id, "feed" => feed_id, "start_ts" => start_ts);
                REP_SAFEPOINT_EVENTS_COUNTER
                    .with_label_values(&["error", &keyspace_str, "ensure_start_ts_safepoint"])
                    .inc();
                Err(err.into())
            }
        }
    }

    pub(crate) fn add_keyspace(
        &self,
        keyspace_id: u32,
        cdc_addr: String,
        rep_pd_cli: Arc<dyn PdClient>,
    ) {
        let task = ServiceSafepointTask::AddKeyspace {
            keyspace_id,
            cdc_addr,
            rep_pd_cli,
        };
        if let Err(err) = self.scheduler.schedule(task) {
            warn!("ServiceSafepointManager: add_keyspace schedule failed: {:?}", err;
                "keyspace" => keyspace_id);
            return;
        }
        if let Err(err) = self
            .scheduler
            .schedule(ServiceSafepointTask::SyncKeyspace { keyspace_id })
        {
            warn!("ServiceSafepointManager: add_keyspace sync schedule failed: {:?}", err;
                "keyspace" => keyspace_id);
        }
    }

    pub(crate) fn remove_keyspace(&self, keyspace_id: u32) {
        let task = ServiceSafepointTask::RemoveKeyspace { keyspace_id };
        if let Err(err) = self.scheduler.schedule(task) {
            warn!("ServiceSafepointManager: remove_keyspace schedule failed: {:?}", err;
                "keyspace" => keyspace_id);
        }
    }

    pub(crate) fn sync_keyspace(&self, keyspace_id: u32) {
        let task = ServiceSafepointTask::SyncKeyspace { keyspace_id };
        if let Err(err) = self.scheduler.schedule(task) {
            warn!("ServiceSafepointManager: sync_keyspace schedule failed: {:?}", err;
                "keyspace" => keyspace_id);
        }
    }

    fn get_ensure_start_ts_service_id(&self, changefeed_id: &str) -> String {
        format!(
            "rep-worker-{}-creating-{}",
            self.merged_store_id, changefeed_id
        )
    }
}

#[derive(Clone)]
pub(crate) struct KeyspaceContext {
    keyspace_id: u32,
    cdc_addr: String,
    rep_pd_cli: Arc<dyn PdClient>,

    last_rep_gc_safepoint: Option<u64>,
    keyspace_safepoint_removed: bool,
}

impl fmt::Debug for KeyspaceContext {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("KeyspaceContext")
            .field("keyspace_id", &self.keyspace_id)
            .field("cdc_addr", &self.cdc_addr)
            .field("last_rep_gc_safepoint", &self.last_rep_gc_safepoint)
            .field(
                "keyspace_safepoint_removed",
                &self.keyspace_safepoint_removed,
            )
            .finish()
    }
}

impl KeyspaceContext {
    pub(crate) fn new(
        keyspace_id: u32,
        states: &KeyspaceStates,
        rep_pd_cli: Arc<dyn PdClient>,
    ) -> Self {
        Self {
            keyspace_id,
            cdc_addr: states.cdc_addr.clone(),
            rep_pd_cli,
            last_rep_gc_safepoint: None,
            keyspace_safepoint_removed: false,
        }
    }

    fn sync(&mut self, updated_ctx: Self) {
        self.last_rep_gc_safepoint = updated_ctx.last_rep_gc_safepoint;
        self.keyspace_safepoint_removed = updated_ctx.keyspace_safepoint_removed
    }
}

struct ServiceSafepointRunner {
    config: SafepointConfig,

    merged_store_id: u64,
    pd: Arc<dyn PdClient>,
    runtime: tokio::runtime::Handle,
    http_client: HttpClient,

    keyspaces: HashMap<u32 /* keyspace_id */, KeyspaceContext>,
}

impl ServiceSafepointRunner {
    #[inline]
    fn mut_keyspace(&mut self, keyspace_id: u32) -> Option<&mut KeyspaceContext> {
        self.keyspaces.get_mut(&keyspace_id)
    }

    fn remove_keyspace(&mut self, keyspace_id: u32) {
        if self.keyspaces.remove(&keyspace_id).is_none() {
            debug_assert!(false);
            return;
        };
        if let Err(err) = self.remove_keyspace_safepoint(keyspace_id) {
            warn!("ServiceSafepointManager: remove_keyspace_safepoint failed: {:?}", err;
                "keyspace" => keyspace_id);
        }
    }

    fn remove_keyspace_safepoint(&self, keyspace_id: u32) -> Result<()> {
        self.runtime.block_on(Self::remove_keyspace_safepoint_async(
            self.pd.as_ref(),
            keyspace_id,
            self.get_gc_service_id(),
        ))
    }

    async fn remove_keyspace_safepoint_async(
        pd: &dyn PdClient,
        keyspace_id: u32,
        gc_service_id: String,
    ) -> Result<()> {
        let keyspace_str = keyspace_id.to_string();
        match pd
            .remove_keyspace_service_safe_point(keyspace_id, gc_service_id)
            .await
        {
            Ok(new_safepoint) => {
                info!("ServiceSafepointManager: remove_keyspace_safepoint ok";
                    "keyspace" => keyspace_id, "new_safepoint" => new_safepoint);
                REP_SAFEPOINT_EVENTS_COUNTER
                    .with_label_values(&["ok", &keyspace_str, "remove_keyspace_safepoint"])
                    .inc();
                let _ = REP_KEYSPACE_SERVICE_SAFEPOINT
                    .remove_label_values(&[&keyspace_str, "safepoint"]);
                let _ = REP_KEYSPACE_SERVICE_SAFEPOINT
                    .remove_label_values(&[&keyspace_str, "min_checkpoint_ts"]);
                Ok(())
            }
            Err(err) => {
                warn!("ServiceSafepointManager: remove_keyspace_safepoint failed: {:?}", err; "keyspace" => keyspace_id);
                REP_SAFEPOINT_EVENTS_COUNTER
                    .with_label_values(&["warn", &keyspace_str, "remove_keyspace_safepoint"])
                    .inc();
                Err(err.into())
            }
        }
    }

    async fn update_keyspace_safepoint(
        pd: &dyn PdClient,
        keyspace_id: u32,
        gc_service_id: String,
        min_checkpoint_ts: u64,
        gc_ttl: Duration,
    ) {
        let keyspace_str = keyspace_id.to_string();
        match pd
            .update_keyspace_service_safe_point(
                keyspace_id,
                gc_service_id,
                min_checkpoint_ts.into(),
                gc_ttl,
            )
            .await
        {
            Ok(new_safepoint) => {
                info!("ServiceSafepointManager: update keyspace safepoint ok";
                    "keyspace" => keyspace_id, "min_checkpoint_ts" => min_checkpoint_ts, "new" => new_safepoint);
                REP_SAFEPOINT_EVENTS_COUNTER
                    .with_label_values(&["ok", &keyspace_str, "update_keyspace_safepoint"])
                    .inc();
                REP_KEYSPACE_SERVICE_SAFEPOINT
                    .with_label_values(&[&keyspace_str, "safepoint"])
                    .set(new_safepoint as i64);
                REP_KEYSPACE_SERVICE_SAFEPOINT
                    .with_label_values(&[&keyspace_str, "min_checkpoint_ts"])
                    .set(min_checkpoint_ts as i64);
            }
            Err(pd_client::Error::UnsafeServiceGcSafePoint {
                requested,
                current_minimal,
            }) => {
                // TODO: Stop the replication ?
                debug_assert_eq!(requested.into_inner(), min_checkpoint_ts);
                error!("ServiceSafepointManager: unsafe service safepoint";
                    "keyspace" => keyspace_id, "min_checkpoint_ts" => min_checkpoint_ts, "current" => current_minimal);
                REP_SAFEPOINT_EVENTS_COUNTER
                    .with_label_values(&["fatal", &keyspace_str, "unsafe_safepoint"])
                    .inc();
            }
            Err(err) => {
                warn!("ServiceSafepointManager: update_keyspace_safepoint failed: {:?}", err;
                    "keyspace" => keyspace_id, "min_checkpoint_ts" => min_checkpoint_ts);
                REP_SAFEPOINT_EVENTS_COUNTER
                    .with_label_values(&["warn", &keyspace_str, "update_keyspace_safepoint"])
                    .inc();
            }
        }
    }

    fn sync_ticdc(&mut self) {
        let runtime = self.runtime.clone();
        match runtime.block_on(self.sync_all_changefeeds_from_ticdc()) {
            Ok(()) => {
                debug!("ServiceSafepointManager: sync ticdc ok");
                REP_SAFEPOINT_EVENTS_COUNTER
                    .with_label_values(&["ok", "", "sync_ticdc"])
                    .inc();
            }
            Err(err) => {
                warn!("ServiceSafepointManager: sync ticdc failed: {:?}", err);
                REP_SAFEPOINT_EVENTS_COUNTER
                    .with_label_values(&["error", "", "sync_ticdc"])
                    .inc();
            }
        }
    }

    fn sync_keyspace_ticdc(&mut self, keyspace_id: u32) {
        let runtime = self.runtime.clone();
        match runtime.block_on(self.sync_one_keyspace_from_ticdc(keyspace_id)) {
            Ok(()) => {
                debug!("ServiceSafepointManager: sync keyspace ticdc ok"; "keyspace" => keyspace_id);
                REP_SAFEPOINT_EVENTS_COUNTER
                    .with_label_values(&["ok", &keyspace_id.to_string(), "sync_keyspace_ticdc"])
                    .inc();
            }
            Err(err) => {
                warn!("ServiceSafepointManager: sync keyspace ticdc failed: {:?}", err; "keyspace" => keyspace_id);
                REP_SAFEPOINT_EVENTS_COUNTER
                    .with_label_values(&["error", &keyspace_id.to_string(), "sync_keyspace_ticdc"])
                    .inc();
            }
        }
    }

    async fn sync_one_keyspace_from_ticdc(&mut self, keyspace_id: u32) -> Result<()> {
        let Some(ks) = self.keyspaces.get(&keyspace_id) else {
            return Ok(());
        };
        let updated_ctx = Self::sync_changefeeds_from_ticdc(
            self.pd.as_ref(),
            ks.clone(),
            self.get_gc_service_id(),
            self.config.gc_ttl.0,
            &self.http_client,
            self.config.sync_ticdc_timeout.0,
        )
        .await?;
        if let Some(ks) = self.mut_keyspace(keyspace_id) {
            ks.sync(updated_ctx);
        }
        Ok(())
    }

    async fn sync_all_changefeeds_from_ticdc(&mut self) -> Result<()> {
        let pd = self.pd.clone();
        let http_client = self.http_client.clone();
        let gc_service_id = self.get_gc_service_id();
        let gc_ttl = self.config.gc_ttl.0;
        let timeout = self.config.sync_ticdc_timeout.0;

        let mut join_set = tokio::task::JoinSet::new();
        let ks_ctxs = self.keyspaces.values().cloned();
        for ks_ctx in ks_ctxs {
            let keyspace_id = ks_ctx.keyspace_id;
            let pd = pd.clone();
            let http_client = http_client.clone();
            let gc_service_id = gc_service_id.clone();
            let sub_task = async move {
                let sync_res = Self::sync_changefeeds_from_ticdc(
                    pd.as_ref(),
                    ks_ctx,
                    gc_service_id,
                    gc_ttl,
                    &http_client,
                    timeout,
                )
                .await;
                (keyspace_id, sync_res)
            };
            join_set.spawn(sub_task);
        }

        while let Some(res) = join_set.join_next().await {
            let (keyspace_id, sync_res) = box_try_join!(res);
            match sync_res {
                Ok(updated_ctx) => {
                    let Some(ks) = self.mut_keyspace(keyspace_id) else {
                        continue;
                    };
                    ks.sync(updated_ctx);
                }
                Err(err) => {
                    warn!("ServiceSafepointManager: sync_changefeeds_from_ticdc failed: {:?}", err; "keyspace" => keyspace_id);
                }
            }
        }
        Ok(())
    }

    async fn sync_changefeeds_from_ticdc(
        pd: &dyn PdClient,
        mut ctx: KeyspaceContext,
        gc_service_id: String,
        gc_ttl: Duration,
        http_client: &HttpClient,
        timeout: Duration,
    ) -> Result<KeyspaceContext> {
        let keyspace_str = ctx.keyspace_id.to_string();
        match list_changefeeds_from_ticdc(
            ctx.keyspace_id,
            &ctx.cdc_addr,
            http_client,
            timeout,
            "get_changefeeds_from_ticdc",
        )
        .await
        {
            Ok(feeds_list) => {
                let now = TimeStamp::now();
                let gc_view = ChangefeedGcView::new(feeds_list.items, now, gc_ttl);

                let min_checkpoint_ts_opt = gc_view.min_synced_checkpoint_ts();
                if let Some(min_checkpoint_ts) = min_checkpoint_ts_opt {
                    info!("ServiceSafepointManager: sync_changefeeds";
                        "min_checkpoint_ts" => min_checkpoint_ts,
                        "feeds" => ?gc_view.synced_feeds,
                        "keyspace" => ctx.keyspace_id);
                    // Update the keyspace even if safepoint not changed, to extend the TTL.
                    Self::update_keyspace_safepoint(
                        pd,
                        ctx.keyspace_id,
                        gc_service_id.clone(),
                        min_checkpoint_ts,
                        gc_ttl,
                    )
                    .await;
                    ctx.keyspace_safepoint_removed = false;
                }

                if !gc_view.expired_feeds.is_empty() {
                    warn!("ServiceSafepointManager: expired changefeeds: {:?}", gc_view.expired_feeds; "keyspace" => ctx.keyspace_id);
                    REP_SAFEPOINT_EVENTS_COUNTER
                        .with_label_values(&["warn", &keyspace_str, "changefeed_expired"])
                        .inc_by(gc_view.expired_feeds.len() as u64);

                    // Set the rep_pd GC safepoint to make TiCDC consider the GC is blocked by
                    // expired changefeeds, then set the changefeeds failed, and cannot resume.
                    // See `isTiCDCBlockGC` in TiCDC code.
                    let max_expired_checkpoint_ts = gc_view.max_expired_checkpoint_ts().unwrap();
                    if ctx
                        .last_rep_gc_safepoint
                        .is_none_or(|sp| sp < max_expired_checkpoint_ts)
                    {
                        let current_gc_safepoint_opt = Self::update_rep_pd_gc_safepoint(
                            ctx.keyspace_id,
                            ctx.rep_pd_cli.as_ref(),
                            max_expired_checkpoint_ts,
                        )
                        .await;
                        if let Some(current) = current_gc_safepoint_opt {
                            ctx.last_rep_gc_safepoint = Some(current);
                        }
                    }
                }

                #[allow(clippy::collapsible_if)]
                if gc_view.synced_feeds.is_empty() && !ctx.keyspace_safepoint_removed {
                    if Self::remove_keyspace_safepoint_async(pd, ctx.keyspace_id, gc_service_id)
                        .await
                        .is_ok()
                    {
                        ctx.keyspace_safepoint_removed = true;
                    }
                }

                Ok(ctx)
            }
            Err(err) => {
                warn!("ServiceSafepointManager: get_changefeeds_from_ticdc failed: {:?}", err; "keyspace" => ctx.keyspace_id);
                REP_SAFEPOINT_EVENTS_COUNTER
                    .with_label_values(&["warn", &keyspace_str, "get_changefeeds_from_ticdc"])
                    .inc();
                Err(err)
            }
        }
    }

    // TiDB use the `update_service_safe_pint` API to set the GC safepoint, with
    // service name "gc_worker" and TTL MaxInt64.
    // See https://github.com/pingcap/tidb/blob/v8.5.3/pkg/store/gcworker/gc_worker.go#L719.
    async fn update_rep_pd_gc_safepoint(
        keyspace_id: u32,
        rep_pd_cli: &dyn PdClient,
        safepoint: u64,
    ) -> Option<u64 /* current_gc_safepoint */> {
        let keyspace_str = keyspace_id.to_string();
        match rep_pd_cli
            .update_service_safe_point(
                GC_WORKER_SERVICE_SAFEPOINT_ID.into(),
                safepoint.into(),
                Duration::from_secs(i64::MAX as u64),
            )
            .await
        {
            Ok(()) => {
                info!("ServiceSafepointManager: update_rep_pd_gc_safepoint ok";
                    "safepoint" => safepoint, "keyspace" => keyspace_id);
                REP_SAFEPOINT_EVENTS_COUNTER
                    .with_label_values(&["ok", &keyspace_str, "update_rep_gc_safepoint"])
                    .inc();
                Some(safepoint)
            }
            Err(pd_client::Error::UnsafeServiceGcSafePoint {
                requested,
                current_minimal,
            }) => {
                // Will happen when replication worker restart.
                warn!("ServiceSafepointManager: update_rep_pd_gc_safepoint stale gc safepoint";
                    "requested" => requested, "current" => current_minimal, "keyspace" => keyspace_id);
                REP_SAFEPOINT_EVENTS_COUNTER
                    .with_label_values(&["warn", &keyspace_str, "stale_rep_gc_safepoint"])
                    .inc();
                Some(current_minimal.into_inner())
            }
            Err(err) => {
                warn!("ServiceSafepointManager: update_rep_pd_gc_safepoint failed: {:?}", err;
                    "keyspace" => keyspace_id);
                REP_SAFEPOINT_EVENTS_COUNTER
                    .with_label_values(&["warn", &keyspace_str, "update_rep_gc_safepoint"])
                    .inc();
                None
            }
        }
    }

    fn get_gc_service_id(&self) -> String {
        format!("rep-worker-{}", self.merged_store_id)
    }
}

pub(crate) enum ServiceSafepointTask {
    AddKeyspace {
        keyspace_id: u32,
        cdc_addr: String,
        rep_pd_cli: Arc<dyn PdClient>,
    },
    RemoveKeyspace {
        keyspace_id: u32,
    },
    SyncAllKeyspaces,
    SyncKeyspace {
        keyspace_id: u32,
    },
}

impl fmt::Display for ServiceSafepointTask {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            ServiceSafepointTask::AddKeyspace { .. } => write!(f, "AddKeyspace"),
            ServiceSafepointTask::RemoveKeyspace { .. } => write!(f, "RemoveKeyspace"),
            ServiceSafepointTask::SyncAllKeyspaces => write!(f, "SyncAllKeyspaces"),
            ServiceSafepointTask::SyncKeyspace { .. } => write!(f, "SyncKeyspace"),
        }
    }
}

impl Runnable for ServiceSafepointRunner {
    type Task = ServiceSafepointTask;

    fn run(&mut self, task: Self::Task) {
        match task {
            ServiceSafepointTask::AddKeyspace {
                keyspace_id,
                cdc_addr,
                rep_pd_cli,
            } => {
                match self.keyspaces.entry(keyspace_id) {
                    HashMapEntry::Vacant(e) => {
                        e.insert(KeyspaceContext {
                            keyspace_id,
                            cdc_addr,
                            rep_pd_cli,
                            last_rep_gc_safepoint: None,
                            keyspace_safepoint_removed: false,
                        });
                    }
                    HashMapEntry::Occupied(mut e) => {
                        warn!(
                            "ServiceSafepointManager: add keyspace: keyspace exists: {}",
                            keyspace_id
                        );
                        debug_assert!(false);
                        // Overwrite ctx for safety.
                        e.get_mut().cdc_addr = cdc_addr;
                        e.get_mut().rep_pd_cli = rep_pd_cli;
                    }
                }
            }
            ServiceSafepointTask::RemoveKeyspace { keyspace_id } => {
                self.remove_keyspace(keyspace_id);
            }
            ServiceSafepointTask::SyncAllKeyspaces => self.sync_ticdc(),
            ServiceSafepointTask::SyncKeyspace { keyspace_id } => {
                self.sync_keyspace_ticdc(keyspace_id)
            }
        }
    }
}

impl RunnableWithTimer for ServiceSafepointRunner {
    fn on_timeout(&mut self) {
        self.sync_ticdc();
    }

    fn get_interval(&self) -> Duration {
        self.config.sync_safepoint_interval.0
    }
}

struct ChangefeedGcView {
    synced_feeds: Vec<ReplicationTaskItem>,
    expired_feeds: Vec<ReplicationTaskItem>,
}

impl ChangefeedGcView {
    fn new(feeds: Vec<ReplicationTaskItem>, now: TimeStamp, gc_ttl: Duration) -> Self {
        let mut synced_feeds = Vec::with_capacity(feeds.len());
        let mut expired_feeds = Vec::new();

        for feed in feeds {
            if now.duration_since(TimeStamp::new(feed.checkpoint_tso)) <= gc_ttl {
                synced_feeds.push(feed);
            } else {
                expired_feeds.push(feed);
            }
        }

        Self {
            synced_feeds,
            expired_feeds,
        }
    }

    fn min_synced_checkpoint_ts(&self) -> Option<u64> {
        self.synced_feeds
            .iter()
            .map(|feed| feed.checkpoint_tso)
            .min()
    }

    fn max_expired_checkpoint_ts(&self) -> Option<u64> {
        self.expired_feeds
            .iter()
            .map(|feed| feed.checkpoint_tso)
            .max()
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn feed(id: &str, state: &str, checkpoint_tso: u64) -> ReplicationTaskItem {
        ReplicationTaskItem {
            id: id.to_string(),
            state: state.to_string(),
            checkpoint_tso,
            checkpoint_time: String::new(),
            error: None,
        }
    }

    #[test]
    fn test_changefeed_gc_view() {
        let now = TimeStamp::compose(chrono::Utc::now().timestamp_millis() as u64, 0);
        let view = ChangefeedGcView::new(
            vec![
                feed("finished", "finished", now.into_inner()),
                feed("unknown", "paused-new-state", now.into_inner()),
            ],
            now,
            Duration::from_secs(60),
        );

        assert_eq!(view.synced_feeds.len(), 2);
        assert_eq!(view.synced_feeds[0].id, "finished");
        assert_eq!(view.synced_feeds[1].id, "unknown");
        assert!(view.expired_feeds.is_empty());
    }

    #[test]
    fn test_changefeed_gc_view_separates_synced_and_expired_feeds() {
        let now_ms = chrono::Utc::now().timestamp_millis() as u64;
        let now = TimeStamp::compose(now_ms, 0);
        let fresh = TimeStamp::compose(now_ms.saturating_sub(500), 0).into_inner();
        let stale = TimeStamp::compose(now_ms.saturating_sub(5_000), 0).into_inner();
        let view = ChangefeedGcView::new(
            vec![
                feed("fresh", "normal", fresh),
                feed("stale", "failed", stale),
            ],
            now,
            Duration::from_secs(1),
        );

        assert_eq!(view.min_synced_checkpoint_ts(), Some(fresh));
        assert_eq!(view.max_expired_checkpoint_ts(), Some(stale));
        assert_eq!(view.synced_feeds.len(), 1);
        assert_eq!(view.expired_feeds.len(), 1);
    }
}