use std::{
    collections::{BTreeMap, HashMap},
    sync::{Arc, atomic::Ordering},
};

use either::Either;
use futures::{TryStreamExt, future::ok};
use http::header::{self, HeaderValue};
use hyper::{Body, Request, Response, StatusCode};
use log::{info, warn};
use online_config::OnlineConfig;
use serde::Serialize;
use tici_common::{index::IndexKey, keyspace::NULLSPACEID};
use tici_grpc::KeyRange;

use crate::{
    meta::{
        MetaServiceCore,
        reader::{ShardCacheManager, UnifiedShardInfo},
        writer::{CompactShardReason, CompactionState, CompactionStateKey},
    },
    status_server::util::{decode_json, make_response},
};

#[derive(Clone)]
pub struct MetaOperator {
    meta: Arc<MetaServiceCore>,
}
impl MetaOperator {
    pub fn new(meta: Arc<MetaServiceCore>) -> Self {
        MetaOperator { meta }
    }
}

#[derive(Clone)]
pub enum Operators {
    Meta(MetaOperator),
    None,
}

#[derive(Serialize)]
struct ReaderAllocatedShardJson {
    shard_id: u64,
    index_key: IndexKey,
    epoch: u64,
    seq: u64,
    start_key: String,
    end_key: String,
    is_warming: bool,
}

#[derive(Serialize)]
struct WorkerScheduledShardJson {
    shard_id: u64,
    keyspace_id: u32,
    table_id: i64,
    index_id: i64,
    epoch: u64,
    seq: u64,
    start_key: String,
    end_key: String,
    cdc_s3_last_file: String,
    upload_bucket: String,
    upload_prefix: String,
}

#[derive(Serialize)]
struct RunningCompactionJson {
    key: CompactionStateKey,
    seq: u64,
    epoch: u64,
    start_ts_ms: u64,
    elapsed_secs: f64,
    level: Option<u8>,
    reason: CompactShardReason,
    input_frag_count: usize,
    input_size: u64,
    current_level_states: [u64; 2],
}

impl RunningCompactionJson {
    fn from_state(state: &CompactionState) -> Self {
        Self {
            key: state.get_compaction_state_key(),
            seq: state.seq,
            epoch: state.epoch,
            start_ts_ms: state.get_start_ts(),
            elapsed_secs: state.get_elapsed_secs(),
            level: state.level,
            reason: state.reason.clone(),
            input_frag_count: state.input_fragments.len(),
            input_size: state.get_input_size(),
            current_level_states: state.current_level_states,
        }
    }
}

#[derive(Serialize)]
struct ShardRunningCompactionsJson {
    shard_id: u64,
    start_key: String,
    end_key: String,
    epoch: u64,
    seq: u64,
    writer: String,
    compactions: Vec<RunningCompactionJson>,
}

#[derive(Serialize)]
struct AbortCompactionsShardJson {
    shard_id: u64,
    aborted: usize,
}

#[derive(Serialize)]
struct AbortCompactionsJson {
    aborted: usize,
    shards: BTreeMap<String, AbortCompactionsShardJson>,
}

#[derive(Serialize)]
struct ShardRangeJson {
    start_key: String,
    end_key: String,
}

#[derive(Serialize)]
struct ShardStatsJson {
    compaction_start: u64,
    compaction_finish: u64,
    compaction_abort: u64,
    size: u64,
    fragment_count: u64,
    append_bps: f64,
    write_qps: f64,
    last_heartbeat_ts_ms: u64,
}

#[derive(Serialize)]
struct ShardCompactionStateJson {
    key: CompactionStateKey,
    state: CompactionState,
}

#[derive(Serialize)]
struct ShardBuildProgressJson {
    cdc_s3_last_file: String,
    compaction_state: Vec<ShardCompactionStateJson>,
}

#[derive(Serialize)]
struct ShardInfoJson {
    index_key: IndexKey,
    shard_id: u64,
    range: ShardRangeJson,
    epoch: u64,
    seq: u64,
    manifest: crate::ShardManifest,
    writer: String,
    progress: ShardBuildProgressJson,
    frag_upload_location: tici_grpc::S3Location,
    stats: ShardStatsJson,
}

impl WorkerScheduledShardJson {
    fn from_shard_schedule(schedule: &crate::meta::writer::worker_node::ShardSchedule) -> Self {
        let shard = schedule.get_shard_header();
        let index = schedule.get_index();
        Self {
            shard_id: shard.shard_id,
            keyspace_id: index.keyspace_id(),
            table_id: index.table_id(),
            index_id: index.index_id(),
            epoch: shard.epoch,
            seq: shard.seq,
            start_key: hex::encode(&shard.start_key),
            end_key: hex::encode(&shard.end_key),
            cdc_s3_last_file: schedule.latest_progress.cdc_s3_last_file.clone(),
            upload_bucket: schedule.upload_location.bucket.clone(),
            upload_prefix: schedule.upload_location.prefix.clone(),
        }
    }
}

impl ReaderAllocatedShardJson {
    fn from_unified_shard(shard: &UnifiedShardInfo) -> Self {
        Self {
            shard_id: shard.shard_id,
            index_key: shard.index_key.clone(),
            epoch: shard.epoch,
            seq: shard.seq.load(Ordering::SeqCst),
            start_key: range_bound_to_hex(shard.start_key_ref()),
            end_key: range_bound_to_hex(shard.end_key_ref()),
            is_warming: shard.is_warming(),
        }
    }
}

