use std::sync::Arc;
use std::sync::atomic::Ordering;

use axum::Json;
use axum::extract::{Path, Query, Request, State};
use axum::http::{HeaderMap, StatusCode};
use axum::middleware::Next;
use axum::response::IntoResponse;
use base64::Engine as _;
use bytes::Bytes;
use subtle::ConstantTimeEq;

use crate::codec::frame::{ErrorCode, HeaderLimits, validate_headers};
use crate::namespace::NamespaceManager;
use crate::server::state::{ServerLifecycle, ServerState};
use crate::types::{
    Header, Headers, MessageEnvelope, MessageId, NamespaceId, Permission, TopicName, TopicOffset,
    millis_to_rfc3339, now_millis,
};

use crate::tls::TlsProviders;

use super::AppState;
use super::error::HttpError;
use super::middleware::{authenticate, extract_bearer_token, verify_namespace};

/// Axum middleware that rejects requests with 503 when the server is not initialized.
///
/// Applied to all `/v1/*` business routes so that individual handlers don't need
/// to check initialization status.
pub async fn require_initialized(
    State(lifecycle): State<Arc<ServerLifecycle>>,
    request: Request,
    next: Next,
) -> impl IntoResponse {
    if !lifecycle.is_initialized() {
        return (
            StatusCode::SERVICE_UNAVAILABLE,
            Json(serde_json::json!({
                "error": {
                    "code": 0x0009,
                    "name": "INTERNAL_ERROR",
                    "message": "server initializing"
                }
            })),
        )
            .into_response();
    }
    next.run(request).await
}

/// Axum middleware that gates `/admin/*` routes behind the admin API key.
///
/// Returns 403 if the key is not configured or does not match the Bearer token.
/// Uses constant-time comparison to prevent timing attacks.
pub async fn require_admin_auth<M: NamespaceManager + 'static>(
    State(state): State<AppState<M>>,
    request: Request,
    next: Next,
) -> axum::response::Response {
    let expected = match &state.admin_api_key {
        Some(key) => key,
        None => {
            return (
                StatusCode::FORBIDDEN,
                Json(serde_json::json!({
                    "error": "admin API key not configured"
                })),
            )
                .into_response();
        }
    };

    let token = match extract_bearer_token(request.headers()) {
        Ok(key) => key,
        Err(_) => {
            return (
                StatusCode::FORBIDDEN,
                Json(serde_json::json!({
                    "error": "invalid admin API key"
                })),
            )
                .into_response();
        }
    };

    if token.0.as_bytes().ct_eq(expected.as_bytes()).into() {
        next.run(request).await
    } else {
        (
            StatusCode::FORBIDDEN,
            Json(serde_json::json!({
                "error": "invalid admin API key"
            })),
        )
            .into_response()
    }
}

/// `GET /health/live` — liveness probe, always 200 while the process is running.
pub async fn health_live() -> impl IntoResponse {
    Json(serde_json::json!({ "status": "alive" }))
}

/// `GET /health/ready` and `GET /health` — readiness probe.
///
/// Returns 200 when initialized and running, 503 otherwise.
pub async fn health_ready<M: NamespaceManager + 'static>(
    State(state): State<AppState<M>>,
) -> impl IntoResponse {
    let lifecycle = &state.lifecycle;

    if lifecycle.is_ready() {
        return (
            StatusCode::OK,
            Json(serde_json::json!({ "status": "ready" })),
        );
    }

    let status_str = match lifecycle.state() {
        ServerState::Running => "initializing", // Running but not initialized
        ServerState::Draining => "draining",
        ServerState::ShuttingDown => "shutting_down",
    };

    (
        StatusCode::SERVICE_UNAVAILABLE,
        Json(serde_json::json!({ "status": status_str })),
    )
}

/// `GET /v1/metrics` — Prometheus-compatible metrics endpoint, no auth required.
pub async fn metrics<M: NamespaceManager + 'static>(
    State(state): State<AppState<M>>,
) -> impl IntoResponse {
    match &state.prom_handle {
        Some(handle) => {
            let body = handle.render();
            (
                StatusCode::OK,
                [("content-type", "text/plain; version=0.0.4; charset=utf-8")],
                body,
            )
        }
        None => (
            StatusCode::NOT_FOUND,
            [("content-type", "text/plain; charset=utf-8")],
            "metrics disabled".to_string(),
        ),
    }
}

