use std::collections::HashMap;
use std::hash::{Hash, Hasher};
use std::sync::{Arc, OnceLock};
use std::time::Duration;

use pd_proto_grpc::{pdpb, tsopb};
use tokio::sync::Mutex;
use tokio::sync::mpsc;
use tokio::sync::oneshot;
use tokio_stream::wrappers::ReceiverStream;
use tonic::transport::{Channel, Endpoint};
use tonic::{Request, Status, Streaming};
use tracing::warn;

use crate::trace_context::TraceContext;

const FORWARD_DIAL_TIMEOUT: Duration = Duration::from_secs(3);
/// Timeout used for one `GetMinTS` fan-out call to one remote TSO pod.
pub(crate) const GET_MIN_TS_FANOUT_TIMEOUT: Duration = Duration::from_secs(1);
/// Default watchdog timeout for one forwarded legacy proxy batch.
pub(crate) const LEGACY_FORWARD_PROXY_TIMEOUT: Duration = Duration::from_secs(3);
const LEGACY_FORWARD_MAX_MERGE_REQUESTS: usize = 10_000;
const LEGACY_FORWARD_QUEUE_CAPACITY: usize = LEGACY_FORWARD_MAX_MERGE_REQUESTS + 1;
const LEGACY_FORWARD_MAX_BATCH_SIZE: usize = LEGACY_FORWARD_MAX_MERGE_REQUESTS + 1;
/// Compatibility token returned when merged legacy forward count overflows `u32`.
pub(crate) const LEGACY_FORWARD_MERGED_COUNT_OVERFLOW_TOKEN: &str =
    "merged tso request count overflow";
const LEGACY_FORWARD_SPLIT_SEND_FAILED_TOKEN: &str = "legacy forwarding split response send failed";
pub(crate) const LEGACY_PROXY_RECV_FROM_CLIENT_TIMEOUT_TOKEN: &str =
    "tso proxy timeout when receiving from client; stream closed by server";
pub(crate) const LEGACY_PROXY_RECV_FROM_CLIENT_EOF_TOKEN: &str = "EOF";
pub(crate) const LEGACY_PROXY_FORWARD_TIMEOUT_TOKEN: &str = "forward tso request timeout";
pub(crate) const LEGACY_PROXY_STREAM_IDLE_TIMEOUT_TOKEN: &str = "TSOProxyStreamIdleTimeout";
pub(crate) const LEGACY_PROXY_NOT_FOUND_TSO_ADDR_TOKEN: &str = "not found tso address";
pub(crate) const LEGACY_PROXY_MAX_CONCURRENT_STREAMINGS_EXCEEDED_TOKEN: &str =
    "max count of concurrent tso proxy routines exceeded";
/// Compatibility failpoint name used to force send-side watchdog timeout handling.
pub(crate) const LEGACY_FORWARD_FAILPOINT_SEND_TO_TSO_TIMEOUT: &str = "tsoProxySendToTSOTimeout";
/// Compatibility failpoint name used to force recv-side watchdog timeout handling.
pub(crate) const LEGACY_FORWARD_FAILPOINT_RECV_FROM_TSO_TIMEOUT: &str =
    "tsoProxyRecvFromTSOTimeout";

/// Deterministic failpoint selector for legacy forwarded watchdog timeout-path tests.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum LegacyForwardFailpoint {
    SendToTsoTimeout,
    RecvFromTsoTimeout,
}

impl LegacyForwardFailpoint {
    pub(crate) fn parse(value: &str) -> Option<Self> {
        match value {
            LEGACY_FORWARD_FAILPOINT_SEND_TO_TSO_TIMEOUT => Some(Self::SendToTsoTimeout),
            LEGACY_FORWARD_FAILPOINT_RECV_FROM_TSO_TIMEOUT => Some(Self::RecvFromTsoTimeout),
            _ => None,
        }
    }