fn range_bound_to_hex(bound: &crate::RangeBound) -> String {
    match bound {
        crate::RangeBound::LeftMost | crate::RangeBound::RightMost => String::new(),
        crate::RangeBound::Key(k) => hex::encode(k),
    }
}

fn handle_range_to_json(range: &crate::HandleRange) -> ShardRangeJson {
    ShardRangeJson {
        start_key: range_bound_to_hex(range.sortable_start_ref()),
        end_key: range_bound_to_hex(range.sortable_end_ref()),
    }
}

fn encode_allocated_shards(
    allocated: &HashMap<u64, UnifiedShardInfo>,
) -> BTreeMap<String, ReaderAllocatedShardJson> {
    let mut out = BTreeMap::new();
    for (shard_id, shard) in allocated.iter() {
        out.insert(
            shard_id.to_string(),
            ReaderAllocatedShardJson::from_unified_shard(shard),
        );
    }
    out
}

fn json_response<T: Serialize>(value: &T) -> Response<Body> {
    match serde_json::to_vec(value) {
        Ok(body) => {
            let mut resp = Response::new(Body::from(body));
            resp.headers_mut().insert(
                header::CONTENT_TYPE,
                HeaderValue::from_static("application/json"),
            );
            resp
        }
        Err(e) => make_response(
            StatusCode::INTERNAL_SERVER_ERROR,
            format!("failed to encode json response, err={e}"),
        ),
    }
}

impl ShardInfoJson {
    async fn from_shard_info(shard: crate::ShardInfo) -> Self {
        let mut compaction_state = shard
            .progress
            .compaction_state
            .iter()
            .map(|(key, state)| ShardCompactionStateJson {
                key: key.clone(),
                state: state.clone(),
            })
            .collect::<Vec<_>>();
        compaction_state.sort_by(|left, right| {
            left.state
                .get_start_ts()
                .cmp(&right.state.get_start_ts())
                .then_with(|| left.key.get_shard_id().cmp(&right.key.get_shard_id()))
        });

        let write_qps = shard.stats.get_write_qps().await;
        let last_heartbeat_ts_ms = shard
            .stats
            .last_heartbeat_ts
            .load(Ordering::Relaxed)
            .try_into()
            .unwrap_or(u64::MAX);

        Self {
            index_key: shard.index_key,
            shard_id: shard.shard_id,
            range: handle_range_to_json(&shard.range),
            epoch: shard.epoch,
            seq: shard.seq,
            manifest: shard.manifest,
            writer: shard.writer,
            progress: ShardBuildProgressJson {
                cdc_s3_last_file: shard.progress.cdc_s3_last_file,
                compaction_state,
            },
            frag_upload_location: shard.frag_upload_location,
            stats: ShardStatsJson {
                compaction_start: shard.stats.get_compaction_start(),
                compaction_finish: shard.stats.get_compaction_finish(),
                compaction_abort: shard.stats.get_compaction_abort(),
                size: shard.stats.get_size(),
                fragment_count: shard.stats.get_fragment_count(),
                append_bps: shard.stats.get_append_bps(),
                write_qps,
                last_heartbeat_ts_ms,
            },
        }
    }
}

fn make_range_bound(query: &str, is_start: bool) -> Either<hyper::Result<Response<Body>>, Vec<u8>> {
    let query_pairs: HashMap<_, _> = url::form_urlencoded::parse(query.as_bytes()).collect();
    match query_pairs.get(if is_start { "start_key" } else { "end_key" }) {
        Some(s) => {
            let s = s.to_string();
            if let Ok(x) = hex::decode(s) {
                Either::Right(x)
            } else {
                let msg = "Invalid key range";
                info!("{}", msg);
                Either::Left(Ok(make_response(StatusCode::BAD_REQUEST, msg)))
            }
        }
        None => {
            let msg = format!("Key range not provided, query={}", query);
            info!("{}", msg);
            Either::Left(Ok(make_response(StatusCode::BAD_REQUEST, msg)))
        }
    }
}

impl Operators {
    pub fn accept_meta(
        &self,
    ) -> either::Either<hyper::Result<Response<Body>>, Arc<MetaServiceCore>> {
        match &self {
            Operators::Meta(x) => Either::Right(x.meta.clone()),
            _ => {
                let msg = "This operative command only applies for meta service node";
                Either::Left(Ok(make_response(StatusCode::NOT_FOUND, msg)))
            }
        }
    }

    pub fn get_config(&self, req: Request<Body>) -> hyper::Result<Response<Body>> {
        match self.accept_meta() {
            Either::Left(l) => l,
            Either::Right(r) => {
                let mut full = false;
                if let Some(query) = req.uri().query() {
                    let query_pairs: HashMap<_, _> =
                        url::form_urlencoded::parse(query.as_bytes()).collect();
                    full = match query_pairs.get("full") {
                        Some(val) => match val.parse() {
                            Ok(val) => val,
                            Err(err) => {
                                return Ok(make_response(StatusCode::BAD_REQUEST, err.to_string()));
                            }
                        },
                        None => false,
                    };
                }
                let encode_res = if full {
                    // Get all config
                    serde_json::to_string(r.config.load().as_ref())
                } else {
                    // Filter hidden config
                    serde_json::to_string(&r.config.load().get_encoder())
                };
                Ok(match encode_res {
                    Ok(json) => Response::builder()
                        .header(header::CONTENT_TYPE, "application/json")
                        .body(Body::from(json))
                        .unwrap(),
                    Err(_) => {
                        make_response(StatusCode::INTERNAL_SERVER_ERROR, "Internal Server Error")
                    }
                })
            }
        }
    }

