use std::collections::HashMap;
use std::fs::{self, OpenOptions};
use std::io::Write;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

use etcd_client::{Client, Compare, CompareOp, GetOptions, PutOptions, Txn, TxnOp};
use pd_proto_grpc::admin::v1::{
    ApplyConfigRequest, ApplyConfigResponse, ConfigDiffEntry, ConfigHistoryDiffRequest,
    ConfigHistoryDiffResponse, ConfigHistoryItem, ConfigViolation, DrainRequest, DrainResponse,
    GetGroupStateRequest, GetGroupStateResponse, GroupState, PrefetchStatus, SetLogLevelRequest,
    SetLogLevelResponse, TimestampParts, TransferPrimaryRequest, TransferPrimaryResponse,
    TransferPrimaryResult, ValidateConfigRequest, ValidateConfigResponse, admin_server::Admin,
    drain_response, transfer_primary_result,
};
use pd_proto_tso::KeyspaceGroup;
use prost::Message;
use prost_types::Timestamp as ProstTimestamp;
use serde::{Deserialize, Serialize};
use tokio::sync::{Mutex, RwLock};
use tonic::{Request, Response, Status};
use tracing::{info, warn};

use crate::logging::{init_runtime_logging, set_runtime_log_level};
use crate::manager::GroupElectionMeta;
use crate::metrics;
use crate::runtime_config::RuntimeConfigState;
use crate::service::TsoService;

#[path = "admin/config_validation.rs"]
mod config_validation;
#[path = "admin/idempotency_codec.rs"]
mod idempotency_codec;

use config_validation::{
    predicted_apply_mode, strict_profile_guardrails, validate_bool, validate_enum, validate_i64,
    validate_i64_min, validate_i64_non_negative, validate_i64_range, validate_i64_special_range,
    validate_tso_interval_order,
};
use idempotency_codec::{
    decode_etcd_idempotency_record, decode_hex, encode_etcd_idempotency_record, encode_hex,
    parse_idempotency_disk_record, render_idempotency_disk_line,
};

const ADMIN_API_VERSION: &str = "v1";
const PERSISTED_CONFIG_STATE_SCHEMA_VERSION: u32 = 1;
const IDEMPOTENCY_FILE_COMPACT_WRITE_INTERVAL: u32 = 256;
const IDEMPOTENCY_FILE_COMPACT_MAX_BYTES: u64 = 16 * 1024 * 1024;
const MIN_IDEMPOTENCY_RETENTION_MS: i64 = 24 * 60 * 60 * 1_000;

/// In-memory config state used by admin APIs for validation/history operations.
#[derive(Debug, Clone, Default)]
pub struct AdminConfigState {
    /// Currently active normalized runtime/admin configuration snapshot.
    pub active: HashMap<String, String>,
    /// Bounded history of previously applied config snapshots.
    pub history: Vec<ConfigHistoryItem>,
    /// Epoch of the currently active configuration snapshot.
    pub active_epoch: u64,
}

#[derive(Debug, Serialize, Deserialize)]
struct PersistedConfigState {
    #[serde(default)]
    schema_version: u32,
    active: HashMap<String, String>,
    history: Vec<PersistedConfigHistoryItem>,
    active_epoch: u64,
}

#[derive(Debug, Serialize, Deserialize)]
struct PersistedConfigHistoryItem {
    epoch: u64,
    applied_at: String,
    reason: String,
    config: HashMap<String, String>,
}

#[derive(Clone)]
struct CacheEntry<T> {
    created_at_ms: i64,
    response: T,
}

struct IdempotencyCache {
    drain: HashMap<String, CacheEntry<DrainResponse>>,
    transfer: HashMap<String, CacheEntry<TransferPrimaryResponse>>,
    apply_config: HashMap<String, CacheEntry<ApplyConfigResponse>>,
    set_log_level: HashMap<String, CacheEntry<SetLogLevelResponse>>,
    path: PathBuf,
    retention_ms: i64,
    writes_since_compaction: u32,
}

impl Default for IdempotencyCache {
    fn default() -> Self {
        Self::load_or_create(
            default_idempotency_path(),
            default_idempotency_retention_ms(),
        )
        .unwrap_or_else(|_| Self {
            drain: HashMap::new(),
            transfer: HashMap::new(),
            apply_config: HashMap::new(),
            set_log_level: HashMap::new(),
            path: default_idempotency_path(),
            retention_ms: default_idempotency_retention_ms(),
            writes_since_compaction: 0,
        })
    }
}

impl IdempotencyCache {
    fn load_or_create(path: PathBuf, retention_ms: i64) -> Result<Self, std::io::Error> {
        let mut cache = Self {
            drain: HashMap::new(),
            transfer: HashMap::new(),
            apply_config: HashMap::new(),
            set_log_level: HashMap::new(),
            path,
            retention_ms,
            writes_since_compaction: 0,
        };
        cache.load_from_disk()?;
        let _ = cache.compact_disk();
        Ok(cache)
    }

    fn ensure_retention_ms(&mut self, required_retention_ms: i64) {
        self.retention_ms = self
            .retention_ms
            .max(required_retention_ms.max(MIN_IDEMPOTENCY_RETENTION_MS));
    }

    fn is_expired(&self, created_at_ms: i64, now_ms: i64) -> bool {
        now_ms.saturating_sub(created_at_ms) > self.retention_ms
    }

    fn get_drain(&mut self, request_id: &str, now_ms: i64) -> Option<DrainResponse> {
        let created_at_ms = self
            .drain
            .get(request_id)
            .map(|entry| entry.created_at_ms)?;
        if self.is_expired(created_at_ms, now_ms) {
            self.drain.remove(request_id);
            log_expired_idempotency_replay(
                "drain",
                request_id,
                created_at_ms,
                now_ms,
                self.retention_ms,
                "local-cache",
            );
            return None;
        }
        self.drain
            .get(request_id)
            .map(|entry| entry.response.clone())
    }

    fn insert_drain(&mut self, request_id: String, response: DrainResponse, now_ms: i64) {
        self.prune_expired(now_ms);
        let entry = CacheEntry {
            created_at_ms: now_ms,
            response: response.clone(),
        };
        self.drain.insert(request_id.clone(), entry);
        let _ =
            self.persist_record_with_retention('d', &request_id, now_ms, &response.encode_to_vec());
    }

    fn get_transfer(&mut self, request_id: &str, now_ms: i64) -> Option<TransferPrimaryResponse> {
        let created_at_ms = self
            .transfer
            .get(request_id)
            .map(|entry| entry.created_at_ms)?;
        if self.is_expired(created_at_ms, now_ms) {
            self.transfer.remove(request_id);
            log_expired_idempotency_replay(
                "transfer_primary",
                request_id,
                created_at_ms,
                now_ms,
                self.retention_ms,
                "local-cache",
            );
            return None;
        }
        self.transfer
            .get(request_id)
            .map(|entry| entry.response.clone())
    }

    fn insert_transfer(
        &mut self,
        request_id: String,
        response: TransferPrimaryResponse,
        now_ms: i64,
    ) {
        self.prune_expired(now_ms);
        let entry = CacheEntry {
            created_at_ms: now_ms,
            response: response.clone(),
        };
        self.transfer.insert(request_id.clone(), entry);
        let _ =
            self.persist_record_with_retention('t', &request_id, now_ms, &response.encode_to_vec());
    }

    fn get_apply_config(&mut self, request_id: &str, now_ms: i64) -> Option<ApplyConfigResponse> {
        let created_at_ms = self
            .apply_config
            .get(request_id)
            .map(|entry| entry.created_at_ms)?;
        if self.is_expired(created_at_ms, now_ms) {
            self.apply_config.remove(request_id);
            log_expired_idempotency_replay(
                "apply_config",
                request_id,
                created_at_ms,
                now_ms,
                self.retention_ms,
                "local-cache",
            );
            return None;
        }
        self.apply_config
            .get(request_id)
            .map(|entry| entry.response.clone())
    }

    fn insert_apply_config(
        &mut self,
        request_id: String,
        response: ApplyConfigResponse,
        now_ms: i64,
    ) {
        self.prune_expired(now_ms);
        let entry = CacheEntry {
            created_at_ms: now_ms,
            response: response.clone(),
        };
        self.apply_config.insert(request_id.clone(), entry);
        let _ =
            self.persist_record_with_retention('a', &request_id, now_ms, &response.encode_to_vec());
    }

    fn get_set_log_level(&mut self, request_id: &str, now_ms: i64) -> Option<SetLogLevelResponse> {
        let created_at_ms = self
            .set_log_level
            .get(request_id)
            .map(|entry| entry.created_at_ms)?;
        if self.is_expired(created_at_ms, now_ms) {
            self.set_log_level.remove(request_id);
            log_expired_idempotency_replay(
                "set_log_level",
                request_id,
                created_at_ms,
                now_ms,
                self.retention_ms,
                "local-cache",
            );
            return None;
        }
        self.set_log_level
            .get(request_id)
            .map(|entry| entry.response.clone())
    }

    fn insert_set_log_level(
        &mut self,
        request_id: String,
        response: SetLogLevelResponse,
        now_ms: i64,
    ) {
        self.prune_expired(now_ms);
        let entry = CacheEntry {
            created_at_ms: now_ms,
            response: response.clone(),
        };
        self.set_log_level.insert(request_id.clone(), entry);
        let _ =
            self.persist_record_with_retention('l', &request_id, now_ms, &response.encode_to_vec());
    }

    fn load_from_disk(&mut self) -> Result<(), std::io::Error> {
        if !self.path.exists() {
            if let Some(parent) = self.path.parent() {
                fs::create_dir_all(parent)?;
            }
            let _ = OpenOptions::new()
                .create(true)
                .append(true)
                .open(&self.path)?;
            return Ok(());
        }

        let raw = fs::read_to_string(&self.path)?;
        for line in raw.lines() {
            let Some((kind, created_at_ms, request_id, payload_raw)) =
                parse_idempotency_disk_record(line)
            else {
                continue;
            };
            let Ok(payload) = decode_hex(&payload_raw) else {
                continue;
            };

            match kind.as_str() {
                "d" => {
                    if let Ok(response) = DrainResponse::decode(payload.as_slice()) {
                        self.drain.insert(
                            request_id.clone(),
                            CacheEntry {
                                created_at_ms,
                                response,
                            },
                        );
                    }
                }
                "t" => {
                    if let Ok(response) = TransferPrimaryResponse::decode(payload.as_slice()) {
                        self.transfer.insert(
                            request_id.clone(),
                            CacheEntry {
                                created_at_ms,
                                response,
                            },
                        );
                    }
                }
                "a" => {
                    if let Ok(response) = ApplyConfigResponse::decode(payload.as_slice()) {
                        self.apply_config.insert(
                            request_id.clone(),
                            CacheEntry {
                                created_at_ms,
                                response,
                            },
                        );
                    }
                }
                "l" => {
                    if let Ok(response) = SetLogLevelResponse::decode(payload.as_slice()) {
                        self.set_log_level.insert(
                            request_id.clone(),
                            CacheEntry {
                                created_at_ms,
                                response,
                            },
                        );
                    }
                }
                _ => {}
            }
        }
        Ok(())
    }