    fn as_str(self) -> &'static str {
        match self {
            Self::SendToTsoTimeout => LEGACY_FORWARD_FAILPOINT_SEND_TO_TSO_TIMEOUT,
            Self::RecvFromTsoTimeout => LEGACY_FORWARD_FAILPOINT_RECV_FROM_TSO_TIMEOUT,
        }
    }
}

pub(crate) fn legacy_forward_timeout_status() -> Status {
    Status::deadline_exceeded(LEGACY_PROXY_FORWARD_TIMEOUT_TOKEN)
}

/// Reusable channel pool keyed by forwarding endpoint.
#[derive(Clone, Default)]
pub(crate) struct ForwardChannelPool {
    channels: Arc<Mutex<HashMap<String, Channel>>>,
}

impl ForwardChannelPool {
    pub(crate) async fn channel(&self, endpoint: &str) -> Result<Channel, Status> {
        {
            let channels = self.channels.lock().await;
            if let Some(channel) = channels.get(endpoint) {
                return Ok(channel.clone());
            }
        }

        let channel = Endpoint::from_shared(endpoint.to_string())
            .map_err(|err| Status::internal(format!("invalid endpoint: {err}")))?
            .connect_timeout(FORWARD_DIAL_TIMEOUT)
            .connect()
            .await
            .map_err(|err| Status::unavailable(format!("connect failed: {err}")))?;

        let mut channels = self.channels.lock().await;
        let entry = channels
            .entry(endpoint.to_string())
            .or_insert_with(|| channel.clone());
        Ok(entry.clone())
    }
}

pub(crate) struct LegacyForwardDispatchRequest {
    pub(crate) request: pdpb::TsoRequest,
    pub(crate) trace_context: TraceContext,
    pub(crate) failpoint: Option<LegacyForwardFailpoint>,
    pub(crate) response_tx: oneshot::Sender<Result<pdpb::TsoResponse, Status>>,
}

/// Shared-dispatcher pool for legacy forwarded stream path.
#[derive(Clone, Default)]
pub(crate) struct LegacyForwardDispatcherPool {
    queues: Arc<Mutex<HashMap<String, mpsc::Sender<LegacyForwardDispatchRequest>>>>,
    last_errors: Arc<Mutex<HashMap<String, Status>>>,
}

impl LegacyForwardDispatcherPool {
    pub(crate) async fn queue_for_host(
        &self,
        forwarded_host: &str,
        endpoint: &str,
        channel_pool: &ForwardChannelPool,
        idle_timeout: Duration,
    ) -> Result<mpsc::Sender<LegacyForwardDispatchRequest>, Status> {
        loop {
            if let Some(status) = self.take_last_error(forwarded_host).await {
                return Err(status);
            }
            if let Some(existing) = {
                let queues = self.queues.lock().await;
                queues.get(forwarded_host).cloned()
            } {
                if !existing.is_closed() {
                    return Ok(existing);
                }
                self.remove_if_matches(forwarded_host, &existing).await;
                continue;
            }

            let channel = channel_pool.channel(endpoint).await?;
            let (queue_tx, queue_rx) = mpsc::channel(LEGACY_FORWARD_QUEUE_CAPACITY);
            let inserted = {
                let mut queues = self.queues.lock().await;
                if let Some(existing) = queues.get(forwarded_host).cloned() {
                    if !existing.is_closed() {
                        Some(existing)
                    } else {
                        queues.remove(forwarded_host);
                        queues.insert(forwarded_host.to_string(), queue_tx.clone());
                        None
                    }
                } else {
                    queues.insert(forwarded_host.to_string(), queue_tx.clone());
                    None
                }
            };
            if let Some(existing) = inserted {
                return Ok(existing);
            }

            let pool = self.clone();
            let host = forwarded_host.to_string();
            let sender = queue_tx.clone();
            tokio::spawn(async move {
                run_legacy_forward_dispatcher(pool, host, sender, channel, queue_rx, idle_timeout)
                    .await;
            });
            return Ok(queue_tx);
        }
    }