/// Query parameters for the publish endpoint.
#[derive(Debug, serde::Deserialize)]
pub struct PublishParams {
    /// TTL in seconds. Converted to milliseconds internally.
    pub ttl: Option<u64>,
}

/// `POST /v1/{namespace}/{topic}/publish` — publish a message.
///
/// Body is raw bytes (the message payload). Returns JSON with the assigned
/// offset and message ID. Optional `?ttl=<seconds>` query parameter sets
/// per-message TTL (overrides namespace default).
pub async fn publish<M: NamespaceManager + 'static>(
    State(state): State<AppState<M>>,
    Path((namespace, topic)): Path<(String, String)>,
    Query(params): Query<PublishParams>,
    headers: HeaderMap,
    body: Bytes,
) -> Result<impl IntoResponse, HttpError> {
    let session = authenticate(&*state.auth, &headers)?;
    let ns = NamespaceId(namespace);
    verify_namespace(&session, &ns)?;
    session.permissions.check(Permission::Publish)?;

    let topic = TopicName(topic);

    // Extract X-Fafb-* HTTP headers as message headers.
    let msg_headers = extract_message_headers(&headers);
    let cfg = state.shared_config.load();
    if !msg_headers.is_empty() {
        let limits = HeaderLimits {
            max_count: cfg.max_headers_per_message,
            max_key_bytes: cfg.max_header_key_bytes,
            max_value_bytes: cfg.max_header_value_bytes,
            max_total_bytes: cfg.max_header_bytes_total,
        };
        if let Err(e) = validate_headers(&msg_headers, &limits) {
            return Err(HttpError::new(
                StatusCode::BAD_REQUEST,
                ErrorCode::InvalidFrame,
                e.to_string(),
            ));
        }
        metrics::histogram!("fafb_message_header_count").record(msg_headers.len() as f64);
        metrics::histogram!("fafb_message_header_bytes").record(
            msg_headers
                .iter()
                .map(|h| h.key.len() + h.value.len())
                .sum::<usize>() as f64,
        );
    } else {
        metrics::histogram!("fafb_message_header_count").record(0.0);
        metrics::histogram!("fafb_message_header_bytes").record(0.0);
    }

    // Resolve TTL: explicit query param > namespace default > 0 (no expiry).
    let ttl_ms = {
        let quota = cfg.namespace_quotas.get(&ns);
        let default_ttl_ms = quota.map_or(0, |q| q.default_ttl_secs.saturating_mul(1000));
        let max_ttl_ms = quota.map_or(0, |q| q.max_ttl_secs.saturating_mul(1000));
        let min_ttl_ms = quota.map_or(0, |q| q.min_ttl_secs.saturating_mul(1000));

        let effective = if let Some(ttl_secs) = params.ttl {
            ttl_secs.saturating_mul(1000)
        } else if default_ttl_ms > 0 {
            default_ttl_ms
        } else {
            0
        };

        if effective > 0 {
            if min_ttl_ms > 0 && effective < min_ttl_ms {
                return Err(HttpError::new(
                    StatusCode::BAD_REQUEST,
                    ErrorCode::InvalidFrame,
                    format!("TTL {}ms is below minimum {}ms", effective, min_ttl_ms),
                ));
            }
            if max_ttl_ms > 0 && effective > max_ttl_ms {
                tracing::warn!(
                    namespace = %ns, topic = %topic,
                    ttl_ms = effective, max_ttl_ms = max_ttl_ms,
                    "HTTP TTL clamped to namespace maximum"
                );
                max_ttl_ms as u32
            } else {
                effective as u32
            }
        } else {
            0u32
        }
    };

    let msg_id = state.message_id_counter.fetch_add(1, Ordering::Relaxed);
    let ts = now_millis();
    let envelope = MessageEnvelope::new(
        MessageId::new(msg_id),
        ns.clone(),
        topic.clone(),
        body.to_vec(),
        ts,
    )
    .with_headers(msg_headers)
    .with_ttl(ttl_ms);

    // Record TTL metric (always, so 0-TTL messages appear in distribution)
    metrics::histogram!("fafb_message_ttl_seconds").record(ttl_ms as f64 / 1000.0);

    let offset = match state.manager.publish(&ns, &topic, envelope).await {
        Ok(o) => o,
        Err(e) => {
            tracing::warn!(
                namespace = %ns, topic = %topic,
                error = %e,
                "HTTP publish failed"
            );
            return Err(e.into());
        }
    };

    Ok(Json(serde_json::json!({
        "offset": offset.0,
        "message_id": msg_id,
        "timestamp_millis": ts,
    })))
}

