use pd_proto_tso::{
    ErrorType, FindGroupByKeyspaceIdRequest, FindGroupByKeyspaceIdResponse, GetMinTsRequest,
    GetMinTsResponse, ResponseHeader, TsoRequest, TsoResponse,
};
use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering as AtomicOrdering};
use std::sync::{Arc, Mutex, RwLock};
use std::time::Instant;

use crate::error::{GrpcCode, GrpcStatus, ServiceError};
use crate::manager::{KeyspaceGroupManager, PreparedTsoRequest};
use crate::metrics;
use crate::runtime_config::RuntimeConfigState;

const MAX_GET_MIN_TS_CALLER_WINDOWS: usize = 8_192;
const GET_MIN_TS_CALLER_WINDOW_TTL_SECS: i64 = 300;

#[derive(Debug, Clone, Copy, Default)]
struct QpsWindow {
    sec: i64,
    count: u32,
}

impl QpsWindow {
    fn allow(&mut self, now_ms: i64, limit: u32) -> bool {
        if limit == 0 {
            return true;
        }
        let sec = now_ms.saturating_div(1_000);
        if self.sec != sec {
            self.sec = sec;
            self.count = 0;
        }
        if self.count >= limit {
            return false;
        }
        self.count = self.count.saturating_add(1);
        true
    }
}

/// Per-stream route-cache entry used to reduce manager lock/hash-map pressure.
#[derive(Clone, Default)]
pub(crate) struct TsoRequestRouteCache {
    keyspace_id: Option<u32>,
    requested_group_id: Option<u32>,
    route_epoch: u64,
    prepared: Option<PreparedTsoRequest>,
}

impl TsoRequestRouteCache {
    fn cached_prepared(
        &self,
        keyspace_id: u32,
        requested_group_id: u32,
        route_epoch: u64,
    ) -> Option<&PreparedTsoRequest> {
        if self.keyspace_id == Some(keyspace_id)
            && self.requested_group_id == Some(requested_group_id)
            && self.route_epoch == route_epoch
        {
            return self.prepared.as_ref();
        }
        None
    }

    fn cache_prepared(
        &mut self,
        keyspace_id: u32,
        requested_group_id: u32,
        route_epoch: u64,
        prepared: PreparedTsoRequest,
    ) {
        self.keyspace_id = Some(keyspace_id);
        self.requested_group_id = Some(requested_group_id);
        self.route_epoch = route_epoch;
        self.prepared = Some(prepared);
    }

    fn invalidate(&mut self) {
        self.keyspace_id = None;
        self.requested_group_id = None;
        self.route_epoch = 0;
        self.prepared = None;
    }
}

/// Lock-lean TSO request executor used by stream workers.
#[derive(Clone)]
pub(crate) struct TsoRequestExecutor {
    cluster_id: u64,
    started: Arc<AtomicBool>,
    draining: Arc<AtomicBool>,
    membership_resync_blocking: Arc<AtomicBool>,
    manager: Arc<RwLock<KeyspaceGroupManager>>,
    runtime_config: Arc<RwLock<RuntimeConfigState>>,
    route_epoch: Arc<AtomicU64>,
}

/// Runtime service facade for TSO APIs.
pub struct TsoService {
    cluster_id: u64,
    started: Arc<AtomicBool>,
    draining: Arc<AtomicBool>,
    membership_resync_blocking: Arc<AtomicBool>,
    manager: Arc<RwLock<KeyspaceGroupManager>>,
    runtime_config: Arc<RwLock<RuntimeConfigState>>,
    route_epoch: Arc<AtomicU64>,
    get_min_ts_global_window: Mutex<QpsWindow>,
    get_min_ts_per_caller_windows: Mutex<HashMap<u64, QpsWindow>>,
}

impl TsoService {
    /// Returns the serving cluster id used for request-header validation.
    #[must_use]
    pub fn cluster_id(&self) -> u64 {
        self.cluster_id
    }

    /// Creates a started service facade bound to one in-memory manager.
    #[must_use]
    pub fn new(cluster_id: u64, manager: KeyspaceGroupManager) -> Self {
        let route_epoch = manager.route_epoch_counter();
        let runtime_config = RuntimeConfigState {
            allocator: manager.allocator_config(),
            ..RuntimeConfigState::default()
        };
        Self {
            cluster_id,
            started: Arc::new(AtomicBool::new(true)),
            draining: Arc::new(AtomicBool::new(false)),
            membership_resync_blocking: Arc::new(AtomicBool::new(false)),
            manager: Arc::new(RwLock::new(manager)),
            runtime_config: Arc::new(RwLock::new(runtime_config)),
            route_epoch,
            get_min_ts_global_window: Mutex::new(QpsWindow::default()),
            get_min_ts_per_caller_windows: Mutex::new(HashMap::new()),
        }
    }