    pub(crate) async fn remove_if_matches(
        &self,
        forwarded_host: &str,
        sender: &mpsc::Sender<LegacyForwardDispatchRequest>,
    ) {
        let mut queues = self.queues.lock().await;
        let should_remove = queues
            .get(forwarded_host)
            .is_some_and(|existing| existing.same_channel(sender));
        if should_remove {
            queues.remove(forwarded_host);
        }
    }

    async fn record_last_error(&self, forwarded_host: &str, status: Status) {
        let mut errors = self.last_errors.lock().await;
        errors.insert(forwarded_host.to_string(), status);
    }

    async fn take_last_error(&self, forwarded_host: &str) -> Option<Status> {
        let mut errors = self.last_errors.lock().await;
        errors.remove(forwarded_host)
    }
}

async fn run_legacy_forward_dispatcher(
    pool: LegacyForwardDispatcherPool,
    forwarded_host: String,
    queue_sender: mpsc::Sender<LegacyForwardDispatchRequest>,
    channel: Channel,
    mut queue_rx: mpsc::Receiver<LegacyForwardDispatchRequest>,
    idle_timeout: Duration,
) {
    let run_result = run_legacy_forward_dispatcher_loop(&mut queue_rx, channel, idle_timeout).await;
    if let Err(status) = run_result {
        warn!(
            forwarded_host = %forwarded_host,
            error = %status,
            "legacy forwarding dispatcher stopped with error"
        );
        pool.record_last_error(&forwarded_host, status).await;
    }
    pool.remove_if_matches(&forwarded_host, &queue_sender).await;
}

async fn run_legacy_forward_dispatcher_loop(
    queue_rx: &mut mpsc::Receiver<LegacyForwardDispatchRequest>,
    channel: Channel,
    idle_timeout: Duration,
) -> Result<(), Status> {
    let mut client = pdpb::pd_client::PdClient::new(channel);
    let mut upstream_tx = None;
    let mut upstream = None;
    let mut upstream_trace_context = TraceContext::default();

    loop {
        let first = match tokio::time::timeout(idle_timeout, queue_rx.recv()).await {
            Ok(Some(request)) => request,
            Ok(None) => break,
            Err(_) => return Err(Status::unknown(LEGACY_PROXY_STREAM_IDLE_TIMEOUT_TOKEN)),
        };
        let first_trace_context = first.trace_context.clone();
        let mut requests = vec![first];
        if upstream_tx.is_none() || first_trace_context != upstream_trace_context {
            match open_legacy_forward_stream(&mut client, &first_trace_context).await {
                Ok((sender, stream)) => {
                    upstream_tx = Some(sender);
                    upstream = Some(stream);
                    upstream_trace_context = first_trace_context;
                }
                Err(status) => {
                    queue_rx.close();
                    fail_dispatch_requests(requests, queue_rx, status.clone());
                    return Err(status);
                }
            }
        }

        while requests.len() < LEGACY_FORWARD_MAX_BATCH_SIZE {
            match queue_rx.try_recv() {
                Ok(next) => requests.push(next),
                Err(tokio::sync::mpsc::error::TryRecvError::Empty) => break,
                Err(tokio::sync::mpsc::error::TryRecvError::Disconnected) => break,
            }
        }

        let responses = match forward_legacy_merged_batch(
            upstream_tx.as_ref().expect("upstream sender"),
            upstream.as_mut().expect("upstream stream"),
            &requests,
        )
        .await
        {
            Ok(responses) => responses,
            Err(status) => {
                queue_rx.close();
                fail_dispatch_requests(requests, queue_rx, status.clone());
                return Err(status);
            }
        };

        let mut request_iter = requests.into_iter();
        let mut response_iter = responses.into_iter();
        while let Some(request) = request_iter.next() {
            let response = response_iter.next().unwrap_or_default();
            if request.response_tx.send(Ok(response)).is_err() {
                let remaining = request_iter.collect::<Vec<_>>();
                let status = Status::unknown(LEGACY_FORWARD_SPLIT_SEND_FAILED_TOKEN);
                queue_rx.close();
                fail_dispatch_requests(remaining, queue_rx, status.clone());
                return Err(status);
            }
        }
    }
    Ok(())
}