    fn persist_record(
        &self,
        kind: char,
        request_id: &str,
        created_at_ms: i64,
        payload: &[u8],
    ) -> Result<(), std::io::Error> {
        if let Some(parent) = self.path.parent() {
            fs::create_dir_all(parent)?;
        }
        let mut file = OpenOptions::new()
            .create(true)
            .append(true)
            .open(&self.path)?;
        let line = format!(
            "{}\n",
            render_idempotency_disk_line(kind, created_at_ms, request_id, payload)
        );
        file.write_all(line.as_bytes())
    }

    fn persist_record_with_retention(
        &mut self,
        kind: char,
        request_id: &str,
        created_at_ms: i64,
        payload: &[u8],
    ) -> Result<(), std::io::Error> {
        self.persist_record(kind, request_id, created_at_ms, payload)?;
        self.maybe_compact_disk(created_at_ms)
    }

    fn maybe_compact_disk(&mut self, now_ms: i64) -> Result<(), std::io::Error> {
        self.writes_since_compaction = self.writes_since_compaction.saturating_add(1);
        let file_size = fs::metadata(&self.path)
            .map(|meta| meta.len())
            .unwrap_or_default();
        if self.writes_since_compaction < IDEMPOTENCY_FILE_COMPACT_WRITE_INTERVAL
            && file_size <= IDEMPOTENCY_FILE_COMPACT_MAX_BYTES
        {
            return Ok(());
        }
        self.prune_expired(now_ms);
        self.compact_disk()?;
        self.writes_since_compaction = 0;
        Ok(())
    }

    fn compact_disk(&self) -> Result<(), std::io::Error> {
        if let Some(parent) = self.path.parent() {
            fs::create_dir_all(parent)?;
        }
        let mut lines = Vec::new();
        for (request_id, entry) in &self.drain {
            lines.push(render_idempotency_disk_line(
                'd',
                entry.created_at_ms,
                request_id,
                &entry.response.encode_to_vec(),
            ));
        }
        for (request_id, entry) in &self.transfer {
            lines.push(render_idempotency_disk_line(
                't',
                entry.created_at_ms,
                request_id,
                &entry.response.encode_to_vec(),
            ));
        }
        for (request_id, entry) in &self.apply_config {
            lines.push(render_idempotency_disk_line(
                'a',
                entry.created_at_ms,
                request_id,
                &entry.response.encode_to_vec(),
            ));
        }
        for (request_id, entry) in &self.set_log_level {
            lines.push(render_idempotency_disk_line(
                'l',
                entry.created_at_ms,
                request_id,
                &entry.response.encode_to_vec(),
            ));
        }
        lines.sort_unstable();

        let tmp_path = self.path.with_extension("compact.tmp");
        let mut file = OpenOptions::new()
            .create(true)
            .write(true)
            .truncate(true)
            .open(&tmp_path)?;
        for line in lines {
            file.write_all(line.as_bytes())?;
            file.write_all(b"\n")?;
        }
        file.sync_all()?;
        fs::rename(tmp_path, &self.path)
    }

    fn prune_expired(&mut self, now_ms: i64) {
        let retention_ms = self.retention_ms;
        self.drain
            .retain(|_, entry| now_ms.saturating_sub(entry.created_at_ms) <= retention_ms);
        self.transfer
            .retain(|_, entry| now_ms.saturating_sub(entry.created_at_ms) <= retention_ms);
        self.apply_config
            .retain(|_, entry| now_ms.saturating_sub(entry.created_at_ms) <= retention_ms);
        self.set_log_level
            .retain(|_, entry| now_ms.saturating_sub(entry.created_at_ms) <= retention_ms);
    }
}

#[derive(Debug, Clone)]
pub(crate) struct AdminSecurityConfig {
    enabled: bool,
    require_mtls_identity: bool,
    write_token: Option<String>,
    read_token: Option<String>,
}

#[derive(Clone, Copy)]
enum AdminPermission {
    Read,
    Write,
}

#[derive(Clone, Copy, Debug)]
struct TestVerifiedMtlsIdentity;

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ConfigStateSaveFence {
    MissingKey,
    MatchModRevision(i64),
}

/// gRPC admin service implementation aligned to Section-18 contract shape.
#[derive(Clone)]
pub struct AdminGrpcService {
    service: Arc<RwLock<TsoService>>,
    config: Arc<Mutex<AdminConfigState>>,
    idempotency: Arc<Mutex<IdempotencyCache>>,
    durable_idempotency: Option<Arc<Mutex<EtcdIdempotencyStore>>>,
    durable_config_store: Option<Arc<Mutex<EtcdConfigStore>>>,
    mutation_locks: Arc<Mutex<HashMap<String, Arc<Mutex<()>>>>>,
    apply_config_lock: Arc<Mutex<()>>,
    security: AdminSecurityConfig,
}

impl AdminGrpcService {
    /// Creates one admin gRPC service with default security settings derived from environment.
    #[must_use]
    pub fn new(service: Arc<RwLock<TsoService>>, config: Arc<Mutex<AdminConfigState>>) -> Self {
        Self::new_with_security(service, config, default_admin_security_config())
    }

    #[must_use]
    pub(crate) fn new_with_security(
        service: Arc<RwLock<TsoService>>,
        config: Arc<Mutex<AdminConfigState>>,
        security: AdminSecurityConfig,
    ) -> Self {
        let default_log_level =
            std::env::var("PD_TSO_LOG_LEVEL").unwrap_or_else(|_| "info".to_string());
        let _ = init_runtime_logging(&default_log_level);

        let retention_ms = default_idempotency_retention_ms();
        let path = default_idempotency_path();
        let idempotency = IdempotencyCache::load_or_create(path, retention_ms).unwrap_or_default();
        let durable_idempotency = if uses_etcd_admin_transfer_path() {
            Some(Arc::new(Mutex::new(EtcdIdempotencyStore::new(
                parse_admin_etcd_endpoints(),
                default_idempotency_etcd_prefix(),
                retention_ms,
            ))))
        } else {
            None
        };
        let durable_config_store = if uses_etcd_admin_transfer_path() {
            Some(Arc::new(Mutex::new(EtcdConfigStore::new(
                parse_admin_etcd_endpoints(),
                default_config_state_etcd_key(),
            ))))
        } else {
            None
        };
        Self {
            service,
            config,
            idempotency: Arc::new(Mutex::new(idempotency)),
            durable_idempotency,
            durable_config_store,
            mutation_locks: Arc::new(Mutex::new(HashMap::new())),
            apply_config_lock: Arc::new(Mutex::new(())),
            security,
        }
    }

    fn authorize<T>(
        &self,
        request: &Request<T>,
        endpoint: &str,
        permission: AdminPermission,
    ) -> Result<String, Box<Status>> {
        let metadata = request.metadata();
        let caller_identity = caller_identity_from_metadata(metadata);
        if !self.security.enabled {
            return Ok(caller_identity);
        }
        if caller_identity == "anonymous" {
            return Err(Box::new(Status::unauthenticated(
                "x-caller-id is required for admin API",
            )));
        }
        if self.security.require_mtls_identity && !request_has_transport_mtls_identity(request) {
            return Err(Box::new(Status::unauthenticated(
                "transport-authenticated mTLS client identity is required for admin API",
            )));
        }
        let token = admin_auth_token_from_metadata(metadata).ok_or_else(|| {
            Box::new(Status::unauthenticated(
                "authorization token is required for admin API",
            ))
        })?;
        let write_allowed = self
            .security
            .write_token
            .as_deref()
            .is_some_and(|expected| expected == token);
        let read_allowed = write_allowed
            || self
                .security
                .read_token
                .as_deref()
                .is_some_and(|expected| expected == token);
        let allowed = match permission {
            AdminPermission::Read => read_allowed,
            AdminPermission::Write => write_allowed,
        };
        if !allowed {
            return Err(Box::new(Status::permission_denied(format!(
                "caller is not authorized for admin endpoint {endpoint}"
            ))));
        }
        Ok(caller_identity)
    }

    fn validate_proposed_config(
        proposed: &HashMap<String, String>,
        active: &HashMap<String, String>,
        active_epoch: u64,
    ) -> ValidateConfigResponse {
        let mut violations = Vec::new();
        let mut normalized = active.clone();
        normalized.extend(proposed.clone());

        if !normalized.contains_key("compatibility-profile") {
            normalized.insert("compatibility-profile".to_string(), "strict".to_string());
        }

        validate_i64_range(
            &normalized,
            "tso-update-physical-interval-ms",
            1,
            10_000,
            &mut violations,
        );
        validate_i64_min(&normalized, "max-gap-reset-ts", 1, &mut violations);
        validate_i64_min(&normalized, "max-gap-reset-ts-ms", 1, &mut violations);
        validate_i64_min(&normalized, "tso-save-interval-ms", 1, &mut violations);
        validate_i64_non_negative(
            &normalized,
            "get-min-ts-rate-limit-qps-per-caller",
            &mut violations,
        );
        validate_i64_non_negative(
            &normalized,
            "get-min-ts-rate-limit-qps-global",
            &mut violations,
        );
        validate_i64_non_negative(
            &normalized,
            "get-min-ts-authz-allow-sender-id",
            &mut violations,
        );
        validate_i64_special_range(
            &normalized,
            "max-tso-count-per-request",
            0,
            1,
            1_048_576,
            &mut violations,
        );
        validate_enum(
            &normalized,
            "compatibility-profile",
            &["strict", "extension"],
            &mut violations,
        );
        validate_bool(
            &normalized,
            "enable-async-tso-window-prefetch",
            &mut violations,
        );
        validate_bool(
            &normalized,
            "enable-tso-overload-hardening",
            &mut violations,
        );
        validate_i64_range(
            &normalized,
            "tso-overload-max-in-flight-per-group",
            1,
            1_048_576,
            &mut violations,
        );
        validate_i64_range(
            &normalized,
            "max-concurrent-tso-proxy-streamings",
            i64::from(i32::MIN),
            i64::from(i32::MAX),
            &mut violations,
        );
        validate_i64(
            &normalized,
            "tso-proxy-recv-from-client-timeout",
            &mut violations,
        );
        validate_i64(
            &normalized,
            "tso-proxy-stream-idle-timeout-ms",
            &mut violations,
        );
        validate_tso_interval_order(&normalized, &mut violations);

        if normalized.get("compatibility-profile").map(String::as_str) == Some("strict") {
            strict_profile_guardrails(&normalized, &mut violations);
        }

        ValidateConfigResponse {
            accepted: violations.is_empty(),
            violations,
            normalized_config: normalized,
            apply_mode: predicted_apply_mode(proposed, active).to_string(),
            candidate_epoch: active_epoch.saturating_add(1),
        }
    }

    async fn sync_config_state_from_store(&self) -> Result<(), Status> {
        let Some(store) = &self.durable_config_store else {
            return Ok(());
        };
        let mut store = store.lock().await;
        let persisted = store.load_state().await.map_err(Status::unavailable)?;
        let Some(persisted) = persisted else {
            return Ok(());
        };

        let mut apply_runtime: Option<(HashMap<String, String>, u64)> = None;
        let mut state = self.config.lock().await;
        if state.active_epoch < persisted.active_epoch {
            *state = persisted;
            apply_runtime = Some((state.active.clone(), state.active_epoch));
        }
        drop(state);
        if let Some((active, epoch)) = apply_runtime {
            self.apply_runtime_snapshot(&active, epoch).await?;
        }
        Ok(())
    }

    async fn persist_config_state(&self, state: &AdminConfigState) -> Result<(), Status> {
        let Some(store) = &self.durable_config_store else {
            return Ok(());
        };
        let mut store = store.lock().await;
        store.save_state(state).await.map_err(Status::unavailable)
    }