/// `GET /v1/{namespace}/{topic}/info` — queue depth, subscriber count, etc.
///
/// `subscriber_count` and `consumer_group_count` reflect TCP subscribers only.
/// HTTP clients do not maintain persistent subscriptions, so these fields will
/// be zero in an HTTP-only deployment.
pub async fn topic_info<M: NamespaceManager + 'static>(
    State(state): State<AppState<M>>,
    Path((namespace, topic)): Path<(String, String)>,
    headers: HeaderMap,
) -> Result<impl IntoResponse, HttpError> {
    let session = authenticate(&*state.auth, &headers)?;
    let ns = NamespaceId(namespace);
    verify_namespace(&session, &ns)?;
    session.permissions.check(Permission::Query)?;

    let topic = TopicName(topic);
    let info = match state.manager.topic_info(&ns, &topic).await {
        Ok(i) => i,
        Err(e) => {
            tracing::debug!(
                namespace = %ns, topic = %topic,
                error = %e,
                "HTTP topic_info query failed"
            );
            return Err(e.into());
        }
    };

    Ok(Json(serde_json::json!({
        "namespace": ns.0,
        "topic": topic.0,
        "depth": info.depth,
        "bytes_used": info.bytes_used,
        "base_offset": info.base_offset.0,
        "head_offset": info.head_offset.0,
        "subscriber_count": info.subscriber_count,
        "consumer_group_count": info.consumer_group_count,
        "oldest_timestamp_millis": info.oldest_timestamp_millis,
        "newest_timestamp_millis": info.newest_timestamp_millis,
    })))
}

/// `GET /v1/{namespace}/topics` — list topics in namespace.
pub async fn list_topics<M: NamespaceManager + 'static>(
    State(state): State<AppState<M>>,
    Path(namespace): Path<String>,
    headers: HeaderMap,
) -> Result<impl IntoResponse, HttpError> {
    let session = authenticate(&*state.auth, &headers)?;
    let ns = NamespaceId(namespace);
    verify_namespace(&session, &ns)?;
    session.permissions.check(Permission::Query)?;

    let topics = state.manager.list_topics(&ns).await?;
    let topic_names: Vec<&str> = topics.iter().map(|t| t.0.as_str()).collect();

    Ok(Json(serde_json::json!({
        "namespace": ns.0,
        "topics": topic_names,
    })))
}

/// `PUT /v1/admin/log-level` — dynamically change the log level filter.
///
/// Body is a plain-text RUST_LOG-style filter string (e.g. `"info,fafb_q::storage=debug"`).
/// Requires a valid API key (any namespace) with Query permission.
pub async fn log_level<M: NamespaceManager + 'static>(
    State(state): State<AppState<M>>,
    headers: HeaderMap,
    body: String,
) -> Result<impl IntoResponse, HttpError> {
    let session = authenticate(&*state.auth, &headers)?;
    session.permissions.check(Permission::Query)?;

    let filter_str = body.trim();
    match filter_str.parse::<tracing_subscriber::EnvFilter>() {
        Ok(new_filter) => match state.log_reload_handle.reload(new_filter) {
            Ok(()) => {
                tracing::info!(
                    filter = %filter_str,
                    namespace = %session.namespace,
                    "log level changed dynamically"
                );
                Ok((StatusCode::OK, format!("log level set to: {filter_str}")))
            }
            Err(e) => {
                tracing::warn!(error = %e, "failed to reload log filter");
                Ok((
                    StatusCode::INTERNAL_SERVER_ERROR,
                    format!("failed to reload: {e}"),
                ))
            }
        },
        Err(e) => Ok((
            StatusCode::BAD_REQUEST,
            format!("invalid filter string: {e}"),
        )),
    }
}