    /// Starts serving and clears draining state.
    pub fn start(&self) {
        self.started.store(true, AtomicOrdering::Release);
        self.draining.store(false, AtomicOrdering::Release);
    }

    /// Stops serving new requests.
    pub fn stop(&self) {
        self.started.store(false, AtomicOrdering::Release);
    }

    /// Returns whether the service is currently started.
    #[must_use]
    pub fn is_started(&self) -> bool {
        self.started.load(AtomicOrdering::Acquire)
    }

    /// Returns whether membership resync safety gate is currently blocking serving/campaign.
    #[must_use]
    pub fn is_membership_resync_blocking(&self) -> bool {
        self.membership_resync_blocking
            .load(AtomicOrdering::Acquire)
    }

    /// Sets membership resync safety gate for fail-closed watch recovery.
    pub fn set_membership_resync_blocking(&self, blocking: bool) {
        self.membership_resync_blocking
            .store(blocking, AtomicOrdering::Release);
    }

    /// Returns whether the service is currently in draining mode.
    #[must_use]
    pub fn is_draining(&self) -> bool {
        self.draining.load(AtomicOrdering::Acquire)
    }

    /// Sets draining mode for ingress gating.
    pub fn set_draining(&self, draining: bool) {
        self.draining.store(draining, AtomicOrdering::Release);
    }