async fn open_legacy_forward_stream(
    client: &mut pdpb::pd_client::PdClient<Channel>,
    trace_context: &TraceContext,
) -> Result<(mpsc::Sender<pdpb::TsoRequest>, Streaming<pdpb::TsoResponse>), Status> {
    let (upstream_tx, upstream_rx) =
        mpsc::channel::<pdpb::TsoRequest>(LEGACY_FORWARD_MAX_BATCH_SIZE);
    let mut request = Request::new(ReceiverStream::new(upstream_rx));
    trace_context.inject(request.metadata_mut());
    let upstream = tokio::time::timeout(LEGACY_FORWARD_PROXY_TIMEOUT, client.tso(request))
        .await
        .map_err(|_| legacy_forward_timeout_status())??
        .into_inner();
    Ok((upstream_tx, upstream))
}

async fn forward_legacy_merged_batch(
    upstream_tx: &mpsc::Sender<pdpb::TsoRequest>,
    upstream: &mut Streaming<pdpb::TsoResponse>,
    requests: &[LegacyForwardDispatchRequest],
) -> Result<Vec<pdpb::TsoResponse>, Status> {
    let merged_count = merged_forward_count(requests).map_err(|status| *status)?;
    let first_header = requests
        .first()
        .and_then(|request| request.request.header.clone())
        .unwrap_or_default();
    let merged = pdpb::TsoRequest {
        header: Some(pdpb::RequestHeader {
            cluster_id: first_header.cluster_id,
            sender_id: 0,
            caller_id: String::new(),
            caller_component: String::new(),
        }),
        count: merged_count,
        dc_location: String::new(),
    };
    let failpoint = requests.first().and_then(|request| request.failpoint);
    let merged_response = tokio::time::timeout(LEGACY_FORWARD_PROXY_TIMEOUT, async {
        if failpoint == Some(LegacyForwardFailpoint::SendToTsoTimeout) {
            return Err(Status::cancelled("context canceled"));
        }
        upstream_tx
            .send(merged)
            .await
            .map_err(|_| Status::unavailable("legacy forwarding upstream stream closed"))?;
        if failpoint == Some(LegacyForwardFailpoint::RecvFromTsoTimeout) {
            return Err(Status::cancelled("context canceled"));
        }
        upstream.message().await?.ok_or_else(|| {
            Status::unknown("legacy forwarding upstream stream returned empty response")
        })
    })
    .await
    .map_err(|_| Status::cancelled("context canceled"))??;
    Ok(split_forwarded_responses(
        requests,
        merged_response,
        merged_count,
    ))
}

pub(crate) fn merged_forward_count(
    requests: &[LegacyForwardDispatchRequest],
) -> Result<u32, Box<Status>> {
    let mut merged = 0_u64;
    for request in requests {
        merged = merged.saturating_add(u64::from(request.request.count));
    }
    if merged > u64::from(u32::MAX) {
        return Err(Box::new(Status::unknown(
            LEGACY_FORWARD_MERGED_COUNT_OVERFLOW_TOKEN,
        )));
    }
    Ok(merged as u32)
}