/// `GET /v1/admin/log-level` — return the current log level filter.
///
/// Requires a valid API key (any namespace) with Query permission.
pub async fn log_level_get<M: NamespaceManager + 'static>(
    State(state): State<AppState<M>>,
    headers: HeaderMap,
) -> Result<impl IntoResponse, HttpError> {
    let session = authenticate(&*state.auth, &headers)?;
    session.permissions.check(Permission::Query)?;

    let current = state
        .log_reload_handle
        .with_current(|f| f.to_string())
        .unwrap_or_else(|_| "unknown".to_string());
    Ok((StatusCode::OK, current))
}

/// `POST /admin/drain` — trigger a graceful drain via HTTP.
///
/// Accepts an optional JSON body to override drain parameters:
/// `{ "reason": "upgrade", "timeout_secs": 30 }`
///
/// Returns 200 with drain metadata on success, 409 if already draining.
pub async fn admin_drain<M: NamespaceManager + 'static>(
    State(state): State<AppState<M>>,
    body: Option<Json<serde_json::Value>>,
) -> impl IntoResponse {
    // Already draining — return 409 with current status
    if state.lifecycle.is_draining() {
        let status = build_drain_status_json(&state.lifecycle);
        return (StatusCode::CONFLICT, Json(status));
    }

    // Parse optional overrides from body
    let reason = body
        .as_ref()
        .and_then(|Json(v)| v.get("reason").and_then(|r| r.as_str()).map(String::from));
    let timeout_secs = body
        .as_ref()
        .and_then(|Json(v)| v.get("timeout_secs").and_then(|t| t.as_u64()));

    // Send drain request to main.rs via channel
    let trigger = match &state.drain_trigger {
        Some(tx) => tx,
        None => {
            return (
                StatusCode::INTERNAL_SERVER_ERROR,
                Json(serde_json::json!({
                    "error": "drain trigger not available"
                })),
            );
        }
    };

    let (resp_tx, resp_rx) = tokio::sync::oneshot::channel();
    let req = super::DrainRequest {
        reason,
        timeout_secs,
        response: resp_tx,
    };

    if trigger.send(req).await.is_err() {
        return (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(serde_json::json!({
                "error": "drain channel closed"
            })),
        );
    }

    match resp_rx.await {
        Ok(Ok(resp)) => (
            StatusCode::OK,
            Json(serde_json::json!({
                "status": "draining",
                "active_connections": resp.active_connections,
                "deadline": millis_to_rfc3339(resp.drain_deadline_wall),
            })),
        ),
        Ok(Err(msg)) => (
            StatusCode::CONFLICT,
            Json(serde_json::json!({
                "status": "already_draining",
                "message": msg,
            })),
        ),
        Err(_) => (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(serde_json::json!({
                "error": "drain request failed"
            })),
        ),
    }
}

/// `GET /admin/drain/status` — return current drain progress.
pub async fn admin_drain_status<M: NamespaceManager + 'static>(
    State(state): State<AppState<M>>,
) -> impl IntoResponse {
    Json(build_drain_status_json(&state.lifecycle))
}

fn build_drain_status_json(lifecycle: &ServerLifecycle) -> serde_json::Value {
    let state_str = match lifecycle.state() {
        ServerState::Running => "running",
        ServerState::Draining => "draining",
        ServerState::ShuttingDown => "shutting_down",
    };

    serde_json::json!({
        "state": state_str,
        "active_connections": lifecycle.active_connections(),
        "drain_started_at": lifecycle.drain_started_wall().map(millis_to_rfc3339),
        "drain_deadline": lifecycle.drain_deadline_wall().map(millis_to_rfc3339),
        "connections_drained": lifecycle.connections_drained(),
    })
}