    /// Returns immutable access to keyspace-group manager state.
    pub fn manager(&self) -> std::sync::RwLockReadGuard<'_, KeyspaceGroupManager> {
        self.manager
            .read()
            .unwrap_or_else(|poison| poison.into_inner())
    }

    /// Returns mutable access to manager state for control-plane operations.
    pub fn manager_mut(&self) -> std::sync::RwLockWriteGuard<'_, KeyspaceGroupManager> {
        self.manager
            .write()
            .unwrap_or_else(|poison| poison.into_inner())
    }

    /// Returns current runtime-applied hot-reload configuration.
    #[must_use]
    pub fn runtime_config(&self) -> RuntimeConfigState {
        *self
            .runtime_config
            .read()
            .unwrap_or_else(|poison| poison.into_inner())
    }

    /// Applies runtime hot-reload configuration snapshot.
    pub fn apply_runtime_config(&self, config: RuntimeConfigState) {
        let mut guard = self
            .runtime_config
            .write()
            .unwrap_or_else(|poison| poison.into_inner());
        *guard = config;
    }

    /// Returns a lock-lean request executor for stream worker hot paths.
    #[must_use]
    pub(crate) fn tso_executor(&self) -> TsoRequestExecutor {
        TsoRequestExecutor {
            cluster_id: self.cluster_id,
            started: Arc::clone(&self.started),
            draining: Arc::clone(&self.draining),
            membership_resync_blocking: Arc::clone(&self.membership_resync_blocking),
            manager: Arc::clone(&self.manager),
            runtime_config: Arc::clone(&self.runtime_config),
            route_epoch: Arc::clone(&self.route_epoch),
        }
    }

    /// Handles one streaming `Tso` message.
    pub fn tso(&self, now_ms: i64, request: TsoRequest) -> Result<TsoResponse, GrpcStatus> {
        let mut cache = TsoRequestRouteCache::default();
        self.tso_executor().handle_tso(now_ms, request, &mut cache)
    }

    /// Handles `FindGroupByKeyspaceID` with unary envelope semantics.
    #[must_use]
    pub fn find_group_by_keyspace_id(
        &self,
        now_ms: i64,
        request: FindGroupByKeyspaceIdRequest,
    ) -> FindGroupByKeyspaceIdResponse {
        let response_group = request.header.keyspace_group_id;
        if let Err(err) = self.validate_unary_header(&request.header) {
            return FindGroupByKeyspaceIdResponse {
                header: ResponseHeader::with_error(
                    self.cluster_id,
                    response_group,
                    err.to_unary_error(),
                ),
                keyspace_group: None,
                mod_revision: self.manager().mod_revision(),
            };
        }

        let manager = self.manager();
        if request.mod_revision > manager.mod_revision() {
            return FindGroupByKeyspaceIdResponse {
                header: ResponseHeader::with_error(
                    self.cluster_id,
                    response_group,
                    pd_proto_tso::Error::new(
                        ErrorType::InvalidValue,
                        ServiceError::StaleRevision.to_string(),
                    ),
                ),
                keyspace_group: None,
                mod_revision: manager.mod_revision(),
            };
        }

        match manager.find_group_by_keyspace_id(
            now_ms,
            request.keyspace_id,
            request.header.keyspace_group_id,
        ) {
            Ok(result) => FindGroupByKeyspaceIdResponse {
                header: ResponseHeader::ok(self.cluster_id, result.resolved_group_id),
                keyspace_group: Some(result.group),
                mod_revision: result.mod_revision,
            },
            Err(err) => FindGroupByKeyspaceIdResponse {
                header: ResponseHeader::with_error(
                    self.cluster_id,
                    error_group_id(&err, response_group),
                    err.to_unary_error(),
                ),
                keyspace_group: None,
                mod_revision: manager.mod_revision(),
            },
        }
    }

    /// Handles `GetMinTS` with unary envelope semantics.
    pub fn get_min_ts(&self, now_ms: i64, request: GetMinTsRequest) -> GetMinTsResponse {
        let response_group = request.header.keyspace_group_id;
        if let Err(err) = self.validate_unary_header(&request.header) {
            metrics::observe_get_min_ts_request("error");
            return GetMinTsResponse {
                header: ResponseHeader::with_error(
                    self.cluster_id,
                    response_group,
                    err.to_unary_error(),
                ),
                timestamp: Default::default(),
                keyspace_groups_serving: 0,
                keyspace_groups_total: 0,
            };
        }
        let runtime = self.runtime_config();
        if runtime.get_min_ts_authz_allow_sender_id != 0
            && request.header.sender_id != runtime.get_min_ts_authz_allow_sender_id
        {
            metrics::observe_get_min_ts_request("authz_denied");
            metrics::observe_get_min_ts_authz_denied();
            return GetMinTsResponse {
                header: ResponseHeader::with_error(
                    self.cluster_id,
                    response_group,
                    pd_proto_tso::Error::new(
                        ErrorType::Unknown,
                        "get_min_ts caller is not authorized".to_string(),
                    ),
                ),
                timestamp: Default::default(),
                keyspace_groups_serving: 0,
                keyspace_groups_total: 0,
            };
        }
        if !self.allow_get_min_ts(now_ms, request.header.sender_id) {
            metrics::observe_get_min_ts_request("rate_limited");
            metrics::observe_get_min_ts_rate_limited();
            return GetMinTsResponse {
                header: ResponseHeader::with_error(
                    self.cluster_id,
                    response_group,
                    pd_proto_tso::Error::new(
                        ErrorType::Unknown,
                        "get_min_ts rate limit exceeded".to_string(),
                    ),
                ),
                timestamp: Default::default(),
                keyspace_groups_serving: 0,
                keyspace_groups_total: 0,
            };
        }

        let mut manager = self.manager_mut();
        match manager.get_min_ts(now_ms) {
            Ok(summary) => {
                metrics::observe_get_min_ts_request("success");
                GetMinTsResponse {
                    header: ResponseHeader::ok(self.cluster_id, response_group),
                    timestamp: summary.timestamp,
                    keyspace_groups_serving: summary.keyspace_groups_serving,
                    keyspace_groups_total: summary.keyspace_groups_total,
                }
            }
            Err(err) => {
                metrics::observe_get_min_ts_request("error");
                GetMinTsResponse {
                    header: ResponseHeader::with_error(
                        self.cluster_id,
                        response_group,
                        err.to_unary_error(),
                    ),
                    timestamp: Default::default(),
                    keyspace_groups_serving: 0,
                    keyspace_groups_total: 0,
                }
            }
        }
    }

    fn validate_unary_header(
        &self,
        header: &pd_proto_tso::RequestHeader,
    ) -> Result<(), ServiceError> {
        if !self.started.load(AtomicOrdering::Acquire) {
            return Err(ServiceError::NotStarted);
        }
        if header.cluster_id != self.cluster_id {
            return Err(ServiceError::ClusterMismatched {
                expected: self.cluster_id,
                got: header.cluster_id,
            });
        }
        Ok(())
    }

    fn allow_get_min_ts(&self, now_ms: i64, sender_id: u64) -> bool {
        let config = self.runtime_config();
        if config.get_min_ts_rate_limit_qps_global > 0 {
            let mut global = self
                .get_min_ts_global_window
                .lock()
                .unwrap_or_else(|poison| poison.into_inner());
            if !global.allow(now_ms, config.get_min_ts_rate_limit_qps_global) {
                return false;
            }
        }
        if config.get_min_ts_rate_limit_qps_per_caller > 0 {
            let mut windows = self
                .get_min_ts_per_caller_windows
                .lock()
                .unwrap_or_else(|poison| poison.into_inner());
            let current_sec = now_ms.saturating_div(1_000);
            prune_caller_windows(&mut windows, current_sec);
            if !windows.contains_key(&sender_id) && windows.len() >= MAX_GET_MIN_TS_CALLER_WINDOWS {
                evict_oldest_caller_window(&mut windows);
            }
            let window = windows.entry(sender_id).or_default();
            if !window.allow(now_ms, config.get_min_ts_rate_limit_qps_per_caller) {
                return false;
            }
        }
        true
    }
}