    async fn maybe_load_cached_apply_config(
        &self,
        idempotency_key: &str,
        now_ms: i64,
    ) -> Result<Option<ApplyConfigResponse>, Status> {
        let mut cache = self.idempotency.lock().await;
        if let Some(cached) = cache.get_apply_config(idempotency_key, now_ms) {
            return Ok(Some(cached));
        }
        drop(cache);
        if let Some(store) = &self.durable_idempotency {
            let mut store = store.lock().await;
            if let Some(cached) = store
                .get_apply_config(idempotency_key, now_ms)
                .await
                .map_err(Status::unavailable)?
            {
                let mut cache = self.idempotency.lock().await;
                cache.insert_apply_config(idempotency_key.to_string(), cached.clone(), now_ms);
                return Ok(Some(cached));
            }
        }
        Ok(None)
    }

    async fn maybe_load_cached_drain(
        &self,
        idempotency_key: &str,
        now_ms: i64,
        required_retention_ms: i64,
    ) -> Result<Option<DrainResponse>, Status> {
        let mut cache = self.idempotency.lock().await;
        cache.ensure_retention_ms(required_retention_ms);
        if let Some(cached) = cache.get_drain(idempotency_key, now_ms) {
            return Ok(Some(cached));
        }
        drop(cache);

        if let Some(store) = &self.durable_idempotency {
            let mut store = store.lock().await;
            store.ensure_retention_ms(required_retention_ms);
            if let Some(cached) = store
                .get_drain(idempotency_key, now_ms)
                .await
                .map_err(Status::unavailable)?
            {
                let mut cache = self.idempotency.lock().await;
                cache.insert_drain(idempotency_key.to_string(), cached.clone(), now_ms);
                return Ok(Some(cached));
            }
        }

        Ok(None)
    }

    async fn maybe_load_cached_transfer(
        &self,
        idempotency_key: &str,
        now_ms: i64,
    ) -> Result<Option<TransferPrimaryResponse>, Status> {
        let mut cache = self.idempotency.lock().await;
        if let Some(cached) = cache.get_transfer(idempotency_key, now_ms) {
            return Ok(Some(cached));
        }
        drop(cache);

        if let Some(store) = &self.durable_idempotency {
            let mut store = store.lock().await;
            if let Some(cached) = store
                .get_transfer(idempotency_key, now_ms)
                .await
                .map_err(Status::unavailable)?
            {
                let mut cache = self.idempotency.lock().await;
                cache.insert_transfer(idempotency_key.to_string(), cached.clone(), now_ms);
                return Ok(Some(cached));
            }
        }

        Ok(None)
    }

    async fn maybe_load_cached_set_log_level(
        &self,
        idempotency_key: &str,
        now_ms: i64,
    ) -> Result<Option<SetLogLevelResponse>, Status> {
        let mut cache = self.idempotency.lock().await;
        if let Some(cached) = cache.get_set_log_level(idempotency_key, now_ms) {
            return Ok(Some(cached));
        }
        drop(cache);

        if let Some(store) = &self.durable_idempotency {
            let mut store = store.lock().await;
            if let Some(cached) = store
                .get_set_log_level(idempotency_key, now_ms)
                .await
                .map_err(Status::unavailable)?
            {
                let mut cache = self.idempotency.lock().await;
                cache.insert_set_log_level(idempotency_key.to_string(), cached.clone(), now_ms);
                return Ok(Some(cached));
            }
        }

        Ok(None)
    }

    async fn apply_cached_drain_state(&self, state: i32) {
        let service = self.service.write().await;
        match drain_response::State::try_from(state).unwrap_or(drain_response::State::Unspecified) {
            drain_response::State::Drained => service.set_draining(true),
            drain_response::State::Timeout => service.set_draining(false),
            _ => {}
        }
    }

    async fn mutation_lock_for(&self, idempotency_key: &str) -> Arc<Mutex<()>> {
        let mut locks = self.mutation_locks.lock().await;
        Arc::clone(
            locks
                .entry(idempotency_key.to_string())
                .or_insert_with(|| Arc::new(Mutex::new(()))),
        )
    }

    async fn cleanup_mutation_lock(&self, idempotency_key: &str, lock: &Arc<Mutex<()>>) {
        let mut locks = self.mutation_locks.lock().await;
        let should_remove = locks.get(idempotency_key).is_some_and(|candidate| {
            Arc::ptr_eq(candidate, lock)
                // Keep the lock entry while other replays still hold a clone of
                // this lock, otherwise a late-arriving duplicate could create a
                // second lock and run concurrently for the same idempotency key.
                && Arc::strong_count(candidate) == 2
        });
        if should_remove {
            locks.remove(idempotency_key);
        }
    }

    async fn persist_apply_config_idempotency(
        &self,
        idempotency_key: &str,
        response: &ApplyConfigResponse,
    ) -> Result<(), Status> {
        let mut cache = self.idempotency.lock().await;
        cache.insert_apply_config(idempotency_key.to_string(), response.clone(), now_ms());
        drop(cache);
        if let Some(store) = &self.durable_idempotency {
            let mut store = store.lock().await;
            store
                .put_apply_config(idempotency_key, response, now_ms())
                .await
                .map_err(Status::unavailable)?;
        }
        Ok(())
    }

    async fn apply_runtime_snapshot(
        &self,
        active_config: &HashMap<String, String>,
        epoch: u64,
    ) -> Result<(), Status> {
        let service = self.service.write().await;
        let allocator_lease_ms = service.manager().allocator_config().lease_ms;
        let runtime_fallback = service.runtime_config();
        let runtime_next =
            RuntimeConfigState::from_map(active_config, runtime_fallback, allocator_lease_ms);

        {
            let mut manager = service.manager_mut();
            let required = manager.required_config_apply_actors();
            manager
                .apply_config_epoch_with_allocator(epoch, &required, Some(runtime_next.allocator))
                .map_err(|err| Status::failed_precondition(err.to_string()))?;
        }

        service.apply_runtime_config(runtime_next);
        metrics::set_active_config_epoch(epoch);
        Ok(())
    }
}

#[tonic::async_trait]
impl Admin for AdminGrpcService {
    async fn drain(
        &self,
        request: Request<DrainRequest>,
    ) -> Result<Response<DrainResponse>, Status> {
        let caller_identity = self
            .authorize(&request, "drain", AdminPermission::Write)
            .map_err(|status| *status)?;
        let request = request.into_inner();
        if request.request_id.is_empty() {
            return Err(Status::invalid_argument("request_id is required"));
        }
        let now = now_ms();
        let timeout_ms = if request.timeout_ms == 0 {
            300_000
        } else {
            request.timeout_ms
        };
        let required_retention_ms = required_idempotency_retention_ms_for_timeout(timeout_ms);
        let idempotency_key = idempotency_scope_key("drain", &caller_identity, &request.request_id);

        if let Some(cached) = self
            .maybe_load_cached_drain(&idempotency_key, now, required_retention_ms)
            .await?
        {
            self.apply_cached_drain_state(cached.state).await;
            return Ok(Response::new(cached));
        }

        let mutation_lock = self.mutation_lock_for(&idempotency_key).await;
        let mutation_guard = mutation_lock.lock().await;
        let result: Result<Response<DrainResponse>, Status> = async {
            if let Some(cached) = self
                .maybe_load_cached_drain(&idempotency_key, now_ms(), required_retention_ms)
                .await?
            {
                self.apply_cached_drain_state(cached.state).await;
                return Ok(Response::new(cached));
            }

            let started_at_ms = now_ms();
            let deadline_ms =
                started_at_ms.saturating_add((timeout_ms.min(i64::MAX as u64)) as i64);
            let handoff_started = Instant::now();

            let mut expected_primary_writer = ExpectedPrimaryWriter::from_env()
                .await
                .map_err(Status::unavailable)?;
            let mut handoff_candidates = Vec::new();

            {
                let serving_group_ids = {
                    let service = self.service.read().await;
                    let manager = service.manager();
                    manager.serving_group_ids(now_ms())
                };
                let service = self.service.read().await;
                for group_id in serving_group_ids {
                    let manager = service.manager();
                    let meta = match manager.group_election_meta(group_id) {
                        Ok(meta) => meta,
                        Err(err) => {
                            warn!(
                                group_id,
                                error = %err,
                                "failed to build election metadata for drain handoff"
                            );
                            continue;
                        }
                    };
                    let expected_target = match resolve_transfer_target(
                        "",
                        manager.group(group_id),
                        &meta.member_value,
                    ) {
                        Ok(target) => target,
                        Err(err) => {
                            warn!(
                                group_id,
                                error = %err,
                                "failed to resolve explicit transfer target during drain handoff"
                            );
                            continue;
                        }
                    };
                    handoff_candidates.push((meta, expected_target));
                }
            }

            {
                let service = self.service.write().await;
                service.set_draining(true);
            }

            let mut handoff_watch = Vec::with_capacity(handoff_candidates.len());
            for (meta, expected_target) in &handoff_candidates {
                if let Err(err) = expected_primary_writer
                    .write_expected_primary(meta, Some(expected_target.as_str()))
                    .await
                {
                    warn!(
                        group_id = meta.group_id,
                        error = %err,
                        "failed to persist expected-primary during drain handoff"
                    );
                    continue;
                }
                let service = self.service.write().await;
                let mut manager = service.manager_mut();
                if let Err(err) =
                    manager.set_group_expected_primary(meta.group_id, Some(expected_target.clone()))
                {
                    warn!(
                        group_id = meta.group_id,
                        error = %err,
                        "failed to set in-memory expected-primary during drain handoff"
                    );
                }
                if let Err(err) = manager.step_down_group_primary(meta.group_id) {
                    warn!(
                        group_id = meta.group_id,
                        error = %err,
                        "failed to step down local primary during drain handoff"
                    );
                }
                metrics::observe_group_rebalance_operation();
                handoff_watch.push(meta.clone());
            }

            let response = loop {
                let mut remaining = {
                    let service = self.service.read().await;
                    let manager = service.manager();
                    manager.serving_group_ids(now_ms())
                };
                let mut handoff_pending = Vec::new();
                for meta in handoff_watch {
                    match expected_primary_writer.handoff_completed(&meta).await {
                        Ok(true) => {}
                        Ok(false) => {
                            if !remaining.contains(&meta.group_id) {
                                remaining.push(meta.group_id);
                            }
                            handoff_pending.push(meta);
                        }
                        Err(err) => {
                            warn!(
                                group_id = meta.group_id,
                                error = %err,
                                "failed to verify drain handoff completion for group"
                            );
                            if !remaining.contains(&meta.group_id) {
                                remaining.push(meta.group_id);
                            }
                            handoff_pending.push(meta);
                        }
                    }
                }
                handoff_watch = handoff_pending;
                remaining.sort_unstable();
                remaining.dedup();

                if remaining.is_empty() {
                    break DrainResponse {
                        state: drain_response::State::Drained as i32,
                        remaining_primaries: 0,
                        remaining_group_ids: Vec::new(),
                        message: "drain completed".to_string(),
                    };
                }
                if now_ms() >= deadline_ms {
                    let service = self.service.write().await;
                    service.set_draining(false);
                    break DrainResponse {
                        state: drain_response::State::Timeout as i32,
                        remaining_primaries: remaining.len() as u32,
                        remaining_group_ids: remaining,
                        message: "drain timeout reached before all primaries were handed off"
                            .to_string(),
                    };
                }
                tokio::time::sleep(Duration::from_millis(50)).await;
            };
            let handoff_completed = drain_response::State::try_from(response.state)
                .unwrap_or(drain_response::State::Unspecified)
                == drain_response::State::Drained;
            metrics::observe_drain_handoff_duration(
                handoff_completed,
                handoff_started.elapsed().as_secs_f64(),
            );

            let mut cache = self.idempotency.lock().await;
            cache.insert_drain(idempotency_key.clone(), response.clone(), now_ms());
            drop(cache);
            if let Some(store) = &self.durable_idempotency {
                let mut store = store.lock().await;
                store
                    .put_drain(&idempotency_key, &response, now_ms())
                    .await
                    .map_err(Status::unavailable)?;
            }
            Ok(Response::new(response))
        }
        .await;
        drop(mutation_guard);
        self.cleanup_mutation_lock(&idempotency_key, &mutation_lock)
            .await;
        result
    }