    pub async fn update_config(&self, req: Request<Body>) -> hyper::Result<Response<Body>> {
        match self.accept_meta() {
            Either::Left(l) => l,
            Either::Right(r) => {
                let mut body = Vec::new();
                let mut persist = true;
                if let Some(query) = req.uri().query() {
                    let query_pairs: HashMap<_, _> =
                        url::form_urlencoded::parse(query.as_bytes()).collect();
                    persist = match query_pairs.get("persist") {
                        Some(val) => match val.parse() {
                            Ok(val) => val,
                            Err(err) => {
                                return Ok(make_response(StatusCode::BAD_REQUEST, err.to_string()));
                            }
                        },
                        None => true,
                    };
                }
                req.into_body()
                    .try_for_each(|bytes| {
                        body.extend(bytes);
                        ok(())
                    })
                    .await?;
                info!(
                    "received update config req {:?}",
                    std::str::from_utf8(body.as_slice())
                );
                Ok(match decode_json(&body) {
                    Ok(change) => match if persist {
                        // TODO(P2) support persist
                        r.update_config(change)
                    } else {
                        r.update_config(change)
                    } {
                        Err(e) => {
                            info!("failed to update config, root cause={}", e.root_cause());
                            if let Some(e) = e.downcast_ref::<std::io::Error>() {
                                make_response(
                                    StatusCode::INTERNAL_SERVER_ERROR,
                                    format!(
                                        "config changed, but failed to persist change due to err: {:?}",
                                        e
                                    ),
                                )
                            } else {
                                make_response(
                                    StatusCode::INTERNAL_SERVER_ERROR,
                                    format!("failed to update, error: {:?}", e),
                                )
                            }
                        }
                        Ok(_) => {
                            let mut resp = Response::default();
                            *resp.status_mut() = StatusCode::OK;
                            resp
                        }
                    },
                    Err(e) => make_response(
                        StatusCode::INTERNAL_SERVER_ERROR,
                        format!("failed to decode, error: {:?}", e),
                    ),
                })
            }
        }
    }

    pub fn get_index_key(query: &str) -> either::Either<IndexKey, String> {
        let query_pairs: HashMap<_, _> = url::form_urlencoded::parse(query.as_bytes()).collect();
        let keyspace_id = match query_pairs.get("keyspace_id") {
            Some(s) => {
                if let Ok(x) = s.parse::<u32>() {
                    x
                } else {
                    let msg = "Invalid keyspace_id";
                    info!("{}", msg);
                    return either::Either::Right(msg.to_string());
                }
            }
            None => NULLSPACEID,
        };
        let table_id = match query_pairs.get("table_id") {
            Some(s) => {
                if let Ok(x) = s.parse::<i64>() {
                    x
                } else {
                    let msg = "Invalid table_id";
                    info!("{}", msg);
                    return either::Either::Right(msg.to_string());
                }
            }
            None => {
                let msg = "Argument table_id not provided";
                info!("{}", msg);
                return either::Either::Right(msg.to_string());
            }
        };
        let index_id = match query_pairs.get("index_id") {
            Some(s) => {
                if let Ok(x) = s.parse::<i64>() {
                    x
                } else {
                    let msg = "Invalid index_id";
                    info!("{}", msg);
                    return either::Either::Right(msg.to_string());
                }
            }
            None => {
                let msg = "Argument index_id not provided";
                info!("{}", msg);
                return either::Either::Right(msg.to_string());
            }
        };
        either::Either::Left(IndexKey::new(keyspace_id, table_id, index_id))
    }

    pub async fn compact_range(&self, req: Request<Body>) -> hyper::Result<Response<Body>> {
        match self.accept_meta() {
            Either::Left(l) => l,
            Either::Right(r) => {
                let query = req.uri().query().unwrap_or("");
                let query_pairs: HashMap<_, _> =
                    url::form_urlencoded::parse(query.as_bytes()).collect();
                let start = match make_range_bound(query, true) {
                    Either::Left(l) => return l,
                    Either::Right(r) => r,
                };
                let end = match make_range_bound(query, false) {
                    Either::Left(l) => return l,
                    Either::Right(r) => r,
                };
                let range = match query_pairs.get("range") {
                    Some(s) => {
                        let x = s.to_string();
                        if x == "delta" {
                            x
                        } else {
                            let msg = format!("Invalid range: {}, only support delta", x);
                            return Ok(make_response(StatusCode::BAD_REQUEST, msg));
                        }
                    }
                    None => {
                        let msg = "Argument range not provided";
                        return Ok(make_response(StatusCode::BAD_REQUEST, msg));
                    }
                };
                let index_key = match Self::get_index_key(query) {
                    Either::Left(l) => l,
                    Either::Right(r) => return Ok(make_response(StatusCode::BAD_REQUEST, r)),
                };
                match r
                    .shard_task_manager
                    .on_msg_manual_schedule_compaction_by_shard(&index_key, start, end, &range)
                    .await
                {
                    Ok((code, msg)) => {
                        let msg = format!("error_code={:?}, error_msg={}", code, msg);
                        Ok(make_response(StatusCode::OK, msg))
                    }
                    Err(e) => {
                        info!("manual_compact: error={:?}", e.root_cause());
                        Ok(make_response(
                            StatusCode::INTERNAL_SERVER_ERROR,
                            format!("{:?}", e),
                        ))
                    }
                }
            }
        }
    }