/// `POST /admin/tls/reload` — trigger immediate TLS certificate reload.
///
/// Returns 200 with per-provider results on success, 500 on reload failure,
/// 404 when TLS is not enabled.
pub async fn admin_tls_reload<M: NamespaceManager + 'static>(
    axum::Extension(providers): axum::Extension<TlsProviders>,
) -> impl IntoResponse {
    if providers.0.is_empty() {
        return (
            StatusCode::NOT_FOUND,
            Json(serde_json::json!({
                "error": "TLS is not enabled"
            })),
        );
    }

    let mut results = Vec::new();
    let mut any_error = false;

    for lp in &providers.0 {
        match lp.provider.reload() {
            Ok(changed) => {
                let status_str = if changed { "success" } else { "unchanged" };
                metrics::counter!("fafb_tls_reload_total",
                    "status" => status_str, "listener" => lp.label
                )
                .increment(1);
                if changed {
                    metrics::gauge!("fafb_tls_cert_expiry_seconds",
                        "listener" => lp.label
                    )
                    .set(lp.provider.cert_ttl_secs() as f64);
                    tracing::info!(
                        listener = lp.label,
                        cert_ttl_secs = lp.provider.cert_ttl_secs(),
                        "TLS cert reloaded via admin endpoint"
                    );
                }
                results.push(serde_json::json!({
                    "listener": lp.label,
                    "status": status_str,
                    "cert_expires": millis_to_rfc3339(lp.provider.cert_expiry_epoch() * 1000),
                }));
            }
            Err(e) => {
                metrics::counter!("fafb_tls_reload_total",
                    "status" => "error", "listener" => lp.label
                )
                .increment(1);
                tracing::error!(
                    listener = lp.label,
                    error = %e,
                    "TLS cert reload failed via admin endpoint"
                );
                any_error = true;
                results.push(serde_json::json!({
                    "listener": lp.label,
                    "status": "error",
                    "error": e.to_string(),
                }));
            }
        }
    }

    let status_code = if any_error {
        StatusCode::INTERNAL_SERVER_ERROR
    } else {
        StatusCode::OK
    };

    (
        status_code,
        Json(serde_json::json!({ "providers": results })),
    )
}

/// `GET /admin/tls/status` — return TLS certificate details without triggering reload.
pub async fn admin_tls_status<M: NamespaceManager + 'static>(
    axum::Extension(providers): axum::Extension<TlsProviders>,
) -> impl IntoResponse {
    if providers.0.is_empty() {
        return (
            StatusCode::NOT_FOUND,
            Json(serde_json::json!({
                "error": "TLS is not enabled"
            })),
        );
    }

    let certs: Vec<_> = providers
        .0
        .iter()
        .map(|lp| {
            serde_json::json!({
                "listener": lp.label,
                "cert_file": lp.provider.cert_file(),
                "cert_expires": millis_to_rfc3339(lp.provider.cert_expiry_epoch() * 1000),
                "cert_ttl_secs": lp.provider.cert_ttl_secs(),
                "cert_fingerprint_sha256": lp.provider.cert_fingerprint(),
                "client_auth": lp.provider.client_auth(),
                "min_version": lp.provider.min_version(),
            })
        })
        .collect();

    (
        StatusCode::OK,
        Json(serde_json::json!({ "providers": certs })),
    )
}