impl TsoRequestExecutor {
    fn runtime_config(&self) -> RuntimeConfigState {
        *self
            .runtime_config
            .read()
            .unwrap_or_else(|poison| poison.into_inner())
    }

    fn validate_stream_header(
        &self,
        header: &pd_proto_tso::RequestHeader,
    ) -> Result<(), GrpcStatus> {
        if !self.started.load(AtomicOrdering::Acquire) {
            return Err(GrpcStatus::new(
                GrpcCode::Unavailable,
                ServiceError::NotStarted.to_string(),
            ));
        }
        if self.draining.load(AtomicOrdering::Acquire) {
            return Err(GrpcStatus::new(
                GrpcCode::Unavailable,
                "server is draining and rejecting new tso ingress",
            )
            .with_retry_after_ms(100));
        }
        if self
            .membership_resync_blocking
            .load(AtomicOrdering::Acquire)
        {
            return Err(GrpcStatus::new(
                GrpcCode::Unavailable,
                "membership resync in progress; rejecting tso ingress",
            )
            .with_retry_after_ms(100));
        }
        if header.cluster_id != self.cluster_id {
            return Err(ServiceError::ClusterMismatched {
                expected: self.cluster_id,
                got: header.cluster_id,
            }
            .to_grpc_status());
        }
        Ok(())
    }

    fn validate_tso_request_count(
        &self,
        count: u32,
        config: RuntimeConfigState,
    ) -> Result<(), GrpcStatus> {
        if config.max_tso_count_per_request > 0 && count > config.max_tso_count_per_request {
            return Err(GrpcStatus::new(
                GrpcCode::InvalidArgument,
                format!(
                    "tso request count {} exceeds max-tso-count-per-request {}",
                    count, config.max_tso_count_per_request
                ),
            ));
        }
        Ok(())
    }

    pub(crate) fn handle_tso(
        &self,
        now_ms: i64,
        request: TsoRequest,
        route_cache: &mut TsoRequestRouteCache,
    ) -> Result<TsoResponse, GrpcStatus> {
        let started_at = Instant::now();
        let request_count = request.count;
        let result = (|| {
            self.validate_stream_header(&request.header)?;
            let runtime = self.runtime_config();
            self.validate_tso_request_count(request.count, runtime)?;
            let overload_hardening = runtime.enable_tso_overload_hardening;
            let overload_max_in_flight_per_group = runtime.tso_overload_max_in_flight_per_group;
            let route_epoch = self.route_epoch.load(AtomicOrdering::Acquire);

            if let Some(prepared) = route_cache.cached_prepared(
                request.header.keyspace_id,
                request.header.keyspace_group_id,
                route_epoch,
            ) {
                let timestamp = prepared
                    .generate_tso(
                        now_ms,
                        request.count,
                        overload_hardening,
                        overload_max_in_flight_per_group,
                    )
                    .map_err(|err| err.to_grpc_status())?;
                return Ok(TsoResponse {
                    header: ResponseHeader::ok(self.cluster_id, prepared.resolved_group_id()),
                    count: request.count,
                    timestamp,
                });
            }

            let prepared = {
                let manager = self
                    .manager
                    .read()
                    .unwrap_or_else(|poison| poison.into_inner());
                manager.prepare_tso_request(
                    request.header.keyspace_id,
                    request.header.keyspace_group_id,
                )
            };

            let (timestamp, resolved_group_id) = match prepared {
                Ok(prepared) if !prepared.requires_slow_path() => {
                    route_cache.cache_prepared(
                        request.header.keyspace_id,
                        request.header.keyspace_group_id,
                        route_epoch,
                        prepared.clone(),
                    );
                    (
                        prepared
                            .generate_tso(
                                now_ms,
                                request.count,
                                overload_hardening,
                                overload_max_in_flight_per_group,
                            )
                            .map_err(|err| err.to_grpc_status())?,
                        prepared.resolved_group_id(),
                    )
                }
                Ok(_) => {
                    route_cache.invalidate();
                    let mut manager = self
                        .manager
                        .write()
                        .unwrap_or_else(|poison| poison.into_inner());
                    manager
                        .handle_tso_request(
                            now_ms,
                            request.header.keyspace_id,
                            request.header.keyspace_group_id,
                            request.count,
                            overload_hardening,
                            overload_max_in_flight_per_group,
                        )
                        .map_err(|err| err.to_grpc_status())?
                }
                Err(err) => {
                    route_cache.invalidate();
                    return Err(err.to_grpc_status());
                }
            };

            Ok(TsoResponse {
                header: ResponseHeader::ok(self.cluster_id, resolved_group_id),
                count: request.count,
                timestamp,
            })
        })();
        metrics::observe_tso_request(
            result.is_ok(),
            request_count,
            started_at.elapsed().as_secs_f64(),
        );
        result
    }
}