pub(crate) fn split_forwarded_responses(
    requests: &[LegacyForwardDispatchRequest],
    merged_response: pdpb::TsoResponse,
    merged_count: u32,
) -> Vec<pdpb::TsoResponse> {
    let merged_timestamp = merged_response.timestamp.unwrap_or_default();
    let first_logical = merged_timestamp.logical - i64::from(merged_count);

    let mut split_responses = Vec::with_capacity(requests.len());
    let mut count_sum = 0_i64;
    for request in requests {
        count_sum = count_sum.saturating_add(i64::from(request.request.count));
        let response_cluster_id = request
            .request
            .header
            .as_ref()
            .map_or(0, |header| header.cluster_id);
        split_responses.push(pdpb::TsoResponse {
            header: Some(pdpb::ResponseHeader {
                cluster_id: response_cluster_id,
                error: None,
            }),
            count: request.request.count,
            timestamp: Some(pdpb::Timestamp {
                physical: merged_timestamp.physical,
                logical: first_logical.saturating_add(count_sum),
                suffix_bits: merged_timestamp.suffix_bits,
            }),
        });
    }
    split_responses
}

/// Bench helper that exercises merged-count and split-response hot paths for one synthetic batch.
pub fn benchmark_forwarding_merge_and_split(
    counts: &[u32],
) -> Result<Vec<pdpb::TsoResponse>, Box<Status>> {
    let requests = benchmark_dispatch_requests(counts);
    let merged_count = merged_forward_count(&requests)?;
    let merged_response = pdpb::TsoResponse {
        header: Some(pdpb::ResponseHeader {
            cluster_id: 1,
            error: None,
        }),
        count: merged_count,
        timestamp: Some(pdpb::Timestamp {
            physical: 1,
            logical: i64::from(merged_count),
            suffix_bits: 0,
        }),
    };
    Ok(split_forwarded_responses(
        &requests,
        merged_response,
        merged_count,
    ))
}

fn benchmark_dispatch_requests(counts: &[u32]) -> Vec<LegacyForwardDispatchRequest> {
    let mut requests = Vec::with_capacity(counts.len());
    for count in counts {
        let (response_tx, _response_rx) = oneshot::channel();
        requests.push(LegacyForwardDispatchRequest {
            request: pdpb::TsoRequest {
                header: Some(pdpb::RequestHeader {
                    cluster_id: 1,
                    sender_id: 0,
                    caller_id: String::new(),
                    caller_component: String::new(),
                }),
                count: *count,
                dc_location: String::new(),
            },
            trace_context: TraceContext::default(),
            failpoint: None,
            response_tx,
        });
    }
    requests
}

pub(crate) fn dispatch_host_key(
    forwarded_host: &str,
    failpoint: Option<LegacyForwardFailpoint>,
) -> String {
    if let Some(failpoint) = failpoint {
        return format!("{forwarded_host}#{}", failpoint.as_str());
    }
    forwarded_host.to_string()
}

fn fail_dispatch_requests(
    mut current_batch: Vec<LegacyForwardDispatchRequest>,
    queue_rx: &mut mpsc::Receiver<LegacyForwardDispatchRequest>,
    status: Status,
) {
    for request in current_batch.drain(..) {
        let _ = request.response_tx.send(Err(status.clone()));
    }
    while let Ok(request) = queue_rx.try_recv() {
        let _ = request.response_tx.send(Err(status.clone()));
    }
}

/// Legacy forwarding helper that maps `pdpb` requests to `tsopb` transport and back.
pub struct LegacyForwarder {
    client: tsopb::tso_client::TsoClient<Channel>,
}

impl LegacyForwarder {
    fn from_channel(channel: Channel) -> Self {
        Self {
            client: tsopb::tso_client::TsoClient::new(channel),
        }
    }

    /// Connects a new forwarding client to one remote TSO endpoint.
    pub async fn connect(endpoint: String) -> Result<Self, Status> {
        let channel = Endpoint::from_shared(endpoint)
            .map_err(|err| Status::internal(format!("invalid endpoint: {err}")))?
            .connect_timeout(FORWARD_DIAL_TIMEOUT)
            .connect()
            .await
            .map_err(|err| Status::unavailable(format!("connect failed: {err}")))?;
        Ok(Self::from_channel(channel))
    }