/// `POST /admin/reload` — re-read config file, validate, diff, and apply changes.
///
/// Returns detailed JSON showing which fields changed, which were ignored
/// (Tier 3 / restart-only), and any namespace additions/removals.
pub async fn admin_reload<M: NamespaceManager + 'static>(
    State(state): State<AppState<M>>,
) -> impl IntoResponse {
    let reload_state = match &state.reload_state {
        Some(rs) => rs,
        None => {
            return (
                StatusCode::BAD_REQUEST,
                Json(serde_json::json!({
                    "status": "error",
                    "message": "no config file loaded — server started with compiled defaults"
                })),
            );
        }
    };

    let start = std::time::Instant::now();
    match reload_state.reload().await {
        Ok(diff) if diff.is_empty() => {
            let elapsed = start.elapsed();
            metrics::counter!("fafb_config_reloads_total",
                "trigger" => "http", "result" => "unchanged")
            .increment(1);
            metrics::histogram!("fafb_config_reload_duration_seconds")
                .record(elapsed.as_secs_f64());
            (
                StatusCode::OK,
                Json(serde_json::json!({
                    "status": "unchanged"
                })),
            )
        }
        Ok(diff) => {
            let elapsed = start.elapsed();
            let changes = diff.applied.len();
            metrics::counter!("fafb_config_reloads_total",
                "trigger" => "http", "result" => "success")
            .increment(1);
            metrics::histogram!("fafb_config_reload_duration_seconds")
                .record(elapsed.as_secs_f64());
            metrics::gauge!("fafb_config_last_reload_timestamp")
                .set(crate::types::now_millis() as f64 / 1000.0);
            metrics::gauge!("fafb_config_last_reload_changes").set(changes as f64);
            (
                StatusCode::OK,
                Json(serde_json::json!({
                    "status": "applied",
                    "changes_applied": changes,
                    "changes_ignored": diff.ignored.len(),
                    "applied": diff.applied,
                    "ignored": diff.ignored,
                    "namespaces_added": diff.namespaces_added,
                    "namespaces_removed": diff.namespaces_removed,
                })),
            )
        }
        Err(crate::config::ReloadError::ValidationError(errors)) => {
            metrics::counter!("fafb_config_reloads_total",
                "trigger" => "http", "result" => "error")
            .increment(1);
            metrics::histogram!("fafb_config_reload_duration_seconds")
                .record(start.elapsed().as_secs_f64());
            let messages: Vec<String> = errors.iter().map(|e| e.to_string()).collect();
            (
                StatusCode::BAD_REQUEST,
                Json(serde_json::json!({
                    "status": "error",
                    "message": format!("config validation failed: {}", messages.join("; ")),
                    "errors": messages,
                })),
            )
        }
        Err(crate::config::ReloadError::NamespaceRemovalBlocked {
            namespace,
            active_connections,
        }) => {
            metrics::counter!("fafb_config_reloads_total",
                "trigger" => "http", "result" => "error")
            .increment(1);
            metrics::histogram!("fafb_config_reload_duration_seconds")
                .record(start.elapsed().as_secs_f64());
            (
                StatusCode::CONFLICT,
                Json(serde_json::json!({
                    "status": "error",
                    "message": format!(
                        "cannot remove namespace '{}': {} active connections",
                        namespace, active_connections
                    ),
                })),
            )
        }
        Err(crate::config::ReloadError::ServerDraining) => {
            metrics::counter!("fafb_config_reloads_total",
                "trigger" => "http", "result" => "error")
            .increment(1);
            (
                StatusCode::SERVICE_UNAVAILABLE,
                Json(serde_json::json!({
                    "status": "error",
                    "message": "config reload rejected: server is draining",
                })),
            )
        }
        Err(e) => {
            metrics::counter!("fafb_config_reloads_total",
                "trigger" => "http", "result" => "error")
            .increment(1);
            metrics::histogram!("fafb_config_reload_duration_seconds")
                .record(start.elapsed().as_secs_f64());
            (
                StatusCode::BAD_REQUEST,
                Json(serde_json::json!({
                    "status": "error",
                    "message": e.to_string(),
                })),
            )
        }
    }
}

/// `GET /admin/config` — return the current runtime config as JSON with
/// API keys redacted. Requires admin auth.
pub async fn admin_config<M: NamespaceManager + 'static>(
    State(state): State<AppState<M>>,
) -> impl IntoResponse {
    let runtime_config = state.shared_config.load();
    let json = runtime_config.to_json_redacted();
    (StatusCode::OK, Json(json))
}

/// Query parameters for the fetch endpoint.
///
/// `offset` accepts negative values for relative-from-head semantics:
/// `-1` = last message, `-N` = N messages from head.
#[derive(Debug, serde::Deserialize)]
pub struct FetchParams {
    pub offset: Option<i64>,
    pub timestamp: Option<u64>,
    pub max_count: Option<u32>,
}