fn prune_caller_windows(windows: &mut HashMap<u64, QpsWindow>, current_sec: i64) {
    windows.retain(|_, window| {
        current_sec.saturating_sub(window.sec) <= GET_MIN_TS_CALLER_WINDOW_TTL_SECS
    });
}

fn evict_oldest_caller_window(windows: &mut HashMap<u64, QpsWindow>) {
    let oldest = windows
        .iter()
        .min_by_key(|(_, window)| window.sec)
        .map(|(sender_id, _)| *sender_id);
    if let Some(sender_id) = oldest {
        windows.remove(&sender_id);
    }
}

fn error_group_id(err: &ServiceError, default: u32) -> u32 {
    match err {
        ServiceError::NotServed { group_id } | ServiceError::KeyspaceGroupMerging { group_id } => {
            *group_id
        }
        _ => default,
    }
}

#[cfg(test)]
mod tests {
    use std::fs;
    use std::path::Path;
    use std::str::FromStr;
    use std::sync::Arc;

    use pd_proto_tso::{FindGroupByKeyspaceIdRequest, RequestHeader, TsoRequest};

    use crate::allocator::AllocatorConfig;
    use crate::manager::KeyspaceGroupManager;
    use crate::metrics;
    use crate::persistence::InMemoryEtcdStore;
    use crate::runtime_config::RuntimeConfigState;

    use super::TsoService;

    fn parse_report_field(content: &str, key: &str) -> Option<String> {
        content.lines().find_map(|line| {
            let trimmed = line.trim();
            let (parsed_key, parsed_value) = trimmed.split_once(':')?;
            if parsed_key.trim() == key {
                Some(parsed_value.trim().to_string())
            } else {
                None
            }
        })
    }

    fn parse_report_f64(content: &str, key: &str) -> f64 {
        let raw = parse_report_field(content, key)
            .unwrap_or_else(|| panic!("missing benchmark gate field `{key}`"));
        f64::from_str(&raw).unwrap_or_else(|err| {
            panic!("invalid numeric benchmark gate field `{key}` (`{raw}`): {err}")
        })
    }

    fn parse_report_u64(content: &str, key: &str) -> u64 {
        let raw = parse_report_field(content, key)
            .unwrap_or_else(|| panic!("missing benchmark gate field `{key}`"));
        u64::from_str(&raw).unwrap_or_else(|err| {
            panic!("invalid integer benchmark gate field `{key}` (`{raw}`): {err}")
        })
    }

    #[test]
    fn stream_cluster_mismatch_returns_failed_precondition() {
        let store = Arc::new(InMemoryEtcdStore::new());
        let manager = KeyspaceGroupManager::new(1, store, AllocatorConfig::default(), "node");
        let service = TsoService::new(1, manager);

        let err = service
            .tso(
                1_000,
                TsoRequest {
                    header: RequestHeader {
                        cluster_id: 99,
                        ..RequestHeader::default()
                    },
                    count: 1,
                    dc_location: String::new(),
                },
            )
            .expect_err("must fail");

        assert_eq!(err.code, crate::error::GrpcCode::FailedPrecondition);
        assert_eq!(err.message, "mismatch cluster id, need 1 but got 99");
    }