    pub async fn compactions(&self, req: Request<Body>) -> hyper::Result<Response<Body>> {
        let meta = match self.accept_meta() {
            Either::Left(l) => return l,
            Either::Right(r) => r,
        };

        let query = req.uri().query().unwrap_or("");
        let index_key = match Self::get_index_key(query) {
            Either::Left(l) => l,
            Either::Right(r) => return Ok(make_response(StatusCode::BAD_REQUEST, r)),
        };
        let start = match make_range_bound(query, true) {
            Either::Left(l) => return l,
            Either::Right(r) => r,
        };
        let end = match make_range_bound(query, false) {
            Either::Left(l) => return l,
            Either::Right(r) => r,
        };

        let index = match meta.get_index(&index_key).await {
            Ok(index) => index,
            Err(e) => {
                info!(
                    "compactions: failed to get index, index_key={:?}, err={:?}",
                    index_key,
                    e.root_cause()
                );
                return Ok(make_response(
                    StatusCode::INTERNAL_SERVER_ERROR,
                    format!("{:?}", e),
                ));
            }
        };

        let shards = index
            .find_overlapping_shards(
                &[KeyRange {
                    start_key: start,
                    end_key: end,
                }],
                false,
            )
            .await;

        let mut out = BTreeMap::<String, ShardRunningCompactionsJson>::new();
        for shard in shards {
            let shard = shard.read().await;

            let mut compactions = shard
                .progress
                .compaction_state
                .values()
                .map(RunningCompactionJson::from_state)
                .collect::<Vec<_>>();
            compactions.sort_by(|a, b| a.start_ts_ms.cmp(&b.start_ts_ms));

            out.insert(
                shard.shard_id.to_string(),
                ShardRunningCompactionsJson {
                    shard_id: shard.shard_id,
                    start_key: range_bound_to_hex(shard.range.sortable_start_ref()),
                    end_key: range_bound_to_hex(shard.range.sortable_end_ref()),
                    epoch: shard.epoch,
                    seq: shard.seq,
                    writer: shard.writer.clone(),
                    compactions,
                },
            );
        }

        Ok(json_response(&out))
    }

    pub async fn shard_info(&self, req: Request<Body>) -> hyper::Result<Response<Body>> {
        let meta = match self.accept_meta() {
            Either::Left(l) => return l,
            Either::Right(r) => r,
        };

        let query = req.uri().query().unwrap_or("");
        let query_pairs: HashMap<_, _> = url::form_urlencoded::parse(query.as_bytes()).collect();
        let shard_id = match query_pairs.get("shard_id") {
            Some(value) => match value.parse::<u64>() {
                Ok(shard_id) => shard_id,
                Err(_) => {
                    return Ok(make_response(StatusCode::BAD_REQUEST, "Invalid shard_id"));
                }
            },
            None => {
                return Ok(make_response(
                    StatusCode::BAD_REQUEST,
                    "shard_id is required",
                ));
            }
        };

        let Some((_, shard)) = meta.shard_task_manager.find_shard_by_id(shard_id).await else {
            return Ok(make_response(
                StatusCode::NOT_FOUND,
                format!("shard_id={} not found", shard_id),
            ));
        };

        let shard = shard.read().await.clone();
        Ok(json_response(&ShardInfoJson::from_shard_info(shard).await))
    }

    pub async fn abort_compactions(&self, req: Request<Body>) -> hyper::Result<Response<Body>> {
        let meta = match self.accept_meta() {
            Either::Left(l) => return l,
            Either::Right(r) => r,
        };

        let query = req.uri().query().unwrap_or("");
        let index_key = match Self::get_index_key(query) {
            Either::Left(l) => l,
            Either::Right(r) => return Ok(make_response(StatusCode::BAD_REQUEST, r)),
        };
        let start = match make_range_bound(query, true) {
            Either::Left(l) => return l,
            Either::Right(r) => r,
        };
        let end = match make_range_bound(query, false) {
            Either::Left(l) => return l,
            Either::Right(r) => r,
        };

        let index = match meta.get_index(&index_key).await {
            Ok(index) => index,
            Err(e) => {
                info!(
                    "abort_compactions: failed to get index, index_key={:?}, err={:?}",
                    index_key,
                    e.root_cause()
                );
                return Ok(make_response(
                    StatusCode::INTERNAL_SERVER_ERROR,
                    format!("{:?}", e),
                ));
            }
        };

        let shards = index
            .find_overlapping_shards(
                &[KeyRange {
                    start_key: start,
                    end_key: end,
                }],
                false,
            )
            .await;

        // Collect keys first (avoid holding shard locks while awaiting on other
        // components).
        let mut targets = Vec::with_capacity(shards.len());
        for shard in shards {
            let (shard_id, header, before, keys) = {
                let guard = shard.read().await;
                (
                    guard.shard_id,
                    crate::meta::writer::ShardTaskHeader::from(&*guard),
                    guard.progress.compaction_state.len(),
                    guard
                        .progress
                        .compaction_state
                        .keys()
                        .cloned()
                        .collect::<Vec<CompactionStateKey>>(),
                )
            };
            targets.push((shard, shard_id, header, before, keys));
        }

        let shard_task_tx = meta.shard_task_manager.get_shard_task_tx();
        let mut abort_receivers = Vec::new();
        for (_, shard_id, header, _, keys) in targets.iter() {
            for key in keys.iter() {
                let worker_addr = key.get_worker();
                if let Some(worker) = meta
                    .worker_node_mgr
                    .get_worker_node(worker_addr.as_str())
                    .await
                {
                    let _ = worker.remove_compaction(key).await;
                }

                let (ctx, rx) =
                    crate::meta::writer::ShardTaskCtx::create(Some("compaction_abort".to_string()));
                if let Err(e) =
                    shard_task_tx.send(crate::meta::writer::ShardTask::CompactionAborted(
                        crate::meta::writer::CompactionAbortTask {
                            ctx,
                            header: header.clone(),
                            key: key.clone(),
                        },
                    ))
                {
                    warn!(
                        "abort_compactions: failed to send shard task, shard_id={}, key={:?}, err={:?}",
                        shard_id, key, e
                    );
                    return Ok(make_response(
                        StatusCode::INTERNAL_SERVER_ERROR,
                        format!("failed to send abort task for shard_id={}", shard_id),
                    ));
                }
                abort_receivers.push((*shard_id, key.clone(), rx));
            }
        }

        for (shard_id, key, rx) in abort_receivers {
            match rx.await {
                Ok(Ok(())) => {}
                Ok(Err(e)) => {
                    warn!(
                        "abort_compactions: shard task failed, shard_id={}, key={:?}, err={:?}",
                        shard_id, key, e
                    );
                }
                Err(e) => {
                    warn!(
                        "abort_compactions: shard task callback dropped, shard_id={}, key={:?}, err={:?}",
                        shard_id, key, e
                    );
                }
            }
        }

        let mut total_aborted = 0usize;
        let mut per_shard = BTreeMap::<String, AbortCompactionsShardJson>::new();

        for (shard, shard_id, _, before, _) in targets {
            let after = shard.read().await.progress.compaction_state.len();
            let aborted = before.saturating_sub(after);

            total_aborted += aborted;
            per_shard.insert(
                shard_id.to_string(),
                AbortCompactionsShardJson { shard_id, aborted },
            );
        }

        Ok(json_response(&AbortCompactionsJson {
            aborted: total_aborted,
            shards: per_shard,
        }))
    }