    async fn transfer_primary(
        &self,
        request: Request<TransferPrimaryRequest>,
    ) -> Result<Response<TransferPrimaryResponse>, Status> {
        let caller_identity = self
            .authorize(&request, "transfer_primary", AdminPermission::Write)
            .map_err(|status| *status)?;
        let request = request.into_inner();
        if request.request_id.is_empty() {
            return Err(Status::invalid_argument("request_id is required"));
        }
        if request.group_ids.is_empty() {
            return Err(Status::invalid_argument("group_ids must not be empty"));
        }
        let now = now_ms();
        let idempotency_key =
            idempotency_scope_key("transfer_primary", &caller_identity, &request.request_id);

        if let Some(cached) = self
            .maybe_load_cached_transfer(&idempotency_key, now)
            .await?
        {
            return Ok(Response::new(cached));
        }

        let mutation_lock = self.mutation_lock_for(&idempotency_key).await;
        let mutation_guard = mutation_lock.lock().await;
        let result: Result<Response<TransferPrimaryResponse>, Status> = async {
            if let Some(cached) = self
                .maybe_load_cached_transfer(&idempotency_key, now_ms())
                .await?
            {
                return Ok(Response::new(cached));
            }

            let mut expected_primary_writer = ExpectedPrimaryWriter::from_env()
                .await
                .map_err(Status::unavailable)?;
            let mut results = Vec::with_capacity(request.group_ids.len());
            for group_id in &request.group_ids {
                let group_id = *group_id;
                let (state, election_meta, group_meta) = {
                    let service = self.service.read().await;
                    let manager = service.manager();
                    (
                        manager.group_debug_state(group_id, now_ms()),
                        manager.group_election_meta(group_id),
                        manager.group(group_id).cloned(),
                    )
                };
                match state {
                    Ok(debug) if !debug.is_primary => results.push(TransferPrimaryResult {
                        group_id,
                        status: transfer_primary_result::Status::SkippedNotPrimary as i32,
                        message: "group is not primary on this node".to_string(),
                    }),
                    Ok(_) => {
                        let election_meta = match election_meta {
                            Ok(meta) => meta,
                            Err(err) => {
                                results.push(TransferPrimaryResult {
                                    group_id,
                                    status: transfer_primary_result::Status::Failed as i32,
                                    message: err.to_string(),
                                });
                                continue;
                            }
                        };

                        let expected = match resolve_transfer_target(
                            &request.target_member_id,
                            group_meta.as_ref(),
                            &election_meta.member_value,
                        ) {
                            Ok(target) => target,
                            Err(err) => {
                                results.push(TransferPrimaryResult {
                                    group_id,
                                    status: transfer_primary_result::Status::Failed as i32,
                                    message: err,
                                });
                                continue;
                            }
                        };

                        if let Err(err) = expected_primary_writer
                            .write_expected_primary(&election_meta, Some(expected.as_str()))
                            .await
                        {
                            results.push(TransferPrimaryResult {
                                group_id,
                                status: transfer_primary_result::Status::Failed as i32,
                                message: format!("persist expected-primary failed: {err}"),
                            });
                            continue;
                        }

                        let service = self.service.write().await;
                        let status = {
                            let mut manager = service.manager_mut();
                            let _ = manager
                                .set_group_expected_primary(group_id, Some(expected.clone()));
                            manager.step_down_group_primary(group_id)
                        };
                        let succeeded = status.is_ok();
                        match status {
                            Ok(()) => results.push(TransferPrimaryResult {
                                group_id,
                                status: transfer_primary_result::Status::Success as i32,
                                message: "primary stepped down for transfer".to_string(),
                            }),
                            Err(err) => results.push(TransferPrimaryResult {
                                group_id,
                                status: transfer_primary_result::Status::Failed as i32,
                                message: err.to_string(),
                            }),
                        }
                        if succeeded {
                            metrics::observe_group_rebalance_operation();
                        }
                    }
                    Err(err) => results.push(TransferPrimaryResult {
                        group_id,
                        status: transfer_primary_result::Status::Failed as i32,
                        message: err.to_string(),
                    }),
                }
            }

            let response = TransferPrimaryResponse { results };
            let mut cache = self.idempotency.lock().await;
            cache.insert_transfer(idempotency_key.clone(), response.clone(), now_ms());
            drop(cache);
            if let Some(store) = &self.durable_idempotency {
                let mut store = store.lock().await;
                store
                    .put_transfer(&idempotency_key, &response, now_ms())
                    .await
                    .map_err(Status::unavailable)?;
            }
            Ok(Response::new(response))
        }
        .await;
        drop(mutation_guard);
        self.cleanup_mutation_lock(&idempotency_key, &mutation_lock)
            .await;
        result
    }

    async fn validate_config(
        &self,
        request: Request<ValidateConfigRequest>,
    ) -> Result<Response<ValidateConfigResponse>, Status> {
        let _caller_identity = self
            .authorize(&request, "validate_config", AdminPermission::Read)
            .map_err(|status| *status)?;
        let request = request.into_inner();
        if request.request_id.is_empty() {
            return Err(Status::invalid_argument("request_id is required"));
        }

        let state = self.config.lock().await;
        let response = Self::validate_proposed_config(
            &request.proposed_config,
            &state.active,
            state.active_epoch,
        );
        Ok(Response::new(response))
    }

    async fn apply_config(
        &self,
        request: Request<ApplyConfigRequest>,
    ) -> Result<Response<ApplyConfigResponse>, Status> {
        let caller_identity = self
            .authorize(&request, "apply_config", AdminPermission::Write)
            .map_err(|status| *status)?;
        let request = request.into_inner();
        if request.request_id.is_empty() {
            return Err(Status::invalid_argument("request_id is required"));
        }
        let now = now_ms();
        let idempotency_key =
            idempotency_scope_key("apply_config", &caller_identity, &request.request_id);
        if let Some(cached) = self
            .maybe_load_cached_apply_config(&idempotency_key, now)
            .await?
        {
            return Ok(Response::new(cached));
        }

        let _guard = self.apply_config_lock.lock().await;
        if let Some(cached) = self
            .maybe_load_cached_apply_config(&idempotency_key, now_ms())
            .await?
        {
            return Ok(Response::new(cached));
        }

        metrics::config_apply_attempt("apply_config");
        if let Err(err) = self.sync_config_state_from_store().await {
            metrics::config_apply_failure("sync_store");
            return Err(err);
        }
        let (active_snapshot, active_epoch_snapshot, history_snapshot) = {
            let state = self.config.lock().await;
            (
                state.active.clone(),
                state.active_epoch,
                state.history.clone(),
            )
        };
        let validation = Self::validate_proposed_config(
            &request.proposed_config,
            &active_snapshot,
            active_epoch_snapshot,
        );
        let apply_mode = validation.apply_mode.clone();

        if !validation.accepted {
            let response = ApplyConfigResponse {
                accepted: false,
                violations: validation.violations,
                active_config: active_snapshot,
                active_epoch: active_epoch_snapshot,
                apply_mode,
                message: "config rejected by validation".to_string(),
            };
            self.persist_apply_config_idempotency(&idempotency_key, &response)
                .await?;
            metrics::config_apply_failure("validation");
            return Ok(Response::new(response));
        }

        let mut next_active = active_snapshot.clone();
        next_active.extend(request.proposed_config.clone());
        let next_epoch = active_epoch_snapshot.saturating_add(1);
        if apply_mode == "restart-required" {
            let response = ApplyConfigResponse {
                accepted: false,
                violations: vec![ConfigViolation {
                    code: "restart_required_config_change".to_string(),
                    message:
                        "one or more changed keys require restart; runtime hot-apply is rejected"
                            .to_string(),
                }],
                active_config: active_snapshot.clone(),
                active_epoch: active_epoch_snapshot,
                apply_mode,
                message: format!(
                    "config rejected because candidate epoch {next_epoch} includes restart-required key changes"
                ),
            };
            self.persist_apply_config_idempotency(&idempotency_key, &response)
                .await?;
            metrics::config_apply_failure("restart_required");
            return Ok(Response::new(response));
        }
        let (allocator_lease_ms, current_runtime_config) = {
            let service = self.service.read().await;
            let lease_ms = service.manager().allocator_config().lease_ms;
            (lease_ms, service.runtime_config())
        };
        let active_runtime_config = RuntimeConfigState::from_map(
            &active_snapshot,
            current_runtime_config,
            allocator_lease_ms,
        );
        let next_runtime_config =
            RuntimeConfigState::from_map(&next_active, active_runtime_config, allocator_lease_ms);

        let required_actors = {
            let service = self.service.read().await;
            let manager = service.manager();
            manager.required_config_apply_actors()
        };
        let _pending_guard = metrics::ConfigApplyPendingGuard::new(required_actors.len());

        let apply_result = {
            let service = self.service.write().await;
            let mut manager = service.manager_mut();
            let result = manager.apply_config_epoch_with_allocator(
                next_epoch,
                &required_actors,
                Some(next_runtime_config.allocator),
            );
            drop(manager);
            if result.is_ok() {
                service.apply_runtime_config(next_runtime_config);
            }
            result
        };
        if let Err(err) = apply_result {
            let response = ApplyConfigResponse {
                accepted: false,
                violations: vec![ConfigViolation {
                    code: "config_epoch_apply_abort".to_string(),
                    message: format!("candidate epoch {next_epoch} aborted: {err}"),
                }],
                active_config: active_snapshot,
                active_epoch: active_epoch_snapshot,
                apply_mode,
                message: format!(
                    "config apply aborted because not all required actors acknowledged (required_actors={})",
                    required_actors.len()
                ),
            };
            self.persist_apply_config_idempotency(&idempotency_key, &response)
                .await?;
            metrics::config_apply_failure("epoch_apply");
            return Ok(Response::new(response));
        }

        let mut next_history = history_snapshot;
        next_history.push(ConfigHistoryItem {
            epoch: next_epoch,
            applied_at: now_ms().to_string(),
            reason: if request.reason.is_empty() {
                "manual".to_string()
            } else {
                request.reason
            },
            config: next_active.clone(),
        });
        trim_config_history(&mut next_history, default_config_history_limit());
        let persisted_snapshot = AdminConfigState {
            active: next_active.clone(),
            history: next_history.clone(),
            active_epoch: next_epoch,
        };

        if let Err(err) = self.persist_config_state(&persisted_snapshot).await {
            let rollback_result = {
                let service = self.service.write().await;
                let mut manager = service.manager_mut();
                let result = manager.apply_config_epoch_with_allocator(
                    active_epoch_snapshot,
                    &required_actors,
                    Some(active_runtime_config.allocator),
                );
                drop(manager);
                if result.is_ok() {
                    service.apply_runtime_config(active_runtime_config);
                }
                result
            };
            let mut violations = vec![ConfigViolation {
                code: "config_persist_failed".to_string(),
                message: format!("durable config-state write failed: {err}"),
            }];
            if let Err(rollback_err) = rollback_result {
                violations.push(ConfigViolation {
                    code: "config_epoch_rollback_failed".to_string(),
                    message: format!(
                        "rollback to epoch {active_epoch_snapshot} failed: {rollback_err}"
                    ),
                });
            }
            let response = ApplyConfigResponse {
                accepted: false,
                violations,
                active_config: active_snapshot,
                active_epoch: active_epoch_snapshot,
                apply_mode,
                message: "config apply aborted because durable persistence failed".to_string(),
            };
            self.persist_apply_config_idempotency(&idempotency_key, &response)
                .await?;
            metrics::config_apply_failure("persist");
            return Ok(Response::new(response));
        }

        {
            let mut state = self.config.lock().await;
            *state = persisted_snapshot;
        }

        let response = ApplyConfigResponse {
            accepted: true,
            violations: Vec::new(),
            active_config: next_active,
            active_epoch: next_epoch,
            apply_mode,
            message: "config applied".to_string(),
        };
        self.persist_apply_config_idempotency(&idempotency_key, &response)
            .await?;
        metrics::set_active_config_epoch(next_epoch);
        Ok(Response::new(response))
    }