    /// Forwards one legacy `pdpb.TsoRequest` through `tsopb.TSO.Tso`.
    pub async fn forward_tso(
        &mut self,
        request: pdpb::TsoRequest,
    ) -> Result<pdpb::TsoResponse, Status> {
        self.forward_tso_with_context(request, &TraceContext::default())
            .await
    }

    /// Forwards one legacy `pdpb.TsoRequest` with explicit trace metadata propagation.
    pub async fn forward_tso_with_context(
        &mut self,
        request: pdpb::TsoRequest,
        trace_context: &TraceContext,
    ) -> Result<pdpb::TsoResponse, Status> {
        let header = request.header.unwrap_or_default();
        let bootstrap_keyspace_id = bootstrap_keyspace_id();
        let tsopb_request = tsopb::TsoRequest {
            header: Some(tsopb::RequestHeader {
                cluster_id: header.cluster_id,
                sender_id: header.sender_id,
                keyspace_id: bootstrap_keyspace_id,
                keyspace_group_id: 0,
            }),
            count: request.count,
            dc_location: String::new(),
        };

        let (tx, rx) = mpsc::channel(1);
        tx.send(tsopb_request)
            .await
            .map_err(|_| Status::internal("failed to enqueue forwarded request"))?;
        drop(tx);
        let mut transport = Request::new(ReceiverStream::new(rx));
        trace_context.inject(transport.metadata_mut());
        let response = self
            .client
            .tso(transport)
            .await?
            .into_inner()
            .message()
            .await?
            .ok_or_else(|| Status::unknown("forwarded tso stream returned empty response"))?;

        let response_header = response.header.unwrap_or_default();
        Ok(pdpb::TsoResponse {
            header: Some(pdpb::ResponseHeader {
                cluster_id: response_header.cluster_id,
                error: map_tsopb_error_to_pdpb(response_header.error),
            }),
            count: response.count,
            timestamp: response.timestamp,
        })
    }

    /// Forwards legacy `pdpb.GetMinTS` through `tsopb.TSO.GetMinTS`.
    pub async fn forward_get_min_ts(
        &mut self,
        request: pdpb::GetMinTsRequest,
    ) -> Result<pdpb::GetMinTsResponse, Status> {
        self.forward_get_min_ts_with_context(request, &TraceContext::default())
            .await
    }

    /// Forwards legacy `pdpb.GetMinTS` with explicit trace metadata propagation.
    pub async fn forward_get_min_ts_with_context(
        &mut self,
        request: pdpb::GetMinTsRequest,
        trace_context: &TraceContext,
    ) -> Result<pdpb::GetMinTsResponse, Status> {
        let header = request.header.unwrap_or_default();
        let bootstrap_keyspace_id = bootstrap_keyspace_id();
        let mut transport = Request::new(tsopb::GetMinTsRequest {
            header: Some(tsopb::RequestHeader {
                cluster_id: header.cluster_id,
                sender_id: header.sender_id,
                keyspace_id: bootstrap_keyspace_id,
                keyspace_group_id: 0,
            }),
            dc_location: String::new(),
        });
        trace_context.inject(transport.metadata_mut());
        let response = self.client.get_min_ts(transport).await?.into_inner();

        let response_header = response.header.unwrap_or_default();
        Ok(pdpb::GetMinTsResponse {
            header: Some(pdpb::ResponseHeader {
                cluster_id: response_header.cluster_id,
                error: map_tsopb_error_to_pdpb(response_header.error),
            }),
            timestamp: response.timestamp,
        })
    }
}