    fn parse_block_addrs(query: Option<&str>, _: &[u8]) -> Result<Vec<String>, String> {
        let mut addrs: Vec<String> = vec![];
        if let Some(query) = query {
            for (k, v) in url::form_urlencoded::parse(query.as_bytes()) {
                if k == "addr" {
                    addrs.push(v.into_owned());
                } else if k == "addrs" {
                    addrs.extend(v.split(',').map(|s| s.to_string()));
                }
            }
        }

        let mut seen = std::collections::HashSet::<String>::new();
        for addr in addrs {
            let addr = addr.trim();
            if addr.is_empty() {
                continue;
            }
            seen.insert(addr.to_string());
        }
        Ok(seen.into_iter().collect())
    }

    pub async fn block_worker(&self, req: Request<Body>) -> hyper::Result<Response<Body>> {
        let meta = match self.accept_meta() {
            Either::Left(l) => return l,
            Either::Right(r) => r,
        };

        let query = req.uri().query().map(|s| s.to_string());
        let body = hyper::body::to_bytes(req.into_body()).await?;
        let addrs = match Self::parse_block_addrs(query.as_deref(), body.as_ref()) {
            Ok(addrs) => addrs,
            Err(msg) => return Ok(make_response(StatusCode::BAD_REQUEST, msg)),
        };
        if addrs.is_empty() {
            return Ok(make_response(
                StatusCode::BAD_REQUEST,
                "no addr provided (use query `addr`/`addrs` or json body)",
            ));
        }

        meta.worker_node_mgr.block_workers(&addrs).await;

        Ok(make_response(
            StatusCode::OK,
            format!("blocked_worker_addrs={}", addrs.join(",")),
        ))
    }

    pub async fn unblock_worker(&self, req: Request<Body>) -> hyper::Result<Response<Body>> {
        let meta = match self.accept_meta() {
            Either::Left(l) => return l,
            Either::Right(r) => r,
        };

        let query = req.uri().query().map(|s| s.to_string());
        let body = hyper::body::to_bytes(req.into_body()).await?;
        let addrs = match Self::parse_block_addrs(query.as_deref(), body.as_ref()) {
            Ok(addrs) => addrs,
            Err(msg) => return Ok(make_response(StatusCode::BAD_REQUEST, msg)),
        };
        if addrs.is_empty() {
            return Ok(make_response(
                StatusCode::BAD_REQUEST,
                "no addr provided (use query `addr`/`addrs` or json body)",
            ));
        }

        meta.worker_node_mgr.unblock_workers(&addrs).await;

        Ok(make_response(
            StatusCode::OK,
            format!("unblocked_worker_addrs={}", addrs.join(",")),
        ))
    }

    pub async fn list_blocked_workers(&self, _req: Request<Body>) -> hyper::Result<Response<Body>> {
        let meta = match self.accept_meta() {
            Either::Left(l) => return l,
            Either::Right(r) => r,
        };

        let blocked = meta.worker_node_mgr.list_blocked_workers().await;
        Ok(json_response(&blocked))
    }

    pub async fn block_reader(&self, req: Request<Body>) -> hyper::Result<Response<Body>> {
        let meta = match self.accept_meta() {
            Either::Left(l) => return l,
            Either::Right(r) => r,
        };

        let query = req.uri().query().map(|s| s.to_string());
        let body = hyper::body::to_bytes(req.into_body()).await?;
        let addrs = match Self::parse_block_addrs(query.as_deref(), body.as_ref()) {
            Ok(addrs) => addrs,
            Err(msg) => return Ok(make_response(StatusCode::BAD_REQUEST, msg)),
        };
        if addrs.is_empty() {
            return Ok(make_response(
                StatusCode::BAD_REQUEST,
                "no addr provided (use query `addr`/`addrs` or json body)",
            ));
        }

        meta.get_reader_pool().block_readers(&addrs).await;

        Ok(make_response(
            StatusCode::OK,
            format!("blocked_reader_addrs={}", addrs.join(",")),
        ))
    }