    #[test]
    fn unary_errors_are_returned_in_header() {
        let store = Arc::new(InMemoryEtcdStore::new());
        let manager = KeyspaceGroupManager::new(1, store, AllocatorConfig::default(), "node");
        let service = TsoService::new(1, manager);

        let resp = service.find_group_by_keyspace_id(
            1_000,
            FindGroupByKeyspaceIdRequest {
                header: RequestHeader {
                    cluster_id: 2,
                    keyspace_group_id: 7,
                    ..RequestHeader::default()
                },
                keyspace_id: 1,
                mod_revision: 0,
            },
        );

        assert_eq!(resp.header.keyspace_group_id, 7);
        assert!(resp.header.error.is_some());
    }

    #[test]
    fn draining_stream_rejection_includes_retry_hint() {
        let store = Arc::new(InMemoryEtcdStore::new());
        let manager = KeyspaceGroupManager::new(1, store, AllocatorConfig::default(), "node");
        let service = TsoService::new(1, manager);
        service.set_draining(true);

        let err = service
            .tso(
                1_000,
                TsoRequest {
                    header: RequestHeader {
                        cluster_id: 1,
                        ..RequestHeader::default()
                    },
                    count: 1,
                    dc_location: String::new(),
                },
            )
            .expect_err("draining must reject stream");
        assert_eq!(err.code, crate::error::GrpcCode::Unavailable);
        assert_eq!(err.retry_after_ms, Some(100));
    }

    #[test]
    fn membership_resync_blocking_rejects_stream_with_retry_hint() {
        let store = Arc::new(InMemoryEtcdStore::new());
        let manager = KeyspaceGroupManager::new(1, store, AllocatorConfig::default(), "node");
        let service = TsoService::new(1, manager);
        service.set_membership_resync_blocking(true);

        let err = service
            .tso(
                1_000,
                TsoRequest {
                    header: RequestHeader {
                        cluster_id: 1,
                        ..RequestHeader::default()
                    },
                    count: 1,
                    dc_location: String::new(),
                },
            )
            .expect_err("membership resync block must reject stream");
        assert_eq!(err.code, crate::error::GrpcCode::Unavailable);
        assert_eq!(err.retry_after_ms, Some(100));
    }

    #[test]
    fn tso_metrics_registered_after_requests() {
        let store = Arc::new(InMemoryEtcdStore::new());
        let mut manager = KeyspaceGroupManager::new(1, store, AllocatorConfig::default(), "node");
        manager
            .register_group(pd_proto_tso::KeyspaceGroup {
                id: 0,
                user_kind: "basic".to_string(),
                split_state: None,
                merge_state: None,
                members: Vec::new(),
                keyspaces: vec![0],
            })
            .expect("register");
        manager
            .campaign_group_primary(0, 1_000)
            .expect("campaign primary");
        let service = TsoService::new(1, manager);

        let _ = service.tso(
            1_000,
            TsoRequest {
                header: RequestHeader {
                    cluster_id: 1,
                    keyspace_id: 0,
                    keyspace_group_id: 0,
                    sender_id: 0,
                },
                count: 1,
                dc_location: String::new(),
            },
        );
        let _ = service.tso(
            1_000,
            TsoRequest {
                header: RequestHeader {
                    cluster_id: 9,
                    keyspace_id: 0,
                    keyspace_group_id: 0,
                    sender_id: 0,
                },
                count: 1,
                dc_location: String::new(),
            },
        );

        let names = metrics::gather_metric_names();
        assert!(names.iter().any(|name| name == "tso_requests_total"));
        assert!(
            names
                .iter()
                .any(|name| name == "tso_request_latency_seconds")
        );
        assert!(names.iter().any(|name| name == "tso_timestamps_total"));
    }