/// `GET /v1/{namespace}/{topic}/fetch` — pull a batch of messages.
///
/// Stateless pull-based message retrieval. Requires `Fetch` permission.
/// Either `offset` or `timestamp` query parameter must be provided (not both).
pub async fn fetch<M: NamespaceManager + 'static>(
    State(state): State<AppState<M>>,
    Path((namespace, topic)): Path<(String, String)>,
    Query(params): Query<FetchParams>,
    headers: HeaderMap,
) -> Result<impl IntoResponse, HttpError> {
    let max_fetch_response_bytes = state.shared_config.load().max_fetch_response_bytes;
    let session = authenticate(&*state.auth, &headers)?;
    let ns = NamespaceId(namespace);
    verify_namespace(&session, &ns)?;
    session.permissions.check(Permission::Fetch)?;

    let topic = TopicName(topic);

    if crate::wildcard::contains_wildcard(&topic.0) {
        return Err(HttpError::new(
            StatusCode::BAD_REQUEST,
            ErrorCode::InvalidTopic,
            "wildcard patterns not supported for Fetch",
        ));
    }

    // Validate query parameters: exactly one of offset or timestamp required.
    let offset = match (params.offset, params.timestamp) {
        (None, None) => {
            return Err(HttpError::new(
                StatusCode::BAD_REQUEST,
                ErrorCode::InvalidFrame,
                "one of 'offset' or 'timestamp' query parameter is required",
            ));
        }
        (Some(_), Some(_)) => {
            return Err(HttpError::new(
                StatusCode::BAD_REQUEST,
                ErrorCode::InvalidFrame,
                "'offset' and 'timestamp' are mutually exclusive",
            ));
        }
        (Some(o), None) => {
            if o >= 0 {
                TopicOffset(o as u64)
            } else {
                // Negative offset: resolve relative to head.
                // -1 = last message, -N = N messages from head.
                let head = state
                    .manager
                    .head_offset(&ns, &topic)
                    .await
                    .unwrap_or(TopicOffset::ZERO);
                TopicOffset(head.0.saturating_sub(o.unsigned_abs()))
            }
        }
        (None, Some(ts)) => {
            // Resolve timestamp to offset.
            match state.manager.offset_at_timestamp(&ns, &topic, ts).await {
                Ok(Some(o)) => o,
                Ok(None) => {
                    // No message at or after timestamp — return empty at head.
                    let head = state
                        .manager
                        .head_offset(&ns, &topic)
                        .await
                        .unwrap_or(TopicOffset(0));
                    return Ok(Json(serde_json::json!({
                        "topic": topic.0,
                        "next_offset": head.0,
                        "head_offset": head.0,
                        "messages": [],
                    })));
                }
                Err(e) => return Err(e.into()),
            }
        }
    };

    // Validate max_count: default 100, max 10,000, must be > 0.
    let max_count = params.max_count.unwrap_or(100);
    if max_count == 0 {
        return Err(HttpError::new(
            StatusCode::BAD_REQUEST,
            ErrorCode::InvalidFrame,
            "max_count must be greater than 0",
        ));
    }
    if max_count > 10_000 {
        return Err(HttpError::new(
            StatusCode::BAD_REQUEST,
            ErrorCode::InvalidFrame,
            "max_count must not exceed 10000",
        ));
    }

    let max_response_bytes = max_fetch_response_bytes;

    match state
        .manager
        .fetch(&ns, &topic, offset, max_count, max_response_bytes, false)
        .await
    {
        Ok(result) => {
            let b64 = base64::engine::general_purpose::STANDARD;
            let messages: Vec<serde_json::Value> = result
                .messages
                .iter()
                .map(|m| {
                    let mut msg = serde_json::json!({
                        "offset": m.offset.0,
                        "timestamp_millis": m.envelope.timestamp,
                        "ttl_ms": m.envelope.ttl_ms,
                        "payload_base64": b64.encode(&m.envelope.payload),
                    });
                    if !m.envelope.headers.is_empty() {
                        msg["headers"] = headers_to_json(&m.envelope.headers);
                    }
                    msg
                })
                .collect();

            let mut body = serde_json::json!({
                "topic": topic.0,
                "next_offset": result.next_offset.0,
                "head_offset": result.head_offset.0,
                "messages": messages,
            });

            if result.clamped_from.is_some() {
                body["warning"] =
                    serde_json::json!("requested offset not available, clamped to oldest retained");
            }

            Ok(Json(body))
        }
        Err(e) => Err(e.into()),
    }
}

/// The HTTP header name prefix for message headers.
const HEADER_PREFIX: &str = "x-fafb-";