    pub async fn unblock_reader(&self, req: Request<Body>) -> hyper::Result<Response<Body>> {
        let meta = match self.accept_meta() {
            Either::Left(l) => return l,
            Either::Right(r) => r,
        };

        let query = req.uri().query().map(|s| s.to_string());
        let body = hyper::body::to_bytes(req.into_body()).await?;
        let addrs = match Self::parse_block_addrs(query.as_deref(), body.as_ref()) {
            Ok(addrs) => addrs,
            Err(msg) => return Ok(make_response(StatusCode::BAD_REQUEST, msg)),
        };
        if addrs.is_empty() {
            return Ok(make_response(
                StatusCode::BAD_REQUEST,
                "no addr provided (use query `addr`/`addrs` or json body)",
            ));
        }

        meta.get_reader_pool().unblock_readers(&addrs).await;

        Ok(make_response(
            StatusCode::OK,
            format!("unblocked_reader_addrs={}", addrs.join(",")),
        ))
    }

    pub async fn list_blocked_readers(&self, _req: Request<Body>) -> hyper::Result<Response<Body>> {
        let meta = match self.accept_meta() {
            Either::Left(l) => return l,
            Either::Right(r) => r,
        };

        let blocked = meta.get_reader_pool().list_blocked_readers().await;
        Ok(json_response(&blocked))
    }

    pub async fn reschedule_range(&self, req: Request<Body>) -> hyper::Result<Response<Body>> {
        match self.accept_meta() {
            Either::Left(l) => l,
            Either::Right(r) => {
                let query = req.uri().query().unwrap_or("");
                let query_pairs: HashMap<_, _> =
                    url::form_urlencoded::parse(query.as_bytes()).collect();
                let start = match make_range_bound(query, true) {
                    Either::Left(l) => return l,
                    Either::Right(r) => r,
                };
                let end = match make_range_bound(query, false) {
                    Either::Left(l) => return l,
                    Either::Right(r) => r,
                };
                let index_key = match Self::get_index_key(query) {
                    Either::Left(l) => l,
                    Either::Right(r) => return Ok(make_response(StatusCode::BAD_REQUEST, r)),
                };
                let writer: Option<String> = query_pairs.get("writer").map(|s| s.to_string());
                match r
                    .shard_task_manager
                    .on_msg_manual_reschedule_by_shard(&index_key, start, end, writer)
                    .await
                {
                    Ok((code, msg)) => {
                        let msg = format!("error_code={:?}, error_msg={}", code, msg);
                        info!("{}", msg);
                        Ok(make_response(StatusCode::OK, msg))
                    }
                    Err(e) => {
                        info!("manual_reschedule: error={:?}", e.root_cause());
                        Ok(make_response(
                            StatusCode::INTERNAL_SERVER_ERROR,
                            format!("{:?}", e),
                        ))
                    }
                }
            }
        }
    }

    pub async fn split_shard(&self, req: Request<Body>) -> hyper::Result<Response<Body>> {
        match self.accept_meta() {
            Either::Left(l) => l,
            Either::Right(r) => {
                let query = req.uri().query().unwrap_or("");
                let index_key = match Self::get_index_key(query) {
                    Either::Left(l) => l,
                    Either::Right(r) => return Ok(make_response(StatusCode::BAD_REQUEST, r)),
                };
                let start = match make_range_bound(query, true) {
                    Either::Left(l) => return l,
                    Either::Right(r) => r,
                };
                match r
                    .shard_task_manager
                    .on_msg_manual_split_by_shard(&index_key, start)
                    .await
                {
                    Ok((code, msg)) => {
                        let msg = format!("error_code={:?}, error_msg={}", code, msg);
                        info!("{}", msg);
                        Ok(make_response(StatusCode::OK, msg))
                    }
                    Err(e) => {
                        info!("manual_split: error={:?}", e.root_cause());
                        Ok(make_response(
                            StatusCode::INTERNAL_SERVER_ERROR,
                            format!("{:?}", e),
                        ))
                    }
                }
            }
        }
    }

    pub async fn reader_reschedule(&self, req: Request<Body>) -> hyper::Result<Response<Body>> {
        match self.accept_meta() {
            Either::Left(l) => l,
            Either::Right(r) => {
                let query = req.uri().query().unwrap_or("");
                let index_key = match Self::get_index_key(query) {
                    Either::Left(l) => l,
                    Either::Right(r) => return Ok(make_response(StatusCode::BAD_REQUEST, r)),
                };
                let start = match make_range_bound(query, true) {
                    Either::Left(l) => return l,
                    Either::Right(r) => r,
                };
                let end = match make_range_bound(query, false) {
                    Either::Left(l) => return l,
                    Either::Right(r) => r,
                };

                match ShardCacheManager::on_msg_manual_reschedule_by_shard(
                    &r.shard_cache_manager,
                    r.get_reader_pool(),
                    &index_key,
                    start,
                    end,
                )
                .await
                {
                    Ok((code, msg)) => {
                        let msg = format!("error_code={:?}, error_msg={}", code, msg);
                        info!("{}", msg);
                        Ok(make_response(StatusCode::OK, msg))
                    }
                    Err(e) => {
                        info!("manual_split: error={:?}", e.root_cause());
                        Ok(make_response(
                            StatusCode::INTERNAL_SERVER_ERROR,
                            format!("{:?}", e),
                        ))
                    }
                }
            }
        }
    }