    #[test]
    fn runtime_max_tso_count_limit_rejects_large_request() {
        let store = Arc::new(InMemoryEtcdStore::new());
        let mut manager = KeyspaceGroupManager::new(1, store, AllocatorConfig::default(), "node");
        manager
            .register_group(pd_proto_tso::KeyspaceGroup {
                id: 0,
                user_kind: "basic".to_string(),
                split_state: None,
                merge_state: None,
                members: Vec::new(),
                keyspaces: vec![0],
            })
            .expect("register");
        manager
            .campaign_group_primary(0, 1_000)
            .expect("campaign primary");
        let service = TsoService::new(1, manager);

        let mut runtime = service.runtime_config();
        runtime.max_tso_count_per_request = 2;
        service.apply_runtime_config(runtime);

        let err = service
            .tso(
                1_000,
                TsoRequest {
                    header: RequestHeader {
                        cluster_id: 1,
                        keyspace_id: 0,
                        keyspace_group_id: 0,
                        sender_id: 0,
                    },
                    count: 3,
                    dc_location: String::new(),
                },
            )
            .expect_err("count over limit should fail");
        assert_eq!(err.code, crate::error::GrpcCode::InvalidArgument);
        assert!(err.message.contains("max-tso-count-per-request"));
    }

    #[test]
    fn runtime_get_min_ts_rate_limit_rejects_excess_requests() {
        let store = Arc::new(InMemoryEtcdStore::new());
        let mut manager = KeyspaceGroupManager::new(1, store, AllocatorConfig::default(), "node");
        manager
            .register_group(pd_proto_tso::KeyspaceGroup {
                id: 0,
                user_kind: "basic".to_string(),
                split_state: None,
                merge_state: None,
                members: Vec::new(),
                keyspaces: vec![0],
            })
            .expect("register");
        manager
            .campaign_group_primary(0, 1_000)
            .expect("campaign primary");
        let service = TsoService::new(1, manager);

        let mut runtime = RuntimeConfigState {
            allocator: service.manager().allocator_config(),
            ..RuntimeConfigState::default()
        };
        runtime.get_min_ts_rate_limit_qps_global = 1;
        service.apply_runtime_config(runtime);

        let first = service.get_min_ts(
            1_000,
            pd_proto_tso::GetMinTsRequest {
                header: RequestHeader {
                    cluster_id: 1,
                    keyspace_id: 0,
                    keyspace_group_id: 0,
                    sender_id: 42,
                },
                dc_location: String::new(),
            },
        );
        assert!(first.header.error.is_none());

        let second = service.get_min_ts(
            1_000,
            pd_proto_tso::GetMinTsRequest {
                header: RequestHeader {
                    cluster_id: 1,
                    keyspace_id: 0,
                    keyspace_group_id: 0,
                    sender_id: 42,
                },
                dc_location: String::new(),
            },
        );
        assert!(
            second.header.error.is_some(),
            "second request must be rate-limited"
        );
    }

    #[test]
    fn runtime_get_min_ts_authz_guardrail_denies_unmatched_sender() {
        let store = Arc::new(InMemoryEtcdStore::new());
        let mut manager = KeyspaceGroupManager::new(1, store, AllocatorConfig::default(), "node");
        manager
            .register_group(pd_proto_tso::KeyspaceGroup {
                id: 0,
                user_kind: "basic".to_string(),
                split_state: None,
                merge_state: None,
                members: Vec::new(),
                keyspaces: vec![0],
            })
            .expect("register");
        manager
            .campaign_group_primary(0, 1_000)
            .expect("campaign primary");
        let service = TsoService::new(1, manager);

        let mut runtime = RuntimeConfigState {
            allocator: service.manager().allocator_config(),
            ..RuntimeConfigState::default()
        };
        runtime.get_min_ts_authz_allow_sender_id = 7;
        service.apply_runtime_config(runtime);

        let denied = service.get_min_ts(
            1_000,
            pd_proto_tso::GetMinTsRequest {
                header: RequestHeader {
                    cluster_id: 1,
                    keyspace_id: 0,
                    keyspace_group_id: 0,
                    sender_id: 42,
                },
                dc_location: String::new(),
            },
        );
        assert!(
            denied.header.error.is_some(),
            "request must be authz denied"
        );
    }

    #[test]
    fn get_min_ts_per_caller_windows_are_capacity_bounded() {
        let store = Arc::new(InMemoryEtcdStore::new());
        let manager = KeyspaceGroupManager::new(1, store, AllocatorConfig::default(), "node");
        let service = TsoService::new(1, manager);

        let mut runtime = service.runtime_config();
        runtime.get_min_ts_rate_limit_qps_per_caller = 1;
        service.apply_runtime_config(runtime);

        let total_senders = super::MAX_GET_MIN_TS_CALLER_WINDOWS + 128;
        for sender_id in 0..total_senders as u64 {
            assert!(
                service.allow_get_min_ts(10_000, sender_id),
                "new sender should pass first request in window"
            );
        }

        let windows = service
            .get_min_ts_per_caller_windows
            .lock()
            .unwrap_or_else(|poison| poison.into_inner());
        assert!(
            windows.len() <= super::MAX_GET_MIN_TS_CALLER_WINDOWS,
            "per-caller window map must be bounded"
        );
    }