    async fn get_group_state(
        &self,
        request: Request<GetGroupStateRequest>,
    ) -> Result<Response<GetGroupStateResponse>, Status> {
        let _caller_identity = self
            .authorize(&request, "get_group_state", AdminPermission::Read)
            .map_err(|status| *status)?;
        let request = request.into_inner();
        let service = self.service.read().await;
        let runtime_config = service.runtime_config();
        let debug = service
            .manager()
            .group_debug_state(request.group_id, now_ms())
            .map_err(|err| Status::not_found(err.to_string()))?;
        let actor_state = if service.is_draining() && debug.is_primary {
            "Draining"
        } else if debug.is_primary {
            "Serving"
        } else {
            "Standby"
        };

        Ok(Response::new(GetGroupStateResponse {
            state: Some(GroupState {
                is_primary: debug.is_primary,
                in_memory_physical: debug.in_memory_timestamp.physical,
                in_memory_logical: debug.in_memory_timestamp.logical,
                actor_state: actor_state.to_string(),
                lease_id: debug.lease_id,
                lease_expire_at_ms: debug.lease_expire_at_ms,
                last_persisted_physical: debug.last_persisted_physical_ms,
                last_persisted_logical: debug.last_persisted_logical,
                mailbox_depth: debug.mailbox_depth,
                in_flight_requests: debug.in_flight_requests,
                prefetch_window_start_physical: debug.in_memory_timestamp.physical,
                prefetch_window_end_physical: debug.save_window_end_ms,
                lease_expiration_time: timestamp_from_millis(debug.lease_expire_at_ms),
                in_memory_timestamp: Some(TimestampParts {
                    physical: debug.in_memory_timestamp.physical,
                    logical: debug.in_memory_timestamp.logical,
                }),
                last_persisted_timestamp: Some(TimestampParts {
                    physical: debug.last_persisted_physical_ms,
                    logical: debug.last_persisted_logical,
                }),
                mailbox_size: debug.mailbox_depth,
                mailbox_capacity: runtime_config.tso_overload_max_in_flight_per_group,
                prefetch_status: Some(PrefetchStatus {
                    state: debug.prefetch_state,
                    last_attempt: timestamp_from_millis(debug.prefetch_last_attempt_ms),
                    last_error: debug.prefetch_last_error,
                }),
            }),
        }))
    }

    async fn set_log_level(
        &self,
        request: Request<SetLogLevelRequest>,
    ) -> Result<Response<SetLogLevelResponse>, Status> {
        let caller_identity = self
            .authorize(&request, "set_log_level", AdminPermission::Write)
            .map_err(|status| *status)?;
        let request = request.into_inner();
        if request.request_id.is_empty() {
            return Err(Status::invalid_argument("request_id is required"));
        }
        if request.target.is_empty() {
            return Err(Status::invalid_argument("target is required"));
        }
        if request.level.is_empty() {
            return Err(Status::invalid_argument("level is required"));
        }
        let now = now_ms();
        let idempotency_key =
            idempotency_scope_key("set_log_level", &caller_identity, &request.request_id);

        if let Some(cached) = self
            .maybe_load_cached_set_log_level(&idempotency_key, now)
            .await?
        {
            return Ok(Response::new(cached));
        }

        let mutation_lock = self.mutation_lock_for(&idempotency_key).await;
        let mutation_guard = mutation_lock.lock().await;
        let result: Result<Response<SetLogLevelResponse>, Status> = async {
            if let Some(cached) = self
                .maybe_load_cached_set_log_level(&idempotency_key, now_ms())
                .await?
            {
                return Ok(Response::new(cached));
            }

            if let Err(err) = set_runtime_log_level(&request.target, &request.level) {
                return Err(Status::failed_precondition(err));
            }

            let response = SetLogLevelResponse {
                success: true,
                message: format!(
                    "log level '{}' applied to '{}'",
                    request.level, request.target
                ),
            };
            let mut cache = self.idempotency.lock().await;
            cache.insert_set_log_level(idempotency_key.clone(), response.clone(), now_ms());
            drop(cache);
            if let Some(store) = &self.durable_idempotency {
                let mut store = store.lock().await;
                store
                    .put_set_log_level(&idempotency_key, &response, now_ms())
                    .await
                    .map_err(Status::unavailable)?;
            }
            Ok(Response::new(response))
        }
        .await;
        drop(mutation_guard);
        self.cleanup_mutation_lock(&idempotency_key, &mutation_lock)
            .await;
        result
    }

    async fn config_history_diff(
        &self,
        request: Request<ConfigHistoryDiffRequest>,
    ) -> Result<Response<ConfigHistoryDiffResponse>, Status> {
        let _caller_identity = self
            .authorize(&request, "config_history_diff", AdminPermission::Read)
            .map_err(|status| *status)?;
        let request = request.into_inner();
        self.sync_config_state_from_store().await?;
        let state = self.config.lock().await;
        let mut diff = Vec::new();

        for (key, new_value) in &request.diff_with_active {
            let old_value = state.active.get(key).cloned().unwrap_or_default();
            if old_value != *new_value {
                diff.push(ConfigDiffEntry {
                    key: key.clone(),
                    old_value,
                    new_value: new_value.clone(),
                });
            }
        }

        let limit = if request.limit == 0 {
            state.history.len()
        } else {
            request.limit as usize
        };
        let history = state
            .history
            .iter()
            .rev()
            .take(limit)
            .cloned()
            .collect::<Vec<_>>();

        Ok(Response::new(ConfigHistoryDiffResponse { diff, history }))
    }
}

struct ExpectedPrimaryWriter {
    client: Option<Client>,
}

impl ExpectedPrimaryWriter {
    async fn from_env() -> Result<Self, String> {
        if !uses_etcd_admin_transfer_path() {
            return Ok(Self { client: None });
        }

        let endpoints = parse_admin_etcd_endpoints();
        let client = Client::connect(endpoints.clone(), None)
            .await
            .map_err(|err| format!("connect etcd failed for endpoints {endpoints:?}: {err}"))?;
        Ok(Self {
            client: Some(client),
        })
    }

    async fn write_expected_primary(
        &mut self,
        meta: &GroupElectionMeta,
        expected_member: Option<&str>,
    ) -> Result<(), String> {
        let Some(client) = self.client.as_mut() else {
            return Ok(());
        };

        let leader = client
            .get(meta.leader_key.clone(), None)
            .await
            .map_err(|err| format!("etcd read leader key failed: {err}"))?;
        let kv = leader
            .kvs()
            .first()
            .ok_or_else(|| "leader key is missing during transfer".to_string())?;
        let lease_id = kv.lease();
        if lease_id <= 0 {
            return Err("leader key does not carry a valid lease".to_string());
        }
        if kv.value() != meta.member_value.as_bytes() {
            return Err("leader changed before expected-primary update".to_string());
        }

        let op = match expected_member {
            Some(member) => TxnOp::put(
                meta.expected_primary_key.clone(),
                member,
                Some(PutOptions::new().with_lease(lease_id)),
            ),
            None => TxnOp::delete(meta.expected_primary_key.clone(), None),
        };
        let txn = Txn::new()
            .when(vec![Compare::value(
                meta.leader_key.clone(),
                CompareOp::Equal,
                meta.member_value.as_bytes(),
            )])
            .and_then(vec![op]);
        let resp = client
            .txn(txn)
            .await
            .map_err(|err| format!("etcd expected-primary txn failed: {err}"))?;
        if !resp.succeeded() {
            return Err("leader changed while persisting expected-primary".to_string());
        }
        Ok(())
    }

    async fn handoff_completed(&mut self, meta: &GroupElectionMeta) -> Result<bool, String> {
        let Some(client) = self.client.as_mut() else {
            return Ok(true);
        };
        let leader = client
            .get(meta.leader_key.clone(), Some(GetOptions::new()))
            .await
            .map_err(|err| format!("etcd read leader key failed: {err}"))?;
        let Some(kv) = leader.kvs().first() else {
            return Ok(false);
        };
        Ok(kv.value() != meta.member_value.as_bytes())
    }
}

struct EtcdIdempotencyStore {
    endpoints: Vec<String>,
    prefix: String,
    retention_ms: i64,
    client: Option<Client>,
}

impl EtcdIdempotencyStore {
    fn new(endpoints: Vec<String>, prefix: String, retention_ms: i64) -> Self {
        Self {
            endpoints,
            prefix,
            retention_ms,
            client: None,
        }
    }

    fn ensure_retention_ms(&mut self, required_retention_ms: i64) {
        self.retention_ms = self
            .retention_ms
            .max(required_retention_ms.max(MIN_IDEMPOTENCY_RETENTION_MS));
    }

    async fn get_drain(
        &mut self,
        idempotency_key: &str,
        now_ms: i64,
    ) -> Result<Option<DrainResponse>, String> {
        self.load("drain", idempotency_key, now_ms).await
    }

    async fn put_drain(
        &mut self,
        idempotency_key: &str,
        response: &DrainResponse,
        now_ms: i64,
    ) -> Result<(), String> {
        self.save("drain", idempotency_key, response, now_ms).await
    }

    async fn get_transfer(
        &mut self,
        idempotency_key: &str,
        now_ms: i64,
    ) -> Result<Option<TransferPrimaryResponse>, String> {
        self.load("transfer_primary", idempotency_key, now_ms).await
    }

    async fn put_transfer(
        &mut self,
        idempotency_key: &str,
        response: &TransferPrimaryResponse,
        now_ms: i64,
    ) -> Result<(), String> {
        self.save("transfer_primary", idempotency_key, response, now_ms)
            .await
    }

    async fn get_apply_config(
        &mut self,
        idempotency_key: &str,
        now_ms: i64,
    ) -> Result<Option<ApplyConfigResponse>, String> {
        self.load("apply_config", idempotency_key, now_ms).await
    }

    async fn put_apply_config(
        &mut self,
        idempotency_key: &str,
        response: &ApplyConfigResponse,
        now_ms: i64,
    ) -> Result<(), String> {
        self.save("apply_config", idempotency_key, response, now_ms)
            .await
    }