    pub async fn reader_allocation_add(&self, req: Request<Body>) -> hyper::Result<Response<Body>> {
        match self.accept_meta() {
            Either::Left(l) => l,
            Either::Right(r) => {
                let query = req.uri().query().unwrap_or("");
                let query_pairs: HashMap<_, _> =
                    url::form_urlencoded::parse(query.as_bytes()).collect();
                let addr = query_pairs
                    .get("addr")
                    .map(|s| s.to_string())
                    .filter(|s| !s.is_empty());
                let index_key = match Self::get_index_key(query) {
                    Either::Left(l) => l,
                    Either::Right(r) => return Ok(make_response(StatusCode::BAD_REQUEST, r)),
                };
                let start = match make_range_bound(query, true) {
                    Either::Left(l) => return l,
                    Either::Right(r) => r,
                };
                let end = match make_range_bound(query, false) {
                    Either::Left(l) => return l,
                    Either::Right(r) => r,
                };

                match ShardCacheManager::on_msg_manual_add_replica_by_shard(
                    &r.shard_cache_manager,
                    r.get_reader_pool(),
                    &index_key,
                    start,
                    end,
                    addr,
                )
                .await
                {
                    Ok((code, msg)) => {
                        let msg = format!("error_code={:?}, error_msg={}", code, msg);
                        info!("{}", msg);
                        Ok(make_response(StatusCode::OK, msg))
                    }
                    Err(e) => {
                        info!("manual_add_replica: error={:?}", e.root_cause());
                        Ok(make_response(
                            StatusCode::INTERNAL_SERVER_ERROR,
                            format!("{:?}", e),
                        ))
                    }
                }
            }
        }
    }

    /// Remove a shard replica from a specific reader. Requires addr and
    /// shard_id.
    pub async fn reader_allocation_remove(
        &self,
        req: Request<Body>,
    ) -> hyper::Result<Response<Body>> {
        match self.accept_meta() {
            Either::Left(l) => l,
            Either::Right(r) => {
                let query = req.uri().query().unwrap_or("");
                let query_pairs: HashMap<_, _> =
                    url::form_urlencoded::parse(query.as_bytes()).collect();
                let addr = query_pairs
                    .get("addr")
                    .map(|s| s.to_string())
                    .filter(|s| !s.is_empty());
                let Some(addr) = addr else {
                    return Ok(make_response(
                        StatusCode::BAD_REQUEST,
                        "addr is required for remove",
                    ));
                };
                let shard_id = match query_pairs.get("shard_id") {
                    Some(s) => match s.parse::<u64>() {
                        Ok(id) => id,
                        Err(_) => {
                            return Ok(make_response(StatusCode::BAD_REQUEST, "Invalid shard_id"));
                        }
                    },
                    None => {
                        return Ok(make_response(
                            StatusCode::BAD_REQUEST,
                            "shard_id is required for remove",
                        ));
                    }
                };

                match ShardCacheManager::on_msg_manual_remove_replica_by_shard(
                    &r.shard_cache_manager,
                    r.get_reader_pool(),
                    shard_id,
                    addr,
                )
                .await
                {
                    Ok((code, msg)) => {
                        let msg = format!("error_code={:?}, error_msg={}", code, msg);
                        info!("{}", msg);
                        Ok(make_response(StatusCode::OK, msg))
                    }
                    Err(e) => {
                        info!("manual_remove_replica: error={:?}", e.root_cause());
                        Ok(make_response(
                            StatusCode::INTERNAL_SERVER_ERROR,
                            format!("{:?}", e),
                        ))
                    }
                }
            }
        }
    }

    pub async fn reader_allocation_by_node(
        &self,
        req: Request<Body>,
    ) -> hyper::Result<Response<Body>> {
        let meta = match self.accept_meta() {
            Either::Left(l) => return l,
            Either::Right(r) => r,
        };

        let query = req.uri().query().unwrap_or("");
        let query_pairs: HashMap<_, _> = url::form_urlencoded::parse(query.as_bytes()).collect();
        let addr = query_pairs
            .get("addr")
            .map(|s| s.to_string())
            .filter(|s| !s.is_empty());

        let reader_pool = meta.get_reader_pool();

        if let Some(addr) = addr {
            let reader = reader_pool
                .get_all_readers()
                .await
                .into_iter()
                .find(|r| r.addr == addr);

            let allocated = if let Some(reader) = reader {
                let allocated = reader.allocated_shards_snapshot().await;
                encode_allocated_shards(&allocated)
            } else {
                BTreeMap::new()
            };

            return Ok(json_response(&allocated));
        }

        let mut all = BTreeMap::<String, BTreeMap<String, ReaderAllocatedShardJson>>::new();
        for reader in reader_pool.get_all_readers().await {
            let allocated = reader.allocated_shards_snapshot().await;
            all.insert(reader.addr.clone(), encode_allocated_shards(&allocated));
        }
        Ok(json_response(&all))
    }