pub(crate) async fn fetch_min_ts_from_host(
    pool: ForwardChannelPool,
    host: String,
    cluster_id: u64,
    sender_id: u64,
    trace_context: &TraceContext,
) -> Result<tsopb::GetMinTsResponse, Status> {
    let endpoint = forwarded_channel_endpoint(&host);
    let channel = pool.channel(&endpoint).await?;
    let mut client = tsopb::tso_client::TsoClient::new(channel);
    let mut request = Request::new(tsopb::GetMinTsRequest {
        header: Some(tsopb::RequestHeader {
            cluster_id,
            sender_id,
            keyspace_id: bootstrap_keyspace_id(),
            keyspace_group_id: 0,
        }),
        dc_location: String::new(),
    });
    trace_context.inject(request.metadata_mut());
    tokio::time::timeout(GET_MIN_TS_FANOUT_TIMEOUT, client.get_min_ts(request))
        .await
        .map_err(|_| Status::deadline_exceeded("GetMinTS fan-out call timed out"))?
        .map(|response| response.into_inner())
}

pub(crate) fn forwarded_channel_endpoint(forwarded_host: &str) -> String {
    if forwarded_host.starts_with("http://") || forwarded_host.starts_with("https://") {
        forwarded_host.to_string()
    } else {
        format!("http://{forwarded_host}")
    }
}

pub(crate) fn normalize_endpoint(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()
}

/// Returns cached bootstrap keyspace id used by legacy-to-api request conversion.
pub(crate) fn bootstrap_keyspace_id() -> u32 {
    static BOOTSTRAP_KEYSPACE_ID: OnceLock<u32> = OnceLock::new();
    *BOOTSTRAP_KEYSPACE_ID.get_or_init(|| {
        let mode = bootstrap_mode_from_env(
            std::env::var("PD_TSO_BOOTSTRAP_MODE").ok(),
            std::env::var("PD_TSO_KERNEL_TYPE").ok(),
        );
        parse_bootstrap_keyspace_id(std::env::var("PD_TSO_BOOTSTRAP_KEYSPACE_ID").ok(), mode)
    })
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum BootstrapMode {
    Classic,
    NextGen,
}

impl BootstrapMode {
    fn default_keyspace_id(self) -> u32 {
        match self {
            Self::Classic => 0,
            // SystemKeyspaceID in upstream PD: 0xFFFFFF - 1.
            Self::NextGen => 16_777_214,
        }
    }
}

pub(crate) fn bootstrap_mode_from_env(
    raw_mode: Option<String>,
    raw_kernel: Option<String>,
) -> BootstrapMode {
    let mode = raw_mode.or(raw_kernel).unwrap_or_default();
    let normalized = mode.trim().to_ascii_lowercase();
    match normalized.as_str() {
        "nextgen" | "next-gen" | "next_generation" | "next generation" => BootstrapMode::NextGen,
        _ => BootstrapMode::Classic,
    }
}

pub(crate) fn parse_bootstrap_keyspace_id(raw: Option<String>, mode: BootstrapMode) -> u32 {
    raw.and_then(|value| value.parse::<u32>().ok())
        .unwrap_or_else(|| mode.default_keyspace_id())
}

pub(crate) fn map_tsopb_error_to_pdpb(error: Option<tsopb::Error>) -> Option<pdpb::Error> {
    let error = error?;
    let typ = tsopb::ErrorType::try_from(error.r#type).unwrap_or(tsopb::ErrorType::Unknown);
    let pd_type = match typ {
        tsopb::ErrorType::Ok => pdpb::ErrorType::Ok,
        tsopb::ErrorType::Unknown
        | tsopb::ErrorType::NotBootstrapped
        | tsopb::ErrorType::AlreadyBootstrapped
        | tsopb::ErrorType::InvalidValue
        | tsopb::ErrorType::ClusterMismatched => pdpb::ErrorType::Unknown,
    };
    Some(pdpb::Error {
        r#type: pd_type as i32,
        message: error.message,
    })
}

pub(crate) fn stable_member_id(value: &str) -> u64 {
    let mut hasher = std::collections::hash_map::DefaultHasher::new();
    value.hash(&mut hasher);
    hasher.finish()
}