    async fn get_set_log_level(
        &mut self,
        idempotency_key: &str,
        now_ms: i64,
    ) -> Result<Option<SetLogLevelResponse>, String> {
        self.load("set_log_level", idempotency_key, now_ms).await
    }

    async fn put_set_log_level(
        &mut self,
        idempotency_key: &str,
        response: &SetLogLevelResponse,
        now_ms: i64,
    ) -> Result<(), String> {
        self.save("set_log_level", idempotency_key, response, now_ms)
            .await
    }

    async fn load<T>(
        &mut self,
        kind: &str,
        idempotency_key: &str,
        now_ms: i64,
    ) -> Result<Option<T>, String>
    where
        T: Message + Default,
    {
        let key = self.key(kind, idempotency_key);
        let retention_ms = self.retention_ms;
        let client = self.ensure_client().await?;
        let resp = client
            .get(key.clone(), None)
            .await
            .map_err(|err| format!("etcd idempotency get failed: {err}"))?;
        let Some(kv) = resp.kvs().first() else {
            return Ok(None);
        };
        let raw = String::from_utf8_lossy(kv.value());
        let Some((created_at_ms, payload)) = decode_etcd_idempotency_record(&raw) else {
            warn!(key = %key, "invalid etcd idempotency record; ignoring");
            return Ok(None);
        };
        if now_ms.saturating_sub(created_at_ms) > retention_ms {
            log_expired_idempotency_replay(
                kind,
                idempotency_key,
                created_at_ms,
                now_ms,
                retention_ms,
                "etcd-durable",
            );
            let _ = client.delete(key, None).await;
            return Ok(None);
        }
        let response = T::decode(payload.as_slice())
            .map_err(|err| format!("decode idempotency payload failed: {err}"))?;
        Ok(Some(response))
    }

    async fn save<T>(
        &mut self,
        kind: &str,
        idempotency_key: &str,
        response: &T,
        now_ms: i64,
    ) -> Result<(), String>
    where
        T: Message,
    {
        let key = self.key(kind, idempotency_key);
        let value = encode_etcd_idempotency_record(now_ms, &response.encode_to_vec());
        let ttl_secs = retention_ttl_secs(self.retention_ms);
        let client = self.ensure_client().await?;
        let lease = client
            .lease_grant(ttl_secs, None)
            .await
            .map_err(|err| format!("etcd idempotency lease grant failed: {err}"))?;
        let txn = Txn::new()
            .when(vec![Compare::create_revision(
                key.clone(),
                CompareOp::Equal,
                0,
            )])
            .and_then(vec![TxnOp::put(
                key.clone(),
                value,
                Some(PutOptions::new().with_lease(lease.id())),
            )]);
        let txn_resp = client
            .txn(txn)
            .await
            .map_err(|err| format!("etcd idempotency txn put-if-absent failed: {err}"))?;
        if !txn_resp.succeeded() {
            let _ = client.lease_revoke(lease.id()).await;
        }
        Ok(())
    }

    async fn ensure_client(&mut self) -> Result<&mut Client, String> {
        if self.client.is_none() {
            let client = Client::connect(self.endpoints.clone(), None)
                .await
                .map_err(|err| format!("connect etcd failed for idempotency store: {err}"))?;
            self.client = Some(client);
        }
        self.client
            .as_mut()
            .ok_or_else(|| "idempotency etcd client is not initialized".to_string())
    }

    fn key(&self, kind: &str, idempotency_key: &str) -> String {
        format!(
            "{}/{}/{}",
            self.prefix,
            kind,
            encode_hex(idempotency_key.as_bytes())
        )
    }
}

struct EtcdConfigStore {
    endpoints: Vec<String>,
    state_key: String,
    client: Option<Client>,
    loaded_mod_revision: Option<i64>,
}

impl EtcdConfigStore {
    fn new(endpoints: Vec<String>, state_key: String) -> Self {
        Self {
            endpoints,
            state_key,
            client: None,
            loaded_mod_revision: None,
        }
    }

    async fn load_state(&mut self) -> Result<Option<AdminConfigState>, String> {
        let key = self.state_key.clone();
        let client = self.ensure_client().await?;
        let resp = client
            .get(key.clone(), None)
            .await
            .map_err(|err| format!("etcd config-state get failed: {err}"))?;
        let Some(kv) = resp.kvs().first() else {
            self.loaded_mod_revision = Some(0);
            return Ok(None);
        };
        self.loaded_mod_revision = Some(kv.mod_revision());
        let persisted: PersistedConfigState = serde_json::from_slice(kv.value())
            .map_err(|err| format!("decode persisted config state failed: {err}"))?;
        if persisted.schema_version > PERSISTED_CONFIG_STATE_SCHEMA_VERSION {
            return Err(format!(
                "unsupported config state schema_version {}, max supported {}",
                persisted.schema_version, PERSISTED_CONFIG_STATE_SCHEMA_VERSION
            ));
        }
        Ok(Some(from_persisted_config_state(persisted)))
    }

    async fn save_state(&mut self, state: &AdminConfigState) -> Result<(), String> {
        let key = self.state_key.clone();
        let fence = config_state_save_fence(self.loaded_mod_revision)?;
        let payload = serde_json::to_vec(&to_persisted_config_state(state))
            .map_err(|err| format!("encode persisted config state failed: {err}"))?;
        let client = self.ensure_client().await?;
        let compare = match fence {
            ConfigStateSaveFence::MissingKey => {
                Compare::create_revision(key.clone(), CompareOp::Equal, 0)
            }
            ConfigStateSaveFence::MatchModRevision(mod_revision) => {
                Compare::mod_revision(key.clone(), CompareOp::Equal, mod_revision)
            }
        };
        let txn = Txn::new()
            .when(vec![compare])
            .and_then(vec![TxnOp::put(key, payload, None)]);
        let resp = client
            .txn(txn)
            .await
            .map_err(|err| format!("etcd config-state fenced write failed: {err}"))?;
        if !resp.succeeded() {
            self.loaded_mod_revision = None;
            return Err("etcd config-state write conflicted with a concurrent update".to_string());
        }
        self.loaded_mod_revision = None;
        Ok(())
    }

    async fn ensure_client(&mut self) -> Result<&mut Client, String> {
        if self.client.is_none() {
            let client = Client::connect(self.endpoints.clone(), None)
                .await
                .map_err(|err| format!("connect etcd failed for config store: {err}"))?;
            self.client = Some(client);
        }
        self.client
            .as_mut()
            .ok_or_else(|| "config-store etcd client is not initialized".to_string())
    }
}

fn to_persisted_config_state(state: &AdminConfigState) -> PersistedConfigState {
    PersistedConfigState {
        schema_version: PERSISTED_CONFIG_STATE_SCHEMA_VERSION,
        active: state.active.clone(),
        history: state
            .history
            .iter()
            .map(|item| PersistedConfigHistoryItem {
                epoch: item.epoch,
                applied_at: item.applied_at.clone(),
                reason: item.reason.clone(),
                config: item.config.clone(),
            })
            .collect(),
        active_epoch: state.active_epoch,
    }
}

fn from_persisted_config_state(state: PersistedConfigState) -> AdminConfigState {
    let mut restored = AdminConfigState {
        active: state.active,
        history: state
            .history
            .into_iter()
            .map(|item| ConfigHistoryItem {
                epoch: item.epoch,
                applied_at: item.applied_at,
                reason: item.reason,
                config: item.config,
            })
            .collect(),
        active_epoch: state.active_epoch,
    };
    trim_config_history(&mut restored.history, default_config_history_limit());
    restored
}

fn resolve_transfer_target(
    requested_target_member_id: &str,
    group_meta: Option<&KeyspaceGroup>,
    self_member_value: &str,
) -> Result<String, String> {
    if !requested_target_member_id.trim().is_empty() {
        let target = requested_target_member_id.trim().to_string();
        if target_matches_local_identity(&target, self_member_value) {
            return Err("target_member_id must not equal current primary member".to_string());
        }
        return Ok(target);
    }

    let group_meta = group_meta.ok_or_else(|| {
        "keyspace-group metadata is unavailable; cannot auto-select transfer target".to_string()
    })?;
    for member in &group_meta.members {
        let candidate = member.address.trim();
        if candidate.is_empty() {
            continue;
        }
        if !target_matches_local_identity(candidate, self_member_value) {
            return Ok(candidate.to_string());
        }
    }
    Err("no eligible transfer target found for this group".to_string())
}

fn target_matches_local_identity(candidate: &str, self_member_value: &str) -> bool {
    let mut aliases = vec![self_member_value.to_string()];
    aliases.extend(local_member_addresses_from_env());
    let normalized_candidate = normalize_identity(candidate);
    aliases.into_iter().any(|alias| {
        alias == candidate
            || normalize_identity(&alias) == normalized_candidate
            || alias == normalized_candidate
    })
}

fn normalize_identity(value: &str) -> String {
    let trimmed = value.trim().trim_end_matches('/');
    if let Some(raw) = trimmed.strip_prefix("http://") {
        return raw.to_string();
    }
    if let Some(raw) = trimmed.strip_prefix("https://") {
        return raw.to_string();
    }
    trimmed.to_string()
}

fn local_member_addresses_from_env() -> Vec<String> {
    let mut values = Vec::new();
    for key in ["PD_TSO_ADVERTISE_LISTEN_ADDR", "PD_TSO_LISTEN_ADDR"] {
        let Ok(raw) = std::env::var(key) else {
            continue;
        };
        for part in raw.split(',') {
            let value = part.trim();
            if !value.is_empty() {
                values.push(value.to_string());
            }
        }
    }
    values
}

fn default_admin_security_config() -> AdminSecurityConfig {
    let enabled = parse_env_bool("PD_TSO_ADMIN_AUTH_ENABLED", uses_etcd_admin_transfer_path());
    let require_mtls_identity = parse_env_bool("PD_TSO_ADMIN_REQUIRE_MTLS_IDENTITY", enabled);
    let write_token = std::env::var("PD_TSO_ADMIN_WRITE_TOKEN")
        .ok()
        .or_else(|| std::env::var("PD_TSO_ADMIN_AUTH_TOKEN").ok())
        .filter(|value| !value.trim().is_empty());
    let read_token = std::env::var("PD_TSO_ADMIN_READ_TOKEN")
        .ok()
        .filter(|value| !value.trim().is_empty());
    AdminSecurityConfig {
        enabled,
        require_mtls_identity,
        write_token,
        read_token,
    }
}

fn parse_env_bool(key: &str, default_value: bool) -> bool {
    std::env::var(key)
        .ok()
        .and_then(|raw| match raw.trim().to_ascii_lowercase().as_str() {
            "1" | "true" | "yes" | "on" => Some(true),
            "0" | "false" | "no" | "off" => Some(false),
            _ => None,
        })
        .unwrap_or(default_value)
}

fn request_has_transport_mtls_identity<T>(request: &Request<T>) -> bool {
    if request
        .extensions()
        .get::<TestVerifiedMtlsIdentity>()
        .is_some()
    {
        return true;
    }
    request.peer_certs().is_some_and(|certs| !certs.is_empty())
}