    #[test]
    fn get_min_ts_per_caller_windows_prune_stale_entries() {
        let store = Arc::new(InMemoryEtcdStore::new());
        let manager = KeyspaceGroupManager::new(1, store, AllocatorConfig::default(), "node");
        let service = TsoService::new(1, manager);

        let mut runtime = service.runtime_config();
        runtime.get_min_ts_rate_limit_qps_per_caller = 1;
        service.apply_runtime_config(runtime);

        assert!(service.allow_get_min_ts(1_000, 7));
        let stale_cutoff_ms = (super::GET_MIN_TS_CALLER_WINDOW_TTL_SECS + 10) * 1_000;
        assert!(service.allow_get_min_ts(stale_cutoff_ms, 8));

        let windows = service
            .get_min_ts_per_caller_windows
            .lock()
            .unwrap_or_else(|poison| poison.into_inner());
        assert!(
            !windows.contains_key(&7),
            "stale caller window should be pruned by TTL"
        );
        assert!(windows.contains_key(&8), "new caller window should remain");
    }

    #[test]
    fn strict_profile_benchmark_baseline_report_is_present_and_quantitative() {
        let path = Path::new(env!("CARGO_MANIFEST_DIR"))
            .join("../../../docs/benchmarks/pd-rs-tso-strict-profile-baseline.md");
        assert!(
            path.exists(),
            "benchmark baseline report must exist: {path:?}"
        );

        let content = fs::read_to_string(&path).expect("read benchmark baseline report");
        for required in [
            "# pd-rs TSO Strict-Profile Hot-Path Baseline Report",
            "cargo bench -p pd-service-tso --bench tso_hot_path",
            "## Machine-Readable Gate Fields",
            "routing_local_tso_count_1_median_ns",
            "routing_local_tso_count_64_median_ns",
            "forwarding_merge_and_split_32_median_ns",
            "q_node_sustained_qps_groups_1",
            "q_node_sustained_qps_groups_16",
            "q_node_sustained_qps_groups_128",
        ] {
            assert!(
                content.contains(required),
                "benchmark baseline report missing required marker: {required}"
            );
        }

        assert!(
            !content.contains("baseline recorded from local benchmark run"),
            "benchmark baseline report must include concrete measured values, not placeholders"
        );

        let routing_1_median = parse_report_f64(&content, "routing_local_tso_count_1_median_ns");
        let routing_1_max = parse_report_f64(&content, "routing_local_tso_count_1_max_allowed_ns");
        let routing_64_median = parse_report_f64(&content, "routing_local_tso_count_64_median_ns");
        let routing_64_max =
            parse_report_f64(&content, "routing_local_tso_count_64_max_allowed_ns");
        let forwarding_median =
            parse_report_f64(&content, "forwarding_merge_and_split_32_median_ns");
        let forwarding_max =
            parse_report_f64(&content, "forwarding_merge_and_split_32_max_allowed_ns");
        let forwarding_throughput = parse_report_f64(
            &content,
            "forwarding_merge_and_split_32_throughput_elem_per_sec",
        );
        let forwarding_min_throughput = parse_report_f64(
            &content,
            "forwarding_merge_and_split_32_min_allowed_throughput_elem_per_sec",
        );

        assert!(routing_1_median > 0.0);
        assert!(routing_64_median > 0.0);
        assert!(forwarding_median > 0.0);
        assert!(forwarding_throughput > 0.0);
        assert!(routing_1_max >= routing_1_median);
        assert!(routing_64_max >= routing_64_median);
        assert!(forwarding_max >= forwarding_median);
        assert!(forwarding_throughput >= forwarding_min_throughput);

        for required in [
            "q_node_sustained_qps_groups_1",
            "q_node_sustained_qps_groups_16",
            "q_node_sustained_qps_groups_128",
            "safe_active_group_budget_groups_1",
            "safe_active_group_budget_groups_16",
            "safe_active_group_budget_groups_128",
        ] {
            assert!(
                parse_report_u64(&content, required) > 0,
                "field `{required}` must be a positive integer"
            );
        }
    }
}