    pub async fn worker_allocation_shards(
        &self,
        req: Request<Body>,
    ) -> hyper::Result<Response<Body>> {
        let meta = match self.accept_meta() {
            Either::Left(l) => return l,
            Either::Right(r) => r,
        };

        let query = req.uri().query().unwrap_or("");
        let query_pairs: HashMap<_, _> = url::form_urlencoded::parse(query.as_bytes()).collect();
        let addr = query_pairs
            .get("addr")
            .map(|s| s.to_string())
            .filter(|s| !s.is_empty());

        if let Some(addr) = addr {
            let worker = meta.worker_node_mgr.get_worker_node(&addr).await;
            if let Some(worker) = worker {
                let mut shards = BTreeMap::new();
                for (shard_id, schedule) in worker.collect_shard_schedules().await {
                    shards.insert(
                        shard_id.to_string(),
                        WorkerScheduledShardJson::from_shard_schedule(&schedule),
                    );
                }

                let mut out = BTreeMap::new();
                out.insert(addr, shards);
                return Ok(json_response(&out));
            }

            let out: BTreeMap<String, BTreeMap<String, WorkerScheduledShardJson>> = BTreeMap::new();
            return Ok(json_response(&out));
        }

        let mut all = BTreeMap::<String, BTreeMap<String, WorkerScheduledShardJson>>::new();
        for (addr, worker) in meta.worker_node_mgr.collect_worker_nodes().await {
            let mut shards = BTreeMap::new();
            for (shard_id, schedule) in worker.collect_shard_schedules().await {
                shards.insert(
                    shard_id.to_string(),
                    WorkerScheduledShardJson::from_shard_schedule(&schedule),
                );
            }
            all.insert(addr, shards);
        }
        Ok(json_response(&all))
    }

    pub async fn inactive_import_tasks(&self, req: Request<Body>) -> hyper::Result<Response<Body>> {
        let meta = match self.accept_meta() {
            Either::Left(l) => return l,
            Either::Right(r) => r,
        };

        // try to get query params "seconds" or "reset" from req
        let query = req.uri().query().unwrap_or("");
        let query_pairs: HashMap<_, _> = url::form_urlencoded::parse(query.as_bytes()).collect();
        let inactive_secs = match query_pairs.get("seconds") {
            Some(s) => {
                if let Ok(x) = s.parse::<u64>() {
                    x
                } else {
                    let msg = format!("Invalid parameter: seconds={}", s);
                    return Ok(make_response(StatusCode::BAD_REQUEST, msg));
                }
            }
            None => 3600, // default to 1 hour
        };
        let reset = match query_pairs.get("reset") {
            Some(s) => {
                if s.is_empty() {
                    true
                } else if let Ok(x) = s.parse::<bool>() {
                    x
                } else {
                    let msg = format!("Invalid parameter: reset={}", s);
                    return Ok(make_response(StatusCode::BAD_REQUEST, msg));
                }
            }
            None => false, // default to false
        };
        let n_limit = match query_pairs.get("limit") {
            Some(s) => {
                if let Ok(x) = s.parse::<usize>() {
                    x
                } else {
                    let msg = format!("Invalid parameter: limit={}", s);
                    return Ok(make_response(StatusCode::BAD_REQUEST, msg));
                }
            }
            None => 100, // default to 100
        };

        if !reset {
            info!(
                "Fetching inactive import tasks, inactive_secs={}",
                inactive_secs
            );
            match meta.meta_store.inactive_import_tasks(inactive_secs).await {
                Ok(tasks) => {
                    let json = serde_json::to_string(&tasks).unwrap_or_default();
                    Ok(make_response(StatusCode::OK, json))
                }
                Err(e) => {
                    let msg = format!("failed to get inactive import tasks, err={:?}", e);
                    info!("{}", msg);
                    Ok(make_response(StatusCode::INTERNAL_SERVER_ERROR, msg))
                }
            }
        } else {
            info!(
                "Resetting inactive import tasks to init state, inactive_secs={}, limit={}",
                inactive_secs, n_limit
            );
            match meta
                .meta_store
                .reset_inactive_import_tasks(inactive_secs, n_limit)
                .await
            {
                Ok(tasks) => {
                    info!("reset inactive import tasks, num_tasks={}", tasks.len());
                    let json = serde_json::to_string(&tasks).unwrap_or_default();
                    Ok(make_response(StatusCode::OK, json))
                }
                Err(e) => {
                    let msg = format!("failed to reset inactive import tasks, err={:?}", e);
                    info!("{}", msg);
                    Ok(make_response(StatusCode::INTERNAL_SERVER_ERROR, msg))
                }
            }
        }
    }

    pub async fn manual_gc(&self, req: Request<Body>) -> hyper::Result<Response<Body>> {
        match self.accept_meta() {
            Either::Left(l) => l,
            Either::Right(r) => {
                let query = req.uri().query().unwrap_or("");
                let query_pairs: HashMap<_, _> =
                    url::form_urlencoded::parse(query.as_bytes()).collect();
                let cdc_res = if let Some(val) = query_pairs.get("cdc") {
                    if let Ok(val) = val.parse()
                        && val
                    {
                        let (s, f) = r.shard_task_manager.gc_cdc_files().await;
                        format!("cdc: ({},{})", s, f)
                    } else {
                        "".to_string()
                    }
                } else {
                    "".to_string()
                };
                let frag_res = if let Some(val) = query_pairs.get("frags") {
                    if let Ok(val) = val.parse()
                        && val
                    {
                        match r.shard_task_manager.manual_gc().await {
                            Ok(vec) => {
                                format!("frags: {}", vec.len())
                            }
                            Err(e) => {
                                info!("manual_gc(frags): error={:?}", e.root_cause());
                                return Ok(make_response(
                                    StatusCode::INTERNAL_SERVER_ERROR,
                                    format!("{:?}", e),
                                ));
                            }
                        }
                    } else {
                        "".to_string()
                    }
                } else {
                    "".to_string()
                };
                Ok(make_response(
                    StatusCode::OK,
                    format!("{} {}", cdc_res, frag_res),
                ))
            }
        }
    }
}