fn config_state_save_fence(
    loaded_mod_revision: Option<i64>,
) -> Result<ConfigStateSaveFence, String> {
    let Some(mod_revision) = loaded_mod_revision else {
        return Err(
            "config-state save requires a synchronized load_state() before write".to_string(),
        );
    };
    if mod_revision < 0 {
        return Err(format!(
            "config-state save got invalid loaded mod revision {mod_revision}"
        ));
    }
    if mod_revision == 0 {
        Ok(ConfigStateSaveFence::MissingKey)
    } else {
        Ok(ConfigStateSaveFence::MatchModRevision(mod_revision))
    }
}

fn admin_auth_token_from_metadata(metadata: &tonic::metadata::MetadataMap) -> Option<String> {
    if let Some(raw) = metadata
        .get("authorization")
        .and_then(|value| value.to_str().ok())
    {
        let trimmed = raw.trim();
        if let Some(token) = trimmed.strip_prefix("Bearer ") {
            let token = token.trim();
            if !token.is_empty() {
                return Some(token.to_string());
            }
        }
    }
    metadata
        .get("x-admin-token")
        .and_then(|value| value.to_str().ok())
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .map(ToString::to_string)
}

fn uses_etcd_admin_transfer_path() -> bool {
    !std::env::var("PD_TSO_STORE_EFFECTIVE")
        .or_else(|_| std::env::var("PD_TSO_STORE"))
        .unwrap_or_else(|_| "etcd".to_string())
        .eq_ignore_ascii_case("in-memory")
}

fn parse_admin_etcd_endpoints() -> Vec<String> {
    std::env::var("PD_TSO_ETCD_ENDPOINTS_EFFECTIVE")
        .or_else(|_| std::env::var("PD_TSO_ETCD_ENDPOINTS"))
        .ok()
        .unwrap_or_else(|| "http://127.0.0.1:2379".to_string())
        .split(',')
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .map(ToString::to_string)
        .collect()
}

fn caller_identity_from_metadata(metadata: &tonic::metadata::MetadataMap) -> String {
    metadata
        .get("x-caller-id")
        .and_then(|value| value.to_str().ok())
        .filter(|value| !value.trim().is_empty())
        .map(ToString::to_string)
        .unwrap_or_else(|| "anonymous".to_string())
}

fn idempotency_scope_key(endpoint: &str, caller_identity: &str, request_id: &str) -> String {
    format!("{ADMIN_API_VERSION}\t{endpoint}\t{caller_identity}\t{request_id}")
}

fn log_expired_idempotency_replay(
    endpoint: &str,
    idempotency_key: &str,
    created_at_ms: i64,
    now_ms: i64,
    retention_ms: i64,
    backend: &str,
) {
    info!(
        endpoint,
        backend,
        idempotency_key,
        created_at_ms,
        now_ms,
        retention_ms,
        age_ms = now_ms.saturating_sub(created_at_ms),
        "expired idempotency replay treated as new request"
    );
}

fn default_idempotency_path() -> PathBuf {
    std::env::var("PD_TSO_ADMIN_IDEMPOTENCY_PATH")
        .ok()
        .map(PathBuf::from)
        .unwrap_or_else(|| std::env::temp_dir().join("pd-tso-admin-idempotency.log"))
}

fn default_idempotency_etcd_prefix() -> String {
    if let Ok(prefix) = std::env::var("PD_TSO_ADMIN_IDEMPOTENCY_ETCD_PREFIX")
        && !prefix.trim().is_empty()
    {
        return prefix;
    }
    let cluster_id = std::env::var("PD_TSO_CLUSTER_ID").unwrap_or_else(|_| "0".to_string());
    format!("/pd/{cluster_id}/tso/admin/idempotency")
}

fn default_config_state_etcd_key() -> String {
    if let Ok(key) = std::env::var("PD_TSO_ADMIN_CONFIG_STATE_ETCD_KEY")
        && !key.trim().is_empty()
    {
        return key;
    }
    let cluster_id = std::env::var("PD_TSO_CLUSTER_ID").unwrap_or_else(|_| "0".to_string());
    format!("/pd/{cluster_id}/tso/admin/config/state")
}

fn default_config_history_limit() -> usize {
    std::env::var("PD_TSO_ADMIN_CONFIG_HISTORY_LIMIT")
        .ok()
        .and_then(|raw| raw.parse::<usize>().ok())
        .unwrap_or(256)
        .max(1)
}

fn default_idempotency_retention_ms() -> i64 {
    std::env::var("PD_TSO_ADMIN_IDEMPOTENCY_RETENTION_MS")
        .ok()
        .and_then(|raw| raw.parse::<i64>().ok())
        .unwrap_or(MIN_IDEMPOTENCY_RETENTION_MS)
        .max(MIN_IDEMPOTENCY_RETENTION_MS)
}

fn required_idempotency_retention_ms_for_timeout(timeout_ms: u64) -> i64 {
    let timeout_ms = i64::try_from(timeout_ms).unwrap_or(i64::MAX);
    timeout_ms
        .saturating_mul(2)
        .max(MIN_IDEMPOTENCY_RETENTION_MS)
}

fn retention_ttl_secs(retention_ms: i64) -> i64 {
    retention_ms
        .saturating_add(999)
        .saturating_div(1_000)
        .max(1)
}

fn trim_config_history(history: &mut Vec<ConfigHistoryItem>, limit: usize) {
    if history.len() > limit {
        let remove_count = history.len() - limit;
        history.drain(0..remove_count);
    }
}

fn timestamp_from_millis(timestamp_ms: i64) -> Option<ProstTimestamp> {
    if timestamp_ms <= 0 {
        return None;
    }
    let seconds = timestamp_ms.div_euclid(1_000);
    let millis = timestamp_ms.rem_euclid(1_000);
    Some(ProstTimestamp {
        seconds,
        nanos: (millis as i32) * 1_000_000,
    })
}

fn now_ms() -> i64 {
    let duration = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default();
    (duration.as_millis().min(i64::MAX as u128)) as i64
}

#[cfg(test)]
mod tests {
    use super::{
        AdminGrpcService, AdminPermission, AdminSecurityConfig, ConfigStateSaveFence,
        ConfigViolation, MIN_IDEMPOTENCY_RETENTION_MS, TestVerifiedMtlsIdentity,
        config_state_save_fence, decode_etcd_idempotency_record, decode_hex,
        default_idempotency_retention_ms, encode_etcd_idempotency_record, encode_hex,
        idempotency_scope_key, parse_idempotency_disk_record, render_idempotency_disk_line,
        required_idempotency_retention_ms_for_timeout, resolve_transfer_target, retention_ttl_secs,
        strict_profile_guardrails, trim_config_history, validate_i64_range,
    };
    use std::collections::HashMap;
    use std::sync::Arc;

    use pd_test_util::env::{EnvVarRestore, async_env_lock, remove_env, set_env};
    use tokio::sync::{Mutex, RwLock};
    use tonic::Request;

    #[test]
    fn strict_profile_rejects_non_parity_knobs() {
        let mut violations = Vec::<ConfigViolation>::new();
        let mut config = HashMap::new();
        config.insert(
            "enable-async-tso-window-prefetch".to_string(),
            "true".to_string(),
        );
        strict_profile_guardrails(&config, &mut violations);
        assert!(!violations.is_empty());
    }

    #[test]
    fn range_validation_reports_out_of_range_values() {
        let mut violations = Vec::<ConfigViolation>::new();
        let mut config = HashMap::new();
        config.insert(
            "tso-update-physical-interval-ms".to_string(),
            "10001".to_string(),
        );
        validate_i64_range(
            &config,
            "tso-update-physical-interval-ms",
            1,
            10_000,
            &mut violations,
        );
        assert_eq!(violations.len(), 1);
    }

    #[test]
    fn hex_codec_round_trip() {
        let payload = vec![1, 2, 3, 250, 255];
        let encoded = encode_hex(&payload);
        let decoded = decode_hex(&encoded).expect("decode");
        assert_eq!(decoded, payload);
    }

    #[test]
    fn idempotency_disk_record_round_trip_supports_scoped_keys_with_tabs() {
        let request_id = idempotency_scope_key("drain", "caller-a", "req-1");
        let payload = vec![9, 8, 7];
        let line = render_idempotency_disk_line('d', 1234, &request_id, &payload);
        let (kind, created_at_ms, parsed_request_id, payload_hex) =
            parse_idempotency_disk_record(&line).expect("parse");
        assert_eq!(kind, "d");
        assert_eq!(created_at_ms, 1234);
        assert_eq!(parsed_request_id, request_id);
        assert_eq!(decode_hex(&payload_hex).expect("decode payload"), payload);
    }

    #[test]
    fn idempotency_disk_record_parser_recovers_buggy_v1_scoped_key_lines() {
        let request_id = idempotency_scope_key("drain", "caller-a", "req-legacy");
        let payload = vec![1, 3, 5, 7];
        let line = format!("v1\td\t42\t{request_id}\t{}", encode_hex(&payload));
        let (kind, created_at_ms, parsed_request_id, payload_hex) =
            parse_idempotency_disk_record(&line).expect("parse v1");
        assert_eq!(kind, "d");
        assert_eq!(created_at_ms, 42);
        assert_eq!(parsed_request_id, request_id);
        assert_eq!(decode_hex(&payload_hex).expect("decode payload"), payload);
    }

    #[test]
    fn default_idempotency_retention_enforces_24h_floor() {
        let _guard = async_env_lock().blocking_lock();
        let _restore = EnvVarRestore::capture("PD_TSO_ADMIN_IDEMPOTENCY_RETENTION_MS");
        set_env("PD_TSO_ADMIN_IDEMPOTENCY_RETENTION_MS", "1");
        assert_eq!(
            default_idempotency_retention_ms(),
            MIN_IDEMPOTENCY_RETENTION_MS
        );
    }

    #[test]
    fn required_idempotency_retention_scales_with_timeout() {
        assert_eq!(
            required_idempotency_retention_ms_for_timeout(300_000),
            MIN_IDEMPOTENCY_RETENTION_MS
        );
        assert_eq!(
            required_idempotency_retention_ms_for_timeout(48 * 60 * 60 * 1_000),
            96 * 60 * 60 * 1_000
        );
    }

    #[test]
    fn omitted_profile_still_enforces_strict_guardrails() {
        let mut proposed = HashMap::new();
        proposed.insert(
            "enable-async-tso-window-prefetch".to_string(),
            "true".to_string(),
        );
        let response = AdminGrpcService::validate_proposed_config(&proposed, &HashMap::new(), 7);
        assert!(!response.accepted);
        assert!(
            response
                .violations
                .iter()
                .any(|item| item.code == "strict_profile_guardrail_violation")
        );
        assert_eq!(
            response
                .normalized_config
                .get("compatibility-profile")
                .map(String::as_str),
            Some("strict")
        );
    }

    #[test]
    fn strict_profile_accepts_overload_cap_override_when_hardening_disabled() {
        let proposed = HashMap::from([
            ("compatibility-profile".to_string(), "strict".to_string()),
            (
                "tso-overload-max-in-flight-per-group".to_string(),
                "256".to_string(),
            ),
        ]);
        let response = AdminGrpcService::validate_proposed_config(&proposed, &HashMap::new(), 7);
        assert!(
            response.accepted,
            "strict profile should not reject overload cap override when overload hardening remains disabled"
        );
        assert!(
            !response
                .violations
                .iter()
                .any(|item| item.code == "strict_profile_guardrail_violation")
        );
    }

    #[test]
    fn idempotency_scope_key_includes_caller_and_endpoint() {
        let a = idempotency_scope_key("drain", "caller-a", "req-1");
        let b = idempotency_scope_key("drain", "caller-b", "req-1");
        let c = idempotency_scope_key("set_log_level", "caller-a", "req-1");
        assert_ne!(a, b);
        assert_ne!(a, c);
        assert!(a.starts_with("v1\t"));
    }