/// Extract message headers from HTTP request headers.
///
/// Any HTTP header whose name starts with `X-Fafb-` (case-insensitive) is
/// collected as a message header. The `X-Fafb-` prefix is stripped to produce
/// the message header key. For example, `X-Fafb-Content-Type: application/json`
/// becomes a header with key `Content-Type` and value `application/json`.
fn extract_message_headers(http_headers: &HeaderMap) -> Headers {
    let mut headers = Vec::new();
    for (name, value) in http_headers.iter() {
        let name_str = name.as_str();
        if let Some(key) = name_str.strip_prefix(HEADER_PREFIX) {
            // HTTP header names are lowercased by axum/hyper. Preserve the
            // original casing by using the stripped suffix as-is. Clients who
            // care about case should send the desired casing after the prefix
            // (HTTP header names are case-insensitive, but message header keys
            // are opaque bytes).
            headers.push(Header {
                key: key.as_bytes().to_vec(),
                value: value.as_bytes().to_vec(),
            });
        }
    }
    headers
}

/// `GET /admin/topics/pending-reap` — list topics currently in the grace period
/// before being reaped by the topic inactivity GC.
///
/// Returns a JSON array of objects, each with:
/// - `namespace` — namespace name
/// - `topic` — topic name
/// - `last_activity_millis` — epoch-millis of last observed activity when the topic entered grace
/// - `threshold_secs` — inactivity threshold that was exceeded
/// - `grace_period_secs` — grace period configured at the time the topic entered grace
/// - `grace_entered_secs_ago` — seconds since the topic entered the grace period
/// - `grace_remaining_secs` — seconds until the grace period expires (may be negative
///   if the topic is overdue; it will be reaped on the next retention sweep)
///
/// Requires admin auth.
pub async fn admin_pending_reap<M: NamespaceManager + 'static>(
    State(state): State<AppState<M>>,
) -> axum::Json<Vec<PendingReapEntry>> {
    let map = state.grace_pending.read().await;
    let now = std::time::Instant::now();

    let mut entries: Vec<PendingReapEntry> = map
        .iter()
        .map(|((ns, topic), entry)| {
            let elapsed = now.duration_since(entry.entered).as_secs();
            PendingReapEntry {
                namespace: ns.0.clone(),
                topic: topic.0.clone(),
                last_activity_millis: entry.last_activity_millis,
                threshold_secs: entry.threshold_secs,
                grace_period_secs: entry.grace_period_secs,
                grace_entered_secs_ago: elapsed,
                grace_remaining_secs: entry.grace_period_secs as i64 - elapsed as i64,
            }
        })
        .collect();

    // Sort for deterministic output: namespace, then topic.
    entries.sort_by(|a, b| a.namespace.cmp(&b.namespace).then(a.topic.cmp(&b.topic)));

    axum::Json(entries)
}

/// An entry in the pending-reap list.
#[derive(serde::Serialize)]
pub struct PendingReapEntry {
    pub namespace: String,
    pub topic: String,
    pub last_activity_millis: u64,
    pub threshold_secs: u64,
    pub grace_period_secs: u64,
    pub grace_entered_secs_ago: u64,
    pub grace_remaining_secs: i64,
}

/// Convert message headers to a JSON array for fetch responses.
///
/// Returns an array of objects, each with `"key"` and either `"value"` (UTF-8
/// string) or `"value_base64"` (base64-encoded binary). This format preserves
/// insertion order and allows duplicate keys (matching the in-memory `Headers`
/// semantics).
///
/// ```json
/// [
///   {"key": "Content-Type", "value": "application/json"},
///   {"key": "binary-data", "value_base64": "//4AAQ=="}
/// ]
/// ```
fn headers_to_json(headers: &[Header]) -> serde_json::Value {
    let b64 = base64::engine::general_purpose::STANDARD;
    let arr: Vec<serde_json::Value> = headers
        .iter()
        .map(|header| {
            let key_str = String::from_utf8_lossy(&header.key);
            match std::str::from_utf8(&header.value) {
                Ok(val) => serde_json::json!({
                    "key": key_str,
                    "value": val,
                }),
                Err(_) => serde_json::json!({
                    "key": key_str,
                    "value_base64": b64.encode(&header.value),
                }),
            }
        })
        .collect();
    serde_json::Value::Array(arr)
}