    #[test]
    fn validate_config_rejects_save_interval_smaller_than_update_interval() {
        let proposed = HashMap::from([
            ("tso-save-interval-ms".to_string(), "50".to_string()),
            (
                "tso-update-physical-interval-ms".to_string(),
                "100".to_string(),
            ),
        ]);
        let response = AdminGrpcService::validate_proposed_config(&proposed, &HashMap::new(), 1);
        assert!(!response.accepted);
        assert!(
            response
                .violations
                .iter()
                .any(|item| item.code == "tso_interval_order_invalid")
        );
    }

    #[test]
    fn validate_config_rejects_non_positive_max_gap_reset_ts_keys() {
        let canonical = HashMap::from([("max-gap-reset-ts".to_string(), "0".to_string())]);
        let canonical_response =
            AdminGrpcService::validate_proposed_config(&canonical, &HashMap::new(), 1);
        assert!(!canonical_response.accepted);
        assert!(
            canonical_response
                .violations
                .iter()
                .any(|item| item.code == "max-gap-reset-ts_too_small")
        );

        let legacy = HashMap::from([("max-gap-reset-ts-ms".to_string(), "-1".to_string())]);
        let legacy_response =
            AdminGrpcService::validate_proposed_config(&legacy, &HashMap::new(), 1);
        assert!(!legacy_response.accepted);
        assert!(
            legacy_response
                .violations
                .iter()
                .any(|item| item.code == "max-gap-reset-ts-ms_too_small")
        );
    }

    #[test]
    fn validate_config_predicts_restart_required_mode_for_profile_change() {
        let active = HashMap::from([("compatibility-profile".to_string(), "strict".to_string())]);
        let proposed =
            HashMap::from([("compatibility-profile".to_string(), "extension".to_string())]);
        let response = AdminGrpcService::validate_proposed_config(&proposed, &active, 9);
        assert_eq!(response.apply_mode, "restart-required");
        assert_eq!(response.candidate_epoch, 10);
    }

    #[test]
    fn etcd_idempotency_record_codec_round_trip() {
        let payload = vec![9, 8, 7, 6];
        let encoded = encode_etcd_idempotency_record(1234, &payload);
        let decoded = decode_etcd_idempotency_record(&encoded).expect("decode");
        assert_eq!(decoded.0, 1234);
        assert_eq!(decoded.1, payload);
    }

    #[test]
    fn retention_ttl_secs_rounds_up_and_never_returns_zero() {
        assert_eq!(retention_ttl_secs(1), 1);
        assert_eq!(retention_ttl_secs(999), 1);
        assert_eq!(retention_ttl_secs(1_000), 1);
        assert_eq!(retention_ttl_secs(1_001), 2);
    }

    #[test]
    fn transfer_target_selection_auto_picks_non_self_member() {
        let group = pd_proto_tso::KeyspaceGroup {
            id: 7,
            user_kind: "basic".to_string(),
            split_state: None,
            merge_state: None,
            members: vec![
                pd_proto_tso::KeyspaceGroupMember {
                    address: "node-a:7".to_string(),
                    is_primary: true,
                },
                pd_proto_tso::KeyspaceGroupMember {
                    address: "node-b:7".to_string(),
                    is_primary: false,
                },
            ],
            keyspaces: vec![7],
        };
        let selected = resolve_transfer_target("", Some(&group), "node-a:7").expect("selected");
        assert_eq!(selected, "node-b:7");
    }

    #[test]
    fn transfer_target_selection_rejects_self_target() {
        let err = resolve_transfer_target("node-a:7", None, "node-a:7").expect_err("must fail");
        assert!(err.contains("must not equal current primary"));
    }

    #[test]
    fn transfer_target_selection_does_not_treat_legacy_discovery_urls_as_local_identity() {
        let _guard = async_env_lock().blocking_lock();
        let [_legacy_restore, _advertise_restore, _listen_restore] = EnvVarRestore::capture_many([
            "PD_TSO_LEGACY_DISCOVERY_URLS",
            "PD_TSO_ADVERTISE_LISTEN_ADDR",
            "PD_TSO_LISTEN_ADDR",
        ]);

        set_env(
            "PD_TSO_LEGACY_DISCOVERY_URLS",
            "http://203.0.113.10:3379,http://203.0.113.11:3379",
        );
        remove_env("PD_TSO_ADVERTISE_LISTEN_ADDR");
        remove_env("PD_TSO_LISTEN_ADDR");

        let selected =
            resolve_transfer_target("http://203.0.113.10:3379", None, "node-a:7").expect("target");
        assert_eq!(selected, "http://203.0.113.10:3379");
    }

    #[test]
    fn trim_config_history_keeps_latest_entries() {
        let mut history = vec![
            pd_proto_grpc::admin::v1::ConfigHistoryItem {
                epoch: 1,
                applied_at: "1".to_string(),
                reason: "a".to_string(),
                config: HashMap::new(),
            },
            pd_proto_grpc::admin::v1::ConfigHistoryItem {
                epoch: 2,
                applied_at: "2".to_string(),
                reason: "b".to_string(),
                config: HashMap::new(),
            },
            pd_proto_grpc::admin::v1::ConfigHistoryItem {
                epoch: 3,
                applied_at: "3".to_string(),
                reason: "c".to_string(),
                config: HashMap::new(),
            },
        ];
        trim_config_history(&mut history, 2);
        assert_eq!(history.len(), 2);
        assert_eq!(history[0].epoch, 2);
        assert_eq!(history[1].epoch, 3);
    }

    #[tokio::test]
    async fn admin_authorization_requires_token_and_mtls_identity_when_enabled() {
        let store = Arc::new(crate::persistence::InMemoryEtcdStore::new());
        let manager = crate::manager::KeyspaceGroupManager::new(
            1,
            store,
            crate::allocator::AllocatorConfig::default(),
            "node",
        );
        let service = Arc::new(RwLock::new(crate::service::TsoService::new(1, manager)));
        let config = Arc::new(Mutex::new(crate::admin::AdminConfigState::default()));
        let admin = AdminGrpcService::new_with_security(
            service,
            config,
            AdminSecurityConfig {
                enabled: true,
                require_mtls_identity: true,
                write_token: Some("write-token".to_string()),
                read_token: Some("read-token".to_string()),
            },
        );

        let mut request = Request::new(());
        request
            .metadata_mut()
            .insert("x-caller-id", "tester".parse().expect("caller"));
        request.metadata_mut().insert(
            "authorization",
            "Bearer write-token".parse().expect("token"),
        );
        let unauth = admin.authorize(&request, "drain", AdminPermission::Write);
        assert!(unauth.is_err(), "mTLS identity is required");

        request.metadata_mut().insert(
            "x-tls-client-subject",
            "CN=tester".parse().expect("subject"),
        );
        let spoofed = admin.authorize(&request, "drain", AdminPermission::Write);
        assert!(
            spoofed.is_err(),
            "caller-supplied metadata must not satisfy mTLS identity checks"
        );

        request.extensions_mut().insert(TestVerifiedMtlsIdentity);
        let ok = admin.authorize(&request, "drain", AdminPermission::Write);
        assert!(ok.is_ok(), "write token + mTLS identity must pass");
    }

    #[tokio::test]
    async fn admin_authorization_allows_read_token_for_read_only_endpoints() {
        let store = Arc::new(crate::persistence::InMemoryEtcdStore::new());
        let manager = crate::manager::KeyspaceGroupManager::new(
            1,
            store,
            crate::allocator::AllocatorConfig::default(),
            "node",
        );
        let service = Arc::new(RwLock::new(crate::service::TsoService::new(1, manager)));
        let config = Arc::new(Mutex::new(crate::admin::AdminConfigState::default()));
        let admin = AdminGrpcService::new_with_security(
            service,
            config,
            AdminSecurityConfig {
                enabled: true,
                require_mtls_identity: false,
                write_token: Some("write-token".to_string()),
                read_token: Some("read-token".to_string()),
            },
        );

        let mut request = Request::new(());
        request
            .metadata_mut()
            .insert("x-caller-id", "reader".parse().expect("caller"));
        request
            .metadata_mut()
            .insert("authorization", "Bearer read-token".parse().expect("token"));

        assert!(
            admin
                .authorize(&request, "validate_config", AdminPermission::Read)
                .is_ok()
        );
        assert!(
            admin
                .authorize(&request, "apply_config", AdminPermission::Write)
                .is_err()
        );
    }

    #[tokio::test]
    async fn mutation_lock_cleanup_preserves_waiting_replays_for_same_key() {
        let store = Arc::new(crate::persistence::InMemoryEtcdStore::new());
        let manager = crate::manager::KeyspaceGroupManager::new(
            1,
            store,
            crate::allocator::AllocatorConfig::default(),
            "node",
        );
        let service = Arc::new(RwLock::new(crate::service::TsoService::new(1, manager)));
        let config = Arc::new(Mutex::new(crate::admin::AdminConfigState::default()));
        let admin = AdminGrpcService::new_with_security(
            service,
            config,
            AdminSecurityConfig {
                enabled: false,
                require_mtls_identity: false,
                write_token: None,
                read_token: None,
            },
        );

        let key = "v1\tdrain\tcaller\treq-1";
        let first_lock = admin.mutation_lock_for(key).await;
        let first_guard = first_lock.lock().await;

        let (ready_tx, ready_rx) = tokio::sync::oneshot::channel::<Arc<Mutex<()>>>();
        let (release_tx, release_rx) = tokio::sync::oneshot::channel::<()>();
        let admin_for_waiter = admin.clone();
        let key_for_waiter = key.to_string();
        let waiter = tokio::spawn(async move {
            let waiting_lock = admin_for_waiter.mutation_lock_for(&key_for_waiter).await;
            let _ = ready_tx.send(waiting_lock.clone());
            let _waiting_guard = waiting_lock.lock().await;
            let _ = release_rx.await;
        });

        let second_lock = ready_rx.await.expect("waiter lock");
        assert!(Arc::ptr_eq(&first_lock, &second_lock));

        drop(first_guard);
        admin.cleanup_mutation_lock(key, &first_lock).await;

        let third_lock = admin.mutation_lock_for(key).await;
        assert!(
            Arc::ptr_eq(&first_lock, &third_lock),
            "cleanup must not remove key while duplicate replay is waiting"
        );

        let _ = release_tx.send(());
        waiter.await.expect("waiter joins");
        drop(second_lock);
        drop(third_lock);

        admin.cleanup_mutation_lock(key, &first_lock).await;
        let locks = admin.mutation_locks.lock().await;
        assert!(
            !locks.contains_key(key),
            "lock entry should be removed after all waiters complete"
        );
    }

    #[test]
    fn config_state_save_fence_requires_loaded_revision() {
        assert!(config_state_save_fence(None).is_err());
    }

    #[test]
    fn config_state_save_fence_uses_create_revision_for_missing_state_key() {
        let fence = config_state_save_fence(Some(0)).expect("fence");
        assert_eq!(fence, ConfigStateSaveFence::MissingKey);
    }

    #[test]
    fn config_state_save_fence_uses_mod_revision_for_existing_state_key() {
        let fence = config_state_save_fence(Some(19)).expect("fence");
        assert_eq!(fence, ConfigStateSaveFence::MatchModRevision(19));
    }
}