ursula 0.5.0

//! Ursula HTTP server: axum router, request handlers, response rendering,
//! plus the typed-config bootstrap constructors used by the `ursula` binary.
//!
//! Module map:
//!
//! - [`render`]: response builders, header helpers, SSE/multipart rendering.
//! - [`bootstrap`]: typed-config `spawn_*_runtime` constructors and cold-flush worker.
//! - [`server`]: command arguments and the long-running server service entrypoint.

mod bootstrap;
mod otel_metrics;
pub mod server;
mod http_time {
    #[cfg(madsim)]
    pub use madsim::time::timeout;
    #[cfg(not(madsim))]
    pub use tokio::time::timeout;
}
mod render;
mod wal_disk;

use std::collections::BTreeMap;
use std::collections::BTreeSet;
use std::collections::HashMap;
use std::convert::Infallible;
use std::sync::Arc;
use std::sync::atomic::AtomicU64;
use std::sync::atomic::Ordering;
use std::time::Duration;
#[cfg(not(madsim))]
use std::time::SystemTime;
#[cfg(not(madsim))]
use std::time::UNIX_EPOCH;

use axum::Router;
use axum::body::Body;
use axum::body::Bytes;
use axum::body::HttpBody;
use axum::extract::DefaultBodyLimit;
use axum::extract::OriginalUri;
use axum::extract::Path;
use axum::extract::RawQuery;
use axum::extract::State;
use axum::http::HeaderMap;
use axum::http::HeaderValue;
use axum::http::Method;
use axum::http::Request;
use axum::http::StatusCode;
use axum::http::Uri;
use axum::http::Version;
#[cfg(feature = "jemalloc-prof")]
use axum::http::header::CONTENT_DISPOSITION;
use axum::http::header::CONTENT_LENGTH;
use axum::http::header::CONTENT_TYPE;
use axum::http::header::LOCATION;
use axum::middleware::Next;
use axum::middleware::{self};
use axum::response::IntoResponse;
use axum::response::Response;
use axum::routing::get;
use axum::routing::post;
use axum::routing::put;
use base64::Engine;
use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
pub use bootstrap::Persistence;
pub use bootstrap::SpawnedRuntime;
pub use bootstrap::Topology;
pub use bootstrap::spawn_runtime;
use chrono::DateTime;
use futures_util::stream;
use openraft::BasicNode;
use openraft::rt::WatchReceiver;
use tower_http::compression::CompressionLayer;
use tower_http::compression::CompressionLevel;
use tower_http::compression::predicate::Predicate;
use tower_http::compression::predicate::SizeAbove;
use ursula_raft::LeadershipShedFlag;
use ursula_raft::LeadershipShedReason;
use ursula_raft::RAFT_GRPC_APPEND_PATH;
use ursula_raft::RAFT_GRPC_APPEND_STREAM_PATH;
use ursula_raft::RAFT_GRPC_FULL_SNAPSHOT_PATH;
use ursula_raft::RAFT_GRPC_GROUP_READ_PATH;
use ursula_raft::RAFT_GRPC_GROUP_WRITE_PATH;
use ursula_raft::RAFT_GRPC_MAX_MESSAGE_BYTES;
use ursula_raft::RAFT_GRPC_TRANSFER_LEADER_PATH;
use ursula_raft::RAFT_GRPC_VOTE_PATH;
use ursula_raft::RaftGroupHandle;
use ursula_raft::RaftGroupHandleRegistry;
use ursula_raft::RaftGrpcService;
use ursula_raft::raft_internal_proto;
use ursula_runtime::AdvanceRetentionRequest;
use ursula_runtime::AppendBatchRequest;
use ursula_runtime::AppendExternalRequest;
use ursula_runtime::AppendRequest;
use ursula_runtime::AppendResponse;
use ursula_runtime::AppendTransactionRequest;
use ursula_runtime::BootstrapStreamRequest;
use ursula_runtime::CloseStreamRequest;
use ursula_runtime::CreateStreamExternalRequest;
use ursula_runtime::CreateStreamRequest;
use ursula_runtime::CreateStreamResponse;
use ursula_runtime::DeleteSnapshotRequest;
use ursula_runtime::DeleteStreamRequest;
use ursula_runtime::ErrorStatus;
use ursula_runtime::ExternalPayloadRef;
use ursula_runtime::GetStreamAttrsRequest;
use ursula_runtime::HeadStreamRequest;
use ursula_runtime::ImportGroupStateRequest;
use ursula_runtime::PlanColdFlushRequest;
use ursula_runtime::ProducerRequest;
use ursula_runtime::PublishSnapshotRequest;
use ursula_runtime::ReadSnapshotRequest;
use ursula_runtime::ReadStreamRequest;
use ursula_runtime::RuntimeError;
use ursula_runtime::ShardRuntime;
use ursula_runtime::StreamAttrs;
use ursula_runtime::UpdateStreamAttrsRequest;
use ursula_runtime::new_external_payload_path;
use ursula_shard::BucketStreamId;
use ursula_shard::RaftGroupId;
use ursula_shard::is_reserved_affinity_stream_id;
use wal_disk::WalDiskMonitor;

use crate::render::apply_record_envelope;
use crate::render::bootstrap_response;
use crate::render::clamp_sse_text_read;
use crate::render::http_read_content_type;
use crate::render::insert_cache_control;
use crate::render::insert_content_type;
use crate::render::insert_cursor;
use crate::render::insert_default_response_headers;
use crate::render::insert_header_str;
use crate::render::insert_lifetime_headers;
use crate::render::insert_location;
use crate::render::insert_offset;
use crate::render::insert_producer_ack;
use crate::render::insert_producer_error_headers;
use crate::render::insert_public_location;
use crate::render::insert_snapshot_digest;
use crate::render::insert_snapshot_offset;
use crate::render::insert_static;
use crate::render::insert_stream_error_headers;
use crate::render::insert_stream_error_offset;
use crate::render::insert_u64_header;
use crate::render::long_poll_no_content_response;
use crate::render::normalize_http_write_payload;
use crate::render::offset_now_response;
use crate::render::parse_append_batch;
use crate::render::read_response;
use crate::render::record_envelope_response;
use crate::render::render_batch_results;
use crate::render::render_metrics;
use crate::render::render_sse_read;
use crate::render::response_cursor;
use crate::render::runtime_error_status;
use crate::render::should_base64_encode_sse_data;
use crate::render::snapshot_response;
use crate::render::sse_safe_line;

type BoxResponse = Box<Response>;

const DEFAULT_CONTENT_TYPE: &str = "application/octet-stream";
const HEADER_STREAM_CLOSED: &str = "stream-closed";
const HEADER_STREAM_CURSOR: &str = "stream-cursor";
const HEADER_STREAM_EXPIRES_AT: &str = "stream-expires-at";
const HEADER_STREAM_EXTENSIONS: &str = "stream-extensions";
const HEADER_STREAM_INTEGRITY_EVICTED_RECORDS: &str = "stream-integrity-evicted-records";
const HEADER_STREAM_INTEGRITY_EVICTED_SETSUM: &str = "stream-integrity-evicted-setsum";
const HEADER_STREAM_INTEGRITY_LIVE_RECORDS: &str = "stream-integrity-live-records";
const HEADER_STREAM_INTEGRITY_LIVE_SETSUM: &str = "stream-integrity-live-setsum";
const HEADER_STREAM_INTEGRITY_LIVE_START_OFFSET: &str = "stream-integrity-live-start-offset";
const HEADER_STREAM_INTEGRITY_TOTAL_RECORDS: &str = "stream-integrity-total-records";
const HEADER_STREAM_INTEGRITY_TOTAL_SETSUM: &str = "stream-integrity-total-setsum";
const HEADER_STREAM_COLD_HOT_START_OFFSET: &str = "stream-cold-hot-start-offset";
const HEADER_STREAM_DATA_CONTENT_TYPE: &str = "stream-data-content-type";
const HEADER_STREAM_NEXT_OFFSET: &str = "stream-next-offset";
const HEADER_STREAM_RECORD_FIRST: &str = "stream-record-first";
const HEADER_STREAM_RECORD_MATCH: &str = "stream-record-match";
const HEADER_STREAM_RECORD_NEXT: &str = "stream-record-next";
const HEADER_STREAM_RECORD_START: &str = "stream-record-start";
const HEADER_STREAM_SNAPSHOT_OFFSET: &str = "stream-snapshot-offset";
const HEADER_STREAM_SNAPSHOT_DIGEST: &str = "stream-snapshot-digest";
const HEADER_STREAM_SNAPSHOT_MATCH: &str = "stream-snapshot-match";
const HEADER_STREAM_RETAINED_OFFSET: &str = "stream-retained-offset";
const HEADER_STREAM_SSE_DATA_ENCODING: &str = "stream-sse-data-encoding";
const HEADER_STREAM_ATTRS: &str = "stream-attrs";
const HEADER_STREAM_SEQ: &str = "stream-seq";
const HEADER_STREAM_TTL: &str = "stream-ttl";
const HEADER_STREAM_UP_TO_DATE: &str = "stream-up-to-date";
const JSON_RECORD_COORDINATES_EXTENSION: &str = "json-record-coordinates-v1";
const PATH_AFFINITY_EXTENSION: &str = "path-affinity-v1";
const GROUP_APPEND_TRANSACTION_EXTENSION: &str = "group-append-transaction-v1";
const HEADER_PRODUCER_ID: &str = "producer-id";
const HEADER_PRODUCER_EPOCH: &str = "producer-epoch";
const HEADER_PRODUCER_SEQ: &str = "producer-seq";
const HEADER_PREFER: &str = "prefer";
const HEADER_X_CONTENT_TYPE_OPTIONS: &str = "x-content-type-options";
const HEADER_CROSS_ORIGIN_RESOURCE_POLICY: &str = "cross-origin-resource-policy";
const HEADER_URSULA_RAFT_LEADER_ID: &str = "x-ursula-raft-leader-id";
#[cfg(feature = "jemalloc-prof")]
const HEADER_URSULA_DEBUG_TOKEN: &str = "x-ursula-debug-token";
// tikv-jemalloc-sys forces the `_rjem_` symbol prefix on Apple targets, so
// jemalloc reads `_RJEM_MALLOC_CONF` there instead of `MALLOC_CONF`.
#[cfg(feature = "jemalloc-prof")]
const MALLOC_CONF_ENV_VAR: &str = if cfg!(target_vendor = "apple") {
    "_RJEM_MALLOC_CONF"
} else {
    "MALLOC_CONF"
};
const APPEND_BATCH_MAX_ITEMS: usize = 512;
const APPEND_BATCH_MAX_BYTES: usize = 32 * 1024 * 1024;
const MAX_HTTP_BODY_BYTES: usize = 32 * 1024 * 1024;
const DEFAULT_HTTP_INFLIGHT_BODY_BYTES: usize = MAX_HTTP_BODY_BYTES * 8;
const DEFAULT_LONG_POLL_TIMEOUT_MS: u64 = 1_000;
const MAX_LONG_POLL_TIMEOUT_MS: u64 = 60_000;

#[derive(Debug, serde::Deserialize)]
pub(crate) struct StreamPath {
    bucket: String,
    #[serde(default)]
    affinity: Option<String>,
    stream: String,
}

#[derive(Debug, serde::Deserialize)]
pub(crate) struct AffinityPath {
    bucket: String,
    affinity: String,
}

#[derive(Debug, serde::Deserialize)]
#[serde(deny_unknown_fields)]
struct AppendTransactionHttpRequest {
    operations: Vec<AppendTransactionHttpOperation>,
}

#[derive(Debug, serde::Deserialize)]
#[serde(deny_unknown_fields)]
struct AppendTransactionHttpOperation {
    stream: String,
    content_type: String,
    payload_base64: String,
    #[serde(default)]
    close_after: bool,
    #[serde(default)]
    stream_seq: Option<String>,
    #[serde(default)]
    producer: Option<ProducerRequest>,
    #[serde(default)]
    record_match: Option<u64>,
}

impl StreamPath {
    fn into_stream_id(self) -> BucketStreamId {
        match self.affinity {
            Some(affinity) => BucketStreamId::with_affinity(self.bucket, affinity, self.stream),
            None => BucketStreamId::new(self.bucket, self.stream),
        }
    }

    fn stream_id(&self) -> BucketStreamId {
        match &self.affinity {
            Some(affinity) => BucketStreamId::with_affinity(
                self.bucket.clone(),
                affinity.clone(),
                self.stream.clone(),
            ),
            None => BucketStreamId::new(self.bucket.clone(), self.stream.clone()),
        }
    }
}

#[derive(Debug, serde::Deserialize)]
pub(crate) struct SnapshotPath {
    bucket: String,
    #[serde(default)]
    affinity: Option<String>,
    stream: String,
    snapshot_offset: String,
}

impl SnapshotPath {
    fn into_parts(self) -> (BucketStreamId, String) {
        let stream_id = match self.affinity {
            Some(affinity) => BucketStreamId::with_affinity(self.bucket, affinity, self.stream),
            None => BucketStreamId::new(self.bucket, self.stream),
        };
        (stream_id, self.snapshot_offset)
    }
}

#[derive(Debug, serde::Deserialize)]
pub(crate) struct RetentionPath {
    bucket: String,
    #[serde(default)]
    affinity: Option<String>,
    stream: String,
    retained_offset: String,
}

impl RetentionPath {
    fn into_parts(self) -> (BucketStreamId, String) {
        let stream_id = match self.affinity {
            Some(affinity) => BucketStreamId::with_affinity(self.bucket, affinity, self.stream),
            None => BucketStreamId::new(self.bucket, self.stream),
        };
        (stream_id, self.retained_offset)
    }
}

struct CreateStreamHttpResponseInput<'a> {
    response: CreateStreamResponse,
    stream_id: &'a BucketStreamId,
    content_type: &'a str,
    stream_ttl_seconds: Option<u64>,
    stream_expires_at_ms: Option<u64>,
    producer: Option<&'a ProducerRequest>,
}

pub trait WallClock: Send + Sync + 'static {
    fn unix_time_ms(&self) -> u64;
}

#[derive(Debug, Default)]
pub struct SystemWallClock;

impl WallClock for SystemWallClock {
    fn unix_time_ms(&self) -> u64 {
        unix_time_ms()
    }
}

#[derive(Clone)]
pub struct HttpState {
    runtime: ShardRuntime,
    raft_registry: Option<RaftGroupHandleRegistry>,
    client_write_router: Option<ClientWriteLeaderRouter>,
    http_metrics: Arc<HttpMetrics>,
    wall_clock: Arc<dyn WallClock>,
    pub node_memory: NodeMemoryMonitor,
    leadership_shed: LeadershipShedFlag,
    external_payload_min_bytes: usize,
    /// Raft WAL backend name (`"memory"` / `"disk"`) surfaced in the metrics
    /// JSON so operator tooling can tell a volatile node from a durable one
    /// (e.g. ursulactl auto-enabling empty-log rejoin only for `memory`).
    wal_backend: &'static str,
    wal_disk: WalDiskMonitor,
}

impl HttpState {
    /// Bridge the runtime's metrics to the global OTLP meter (export-time
    /// observable instruments; no hot-path cost). Inert when no OTLP meter
    /// provider is installed.
    pub fn register_otel_metrics(&self) {
        otel_metrics::register(&self.runtime.metrics());
    }

    pub fn new(runtime: ShardRuntime) -> Self {
        Self {
            runtime,
            raft_registry: None,
            client_write_router: None,
            http_metrics: Arc::new(HttpMetrics::default()),
            wall_clock: Arc::new(SystemWallClock),
            node_memory: NodeMemoryMonitor::default(),
            leadership_shed: Arc::new(std::sync::atomic::AtomicU8::new(0)),
            external_payload_min_bytes: 1024 * 1024,
            wal_backend: "memory",
            wal_disk: WalDiskMonitor::default(),
        }
    }

    pub fn with_raft_registry(
        runtime: ShardRuntime,
        raft_registry: RaftGroupHandleRegistry,
    ) -> Self {
        let leadership_shed = raft_registry.leadership_shed_flag();
        Self {
            runtime,
            raft_registry: Some(raft_registry),
            client_write_router: None,
            http_metrics: Arc::new(HttpMetrics::default()),
            wall_clock: Arc::new(SystemWallClock),
            node_memory: NodeMemoryMonitor::default(),
            leadership_shed,
            external_payload_min_bytes: 1024 * 1024,
            wal_backend: "memory",
            wal_disk: WalDiskMonitor::default(),
        }
    }

    pub fn with_static_raft_cluster(
        runtime: ShardRuntime,
        raft_registry: RaftGroupHandleRegistry,
        peers: impl IntoIterator<Item = (u64, String)>,
    ) -> Self {
        Self::with_static_raft_cluster_topology(
            runtime,
            raft_registry,
            None,
            peers,
            BTreeMap::new(),
        )
    }

    pub fn with_static_raft_cluster_topology(
        runtime: ShardRuntime,
        raft_registry: RaftGroupHandleRegistry,
        node_id: impl Into<Option<u64>>,
        peers: impl IntoIterator<Item = (u64, String)>,
        per_group_voters: BTreeMap<RaftGroupId, BTreeSet<u64>>,
    ) -> Self {
        let leadership_shed = raft_registry.leadership_shed_flag();
        Self {
            runtime,
            raft_registry: Some(raft_registry),
            client_write_router: Some(ClientWriteLeaderRouter::with_static_topology(
                node_id,
                peers,
                per_group_voters,
            )),
            http_metrics: Arc::new(HttpMetrics::default()),
            wall_clock: Arc::new(SystemWallClock),
            node_memory: NodeMemoryMonitor::default(),
            leadership_shed,
            external_payload_min_bytes: 1024 * 1024,
            wal_backend: "memory",
            wal_disk: WalDiskMonitor::default(),
        }
    }

    pub fn leadership_shed_flag(&self) -> LeadershipShedFlag {
        self.leadership_shed.clone()
    }

    /// Replace the leadership-shed flag with one shared with the bootstrap
    /// health gates. They set per-gate bits on shed and clear their own bit on
    /// heal; the raft registry policy decides separately whether the node may
    /// campaign, shed current leaders, or accept inbound leadership transfer.
    pub fn with_leadership_shed_flag(mut self, flag: LeadershipShedFlag) -> Self {
        self.leadership_shed = flag;
        self
    }

    pub fn with_wall_clock(mut self, wall_clock: impl WallClock) -> Self {
        self.wall_clock = Arc::new(wall_clock);
        self
    }

    pub fn with_wall_clock_handle(mut self, wall_clock: Arc<dyn WallClock>) -> Self {
        self.wall_clock = wall_clock;
        self
    }

    pub fn with_external_payload_min_bytes(mut self, min_bytes: usize) -> Self {
        self.external_payload_min_bytes = min_bytes;
        self
    }

    /// Record the raft WAL backend so it appears in the metrics JSON.
    pub fn with_wal_backend(mut self, backend: &'static str) -> Self {
        self.wal_backend = backend;
        self
    }

    pub(crate) fn with_wal_disk_monitor(mut self, monitor: WalDiskMonitor) -> Self {
        self.wal_disk = monitor;
        self
    }

    pub(crate) fn wal_disk_monitor(&self) -> WalDiskMonitor {
        self.wal_disk.clone()
    }

    /// Apply runtime-level config (memory monitor, payload threshold) derived
    /// from the typed configuration.  Replaces the hard-coded defaults set by
    /// the constructors.
    pub fn with_runtime_config(mut self, config: &ursula_config::RuntimeConfig) -> Self {
        self.node_memory = NodeMemoryMonitor::new(config);
        if let Some(min_size) = &config.external_payload_min_size {
            self.external_payload_min_bytes = usize::try_from(min_size.as_bytes())
                .expect("config validation ensures payload size fits usize");
        }
        self
    }

    pub fn runtime(&self) -> &ShardRuntime {
        &self.runtime
    }

    pub fn raft_registry(&self) -> Option<&RaftGroupHandleRegistry> {
        self.raft_registry.as_ref()
    }

    pub fn client_write_router(&self) -> Option<&ClientWriteLeaderRouter> {
        self.client_write_router.as_ref()
    }

    pub fn unix_time_ms(&self) -> u64 {
        self.wall_clock.unix_time_ms()
    }
}

#[derive(Debug, Default)]
struct HttpMetrics {
    sse_streams_opened: AtomicU64,
    sse_read_iterations: AtomicU64,
    sse_data_events: AtomicU64,
    sse_control_events: AtomicU64,
    sse_error_events: AtomicU64,
}

impl HttpMetrics {
    fn snapshot(&self) -> HttpMetricsSnapshot {
        HttpMetricsSnapshot {
            sse_streams_opened: self.sse_streams_opened.load(Ordering::Relaxed),
            sse_read_iterations: self.sse_read_iterations.load(Ordering::Relaxed),
            sse_data_events: self.sse_data_events.load(Ordering::Relaxed),
            sse_control_events: self.sse_control_events.load(Ordering::Relaxed),
            sse_error_events: self.sse_error_events.load(Ordering::Relaxed),
        }
    }
}

#[derive(Debug, Clone, Copy, Default, serde::Serialize)]
struct HttpMetricsSnapshot {
    sse_streams_opened: u64,
    sse_read_iterations: u64,
    sse_data_events: u64,
    sse_control_events: u64,
    sse_error_events: u64,
}

/// Resolves the current leader of a raft group to a client-reachable base URL
/// so a write/read that lands on a non-leader can be answered with a 307
/// redirect. `peers` maps raft node id to that node's configured peer URL:
/// `server.listen` when `server.cluster_listen` is unset, or the separate
/// `server.cluster_listen` address when it is set. Peer URLs are also used as
/// HTTP leader-redirect targets, so clients and gateways must be able to reach
/// them too.
#[derive(Clone, Debug)]
pub struct ClientWriteLeaderRouter {
    peers: Arc<BTreeMap<u64, String>>,
    node_id: Option<u64>,
    per_group_voters: Arc<BTreeMap<RaftGroupId, BTreeSet<u64>>>,
}

impl ClientWriteLeaderRouter {
    pub fn new(peers: impl IntoIterator<Item = (u64, String)>) -> Self {
        Self::with_static_topology(None, peers, BTreeMap::new())
    }

    pub fn with_static_topology(
        node_id: impl Into<Option<u64>>,
        peers: impl IntoIterator<Item = (u64, String)>,
        per_group_voters: BTreeMap<RaftGroupId, BTreeSet<u64>>,
    ) -> Self {
        Self {
            peers: Arc::new(
                peers
                    .into_iter()
                    .map(|(node_id, url)| (node_id, url.trim_end_matches('/').to_owned()))
                    .collect(),
            ),
            node_id: node_id.into(),
            per_group_voters: Arc::new(per_group_voters),
        }
    }

    fn leader_base(&self, err: &RuntimeError) -> Option<(u64, String)> {
        let leader_hint = err.leader_hint()?;
        let leader_id = leader_hint.node_id?;
        let leader_base = self
            .peers
            .get(&leader_id)
            .or(leader_hint.address.as_ref())?;
        Some((leader_id, leader_base.trim_end_matches('/').to_owned()))
    }

    fn hosted_group_base(&self, err: &RuntimeError) -> Option<(u64, String)> {
        let RuntimeError::GroupNotHosted { raft_group_id, .. } = err else {
            return None;
        };
        let voters = self.per_group_voters.get(raft_group_id)?;
        voters
            .iter()
            .copied()
            .filter(|node_id| Some(*node_id) != self.node_id)
            .find_map(|node_id| {
                self.peers
                    .get(&node_id)
                    .map(|base| (node_id, base.trim_end_matches('/').to_owned()))
            })
    }

    fn redirect_response(&self, err: &RuntimeError, request_target: &str) -> Option<Response> {
        let (leader_id, leader_base) = self
            .leader_base(err)
            .or_else(|| self.hosted_group_base(err))?;
        let mut headers = HeaderMap::new();
        insert_default_response_headers(&mut headers);
        let leader_url = format!("{}{}", leader_base.trim_end_matches('/'), request_target);
        if let Ok(value) = HeaderValue::from_str(&leader_url) {
            headers.insert(LOCATION, value);
        } else {
            return None;
        }
        insert_u64_header(&mut headers, HEADER_URSULA_RAFT_LEADER_ID, leader_id);
        Some((StatusCode::TEMPORARY_REDIRECT, headers, err.to_string()).into_response())
    }
}

/// Process-wide RSS monitor. It reports RSS in `/__ursula/metrics` and exits
/// when the last-resort abort cap is exceeded. It does not reject writes;
/// ingress byte admission is the write-path memory control.
#[derive(Clone)]
pub struct NodeMemoryMonitor {
    abort_cap_bytes: Option<u64>,
    last_rss_bytes: Arc<AtomicU64>,
}

impl std::fmt::Debug for NodeMemoryMonitor {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("NodeMemoryMonitor")
            .field("abort_cap_bytes", &self.abort_cap_bytes)
            .field(
                "last_rss_bytes",
                &self.last_rss_bytes.load(Ordering::Relaxed),
            )
            .finish()
    }
}

impl Default for NodeMemoryMonitor {
    fn default() -> Self {
        Self {
            abort_cap_bytes: None,
            last_rss_bytes: Arc::new(AtomicU64::new(0)),
        }
    }
}

impl NodeMemoryMonitor {
    pub fn new(cfg: &ursula_config::RuntimeConfig) -> Self {
        let monitor = Self {
            abort_cap_bytes: cfg
                .node_memory_abort_cap_size
                .as_ref()
                .map(|s| s.as_bytes()),
            last_rss_bytes: Arc::new(AtomicU64::new(0)),
        };
        monitor.spawn_rss_sampler();
        monitor
    }

    pub fn last_rss_bytes(&self) -> u64 {
        self.last_rss_bytes.load(Ordering::Relaxed)
    }

    pub fn abort_cap_bytes(&self) -> Option<u64> {
        self.abort_cap_bytes
    }

    #[cfg(madsim)]
    fn spawn_rss_sampler(&self) {
        // Deterministic simulation: never report RSS.
    }

    #[cfg(not(madsim))]
    fn spawn_rss_sampler(&self) {
        let last_rss_bytes = self.last_rss_bytes.clone();
        let abort_cap = self.abort_cap_bytes;
        tokio::spawn(async move {
            loop {
                if let Some(rss) = read_proc_self_status_vm_rss_bytes() {
                    last_rss_bytes.store(rss, Ordering::Relaxed);
                    if let Some(cap) = abort_cap
                        && rss > cap
                    {
                        let host = std::env::var("HOSTNAME")
                            .ok()
                            .or_else(|| std::fs::read_to_string("/proc/sys/kernel/hostname").ok())
                            .map(|s| s.trim().to_string())
                            .unwrap_or_default();
                        let now_ms = std::time::SystemTime::now()
                            .duration_since(std::time::UNIX_EPOCH)
                            .map(|d| u64::try_from(d.as_millis()).unwrap_or(u64::MAX))
                            .unwrap_or(0);
                        let breadcrumb = serde_json::json!({
                            "event": "memory_abort_cap_exit",
                            "ts_ms": now_ms,
                            "host": host,
                            "rss_bytes": rss,
                            "abort_cap_bytes": cap,
                        })
                        .to_string();
                        tracing::error!("{breadcrumb}");
                        use std::io::Write as _;
                        let _ = std::io::stderr().flush();
                        std::process::abort();
                    }
                }
                tokio::time::sleep(Duration::from_millis(500)).await;
            }
        });
    }
}

#[cfg(not(madsim))]
fn read_proc_self_status_vm_rss_bytes() -> Option<u64> {
    // Linux-only: parse `VmRSS:    NNN kB` from /proc/self/status.
    let raw = std::fs::read_to_string("/proc/self/status").ok()?;
    for line in raw.lines() {
        if let Some(rest) = line.strip_prefix("VmRSS:") {
            let kb: u64 = rest.split_whitespace().next()?.parse().ok()?;
            return Some(kb.saturating_mul(1024));
        }
    }
    None
}

#[derive(Clone)]
struct HttpRaftGrpcService {
    raft: RaftGrpcService,
}

impl HttpRaftGrpcService {
    fn new(registry: RaftGroupHandleRegistry, state: HttpState) -> Self {
        let cold_store = state.runtime().cold_store();
        let leadership_shed = state.leadership_shed_flag();
        Self {
            raft: RaftGrpcService::new(registry)
                .with_cold_store(cold_store)
                .with_leadership_shed_flag(leadership_shed),
        }
    }
}

#[tonic::async_trait]
impl raft_internal_proto::raft_internal_server::RaftInternal for HttpRaftGrpcService {
    type AppendStreamStream =
        <RaftGrpcService as raft_internal_proto::raft_internal_server::RaftInternal>::AppendStreamStream;

    async fn append(
        &self,
        request: tonic::Request<raft_internal_proto::RaftRpcEnvelopeV1>,
    ) -> Result<tonic::Response<raft_internal_proto::RaftRpcAckV1>, tonic::Status> {
        raft_internal_proto::raft_internal_server::RaftInternal::append(&self.raft, request).await
    }

    async fn append_stream(
        &self,
        request: tonic::Request<tonic::Streaming<raft_internal_proto::RaftAppendStreamRequest>>,
    ) -> Result<tonic::Response<Self::AppendStreamStream>, tonic::Status> {
        raft_internal_proto::raft_internal_server::RaftInternal::append_stream(&self.raft, request)
            .await
    }

    async fn vote(
        &self,
        request: tonic::Request<raft_internal_proto::RaftRpcEnvelopeV1>,
    ) -> Result<tonic::Response<raft_internal_proto::RaftRpcAckV1>, tonic::Status> {
        raft_internal_proto::raft_internal_server::RaftInternal::vote(&self.raft, request).await
    }

    async fn full_snapshot(
        &self,
        request: tonic::Request<raft_internal_proto::RaftFullSnapshotRequestV1>,
    ) -> Result<tonic::Response<raft_internal_proto::RaftFullSnapshotAckV1>, tonic::Status> {
        raft_internal_proto::raft_internal_server::RaftInternal::full_snapshot(&self.raft, request)
            .await
    }

    async fn group_write(
        &self,
        request: tonic::Request<raft_internal_proto::GroupWriteRequestV1>,
    ) -> Result<tonic::Response<raft_internal_proto::GroupWriteResponseV1>, tonic::Status> {
        raft_internal_proto::raft_internal_server::RaftInternal::group_write(&self.raft, request)
            .await
    }

    async fn group_read(
        &self,
        request: tonic::Request<raft_internal_proto::GroupReadRequestV1>,
    ) -> Result<tonic::Response<raft_internal_proto::GroupReadResponseV1>, tonic::Status> {
        raft_internal_proto::raft_internal_server::RaftInternal::group_read(&self.raft, request)
            .await
    }

    async fn transfer_leader(
        &self,
        request: tonic::Request<raft_internal_proto::RaftTransferLeaderRequestV1>,
    ) -> Result<tonic::Response<raft_internal_proto::RaftTransferLeaderAckV1>, tonic::Status> {
        raft_internal_proto::raft_internal_server::RaftInternal::transfer_leader(
            &self.raft, request,
        )
        .await
    }
}

fn raft_grpc_service(
    state: HttpState,
    registry: RaftGroupHandleRegistry,
) -> raft_internal_proto::raft_internal_server::RaftInternalServer<HttpRaftGrpcService> {
    raft_internal_proto::raft_internal_server::RaftInternalServer::new(HttpRaftGrpcService::new(
        registry, state,
    ))
    .accept_compressed(tonic::codec::CompressionEncoding::Zstd)
    .max_decoding_message_size(RAFT_GRPC_MAX_MESSAGE_BYTES)
    .max_encoding_message_size(RAFT_GRPC_MAX_MESSAGE_BYTES)
}

pub fn router(runtime: ShardRuntime) -> Router {
    let state = HttpState::new(runtime);
    cluster_router_from_state(state.clone())
        .merge(admin_ops_router(state.clone()))
        .merge(client_router_with_admission(
            state,
            IngressAdmission::default(),
        ))
}

pub fn router_with_raft_registry(
    runtime: ShardRuntime,
    raft_registry: RaftGroupHandleRegistry,
) -> Router {
    let state = HttpState::with_raft_registry(runtime, raft_registry);
    cluster_router_from_state(state.clone()).merge(client_router_with_admission(
        state,
        IngressAdmission::default(),
    ))
}

pub fn router_with_static_raft_cluster(
    runtime: ShardRuntime,
    raft_registry: RaftGroupHandleRegistry,
    peers: impl IntoIterator<Item = (u64, String)>,
) -> Router {
    let state = HttpState::with_static_raft_cluster(runtime, raft_registry, peers);
    cluster_router_from_state(state.clone()).merge(client_router_with_admission(
        state,
        IngressAdmission::default(),
    ))
}

pub fn router_with_static_raft_cluster_topology(
    runtime: ShardRuntime,
    raft_registry: RaftGroupHandleRegistry,
    node_id: u64,
    peers: impl IntoIterator<Item = (u64, String)>,
    per_group_voters: BTreeMap<RaftGroupId, BTreeSet<u64>>,
) -> Router {
    let state = HttpState::with_static_raft_cluster_topology(
        runtime,
        raft_registry,
        Some(node_id),
        peers,
        per_group_voters,
    );
    cluster_router_from_state(state.clone())
        .merge(admin_ops_router(state.clone()))
        .merge(client_router_with_admission(
            state,
            IngressAdmission::default(),
        ))
}

/// Convenience wrapper that merges the client, cluster, and admin planes into
/// a single router.  Used by in-process tests and the madsim harness, where
/// per-plane listeners would only add noise.
pub fn router_with_http_state(state: HttpState) -> Router {
    cluster_router_from_state(state.clone())
        .merge(admin_ops_router(state.clone()))
        .merge(client_router_with_admission(
            state,
            IngressAdmission::default(),
        ))
}

/// Admin-plane routes: the mutating operator surface (raft group operations,
/// maintenance drain, cold-flush trigger) plus read-only metrics so operator
/// tooling works over a single tunnel. Production binds this to
/// `server.admin_listen` (loopback by default) — nodes expose no
/// cluster-mutation endpoints on the client or cluster planes.
pub fn admin_router(state: HttpState) -> Router {
    admin_ops_router(state.clone()).merge(
        Router::new()
            .route("/__ursula/metrics", get(metrics))
            .route("/__ursula/usage", get(bucket_usage))
            .route(
                "/__ursula/purge/{bucket}",
                axum::routing::delete(purge_bucket),
            )
            .route("/__ursula/quota/{bucket}", put(set_bucket_quota))
            .with_state(state),
    )
}

/// The mutating admin routes without the metrics alias. The single-router
/// convenience mergers use this directly because the client plane already
/// serves `/__ursula/metrics`.
fn admin_ops_router(state: HttpState) -> Router {
    let router = Router::new()
        .route(
            "/__ursula/flush-cold/{bucket}/{stream}",
            post(flush_cold_stream),
        )
        .route(
            "/__ursula/flush-cold/{bucket}/{affinity}/{stream}",
            post(flush_cold_stream),
        )
        .route("/__ursula/backup/info", get(backup_info))
        .route(
            "/__ursula/backup/group/{raft_group_id}",
            get(export_backup_group),
        )
        .route(
            "/__ursula/backup/group/{raft_group_id}/import",
            post(import_backup_group),
        )
        .route(
            "/__ursula/raft/{raft_group_id}/snapshot",
            post(trigger_raft_snapshot),
        )
        .route(
            "/__ursula/raft/{raft_group_id}/purge",
            post(trigger_raft_purge),
        )
        .route(
            "/__ursula/raft/{raft_group_id}/membership",
            post(change_raft_membership),
        )
        .route(
            "/__ursula/raft/{raft_group_id}/learners/{node_id}",
            post(add_raft_learner),
        )
        .route(
            "/__ursula/raft/quiesce-for-restart",
            post(quiesce_raft_for_restart),
        )
        .route(
            "/__ursula/raft/{raft_group_id}/leader/transfer/{node_id}",
            post(transfer_raft_leader),
        )
        .route(
            "/__ursula/leadership-shed/maintenance",
            post(mark_maintenance_drain).delete(clear_maintenance_drain),
        );
    #[cfg(feature = "jemalloc-prof")]
    let router = router.route("/__ursula/debug/heap-profile", get(heap_profile));
    router
        .layer(DefaultBodyLimit::max(MAX_HTTP_BODY_BYTES))
        .with_state(state)
}

/// Cluster-plane routes: inter-node gRPC carrying Raft RPCs, snapshot
/// transfer, and leader-read checks. In a dual-listener deployment these bind
/// to the private (VPC) interface so chaos applied to the public face never
/// disrupts consensus.
pub fn cluster_router_from_state(state: HttpState) -> Router {
    let raft_registry = state.raft_registry.clone().unwrap_or_default();
    Router::new()
        .route_service(
            RAFT_GRPC_APPEND_PATH,
            raft_grpc_service(state.clone(), raft_registry.clone()),
        )
        .route_service(
            RAFT_GRPC_APPEND_STREAM_PATH,
            raft_grpc_service(state.clone(), raft_registry.clone()),
        )
        .route_service(
            RAFT_GRPC_VOTE_PATH,
            raft_grpc_service(state.clone(), raft_registry.clone()),
        )
        .route_service(
            RAFT_GRPC_FULL_SNAPSHOT_PATH,
            raft_grpc_service(state.clone(), raft_registry.clone()),
        )
        .route_service(
            RAFT_GRPC_GROUP_WRITE_PATH,
            raft_grpc_service(state.clone(), raft_registry.clone()),
        )
        .route_service(
            RAFT_GRPC_GROUP_READ_PATH,
            raft_grpc_service(state.clone(), raft_registry.clone()),
        )
        .route_service(
            RAFT_GRPC_TRANSFER_LEADER_PATH,
            raft_grpc_service(state.clone(), raft_registry),
        )
        .route(LEADERSHIP_SHED_PATH, get(leadership_shed_status))
        .layer(DefaultBodyLimit::max(MAX_HTTP_BODY_BYTES))
        .with_state(state)
}

/// Client-plane ingress admission. The primary control is a process-wide
/// in-flight write-body byte budget: a request must reserve its body bytes
/// before axum drains the body into `Bytes`, and the reservation lives until
/// the response is produced. This is the layer that turns memory pressure into
/// backpressure at the HTTP edge.
///
/// Configurable via `ServerConfig.http_inflight_body_size`.
#[derive(Clone)]
pub struct IngressAdmission {
    body_bytes: Arc<tokio::sync::Semaphore>,
    wal_disk: WalDiskMonitor,
}

impl Default for IngressAdmission {
    fn default() -> Self {
        Self {
            body_bytes: Arc::new(tokio::sync::Semaphore::new(
                DEFAULT_HTTP_INFLIGHT_BODY_BYTES,
            )),
            wal_disk: WalDiskMonitor::default(),
        }
    }
}

impl IngressAdmission {
    pub fn new(cfg: &ursula_config::ServerConfig) -> Self {
        let body_budget = cfg.http_inflight_body_size.as_bytes() as usize;
        Self {
            body_bytes: Arc::new(tokio::sync::Semaphore::new(body_budget)),
            wal_disk: WalDiskMonitor::default(),
        }
    }

    pub fn disabled() -> Self {
        Self {
            body_bytes: Arc::new(tokio::sync::Semaphore::new(usize::MAX)),
            wal_disk: WalDiskMonitor::default(),
        }
    }

    pub(crate) fn with_wal_disk_monitor(mut self, monitor: WalDiskMonitor) -> Self {
        self.wal_disk = monitor;
        self
    }
}

async fn ingress_admission_middleware(
    State(admission): State<IngressAdmission>,
    request: Request<Body>,
    next: Next,
) -> Response {
    if request.uri().path() == CLUSTER_PROBE_PATH {
        return next.run(request).await;
    }
    let Some(body_bytes) = request_write_body_bytes(&request) else {
        return next.run(request).await;
    };
    if body_bytes > u64::try_from(MAX_HTTP_BODY_BYTES).expect("max body bytes fits u64") {
        return (StatusCode::PAYLOAD_TOO_LARGE, "request body is too large").into_response();
    }
    if admission.wal_disk.is_pressured() {
        return retry_after_json("WalDiskPressure");
    }

    let _body_permits = if body_bytes > 0 {
        let Ok(permits) = u32::try_from(body_bytes) else {
            return (StatusCode::PAYLOAD_TOO_LARGE, "request body is too large").into_response();
        };
        match admission.body_bytes.clone().try_acquire_many_owned(permits) {
            Ok(permit) => Some(permit),
            Err(_) => return retry_after_json("IngressBodyBytesLimitReached"),
        }
    } else {
        None
    };

    next.run(request).await
}

async fn path_affinity_extension_middleware(request: Request<Body>, next: Next) -> Response {
    let transaction_path = is_group_append_transaction_uri(request.uri());
    let affinity_path = transaction_path || is_path_affinity_uri(request.uri());
    let mut response = next.run(request).await;
    if affinity_path {
        insert_extension_token(response.headers_mut(), PATH_AFFINITY_EXTENSION);
    }
    if transaction_path {
        insert_extension_token(response.headers_mut(), GROUP_APPEND_TRANSACTION_EXTENSION);
    }
    response
}

fn is_group_append_transaction_uri(uri: &Uri) -> bool {
    let segments = uri
        .path()
        .split('/')
        .filter(|segment| !segment.is_empty())
        .collect::<Vec<_>>();
    matches!(segments.as_slice(), [bucket, _affinity, "$transaction"] if !bucket.starts_with("__ursula"))
}

fn is_path_affinity_uri(uri: &Uri) -> bool {
    let mut segments = uri.path().split('/').filter(|segment| !segment.is_empty());
    let Some(bucket) = segments.next() else {
        return false;
    };
    let Some(_affinity) = segments.next() else {
        return false;
    };
    let Some(stream) = segments.next() else {
        return false;
    };
    !bucket.starts_with("__ursula") && !is_reserved_affinity_stream_id(stream)
}

fn request_write_body_bytes(request: &Request<Body>) -> Option<u64> {
    if !is_write_method(request.method()) {
        return None;
    }
    if let Some(content_length) = request.headers().get(CONTENT_LENGTH)
        && let Ok(content_length) = content_length.to_str()
        && let Ok(parsed) = content_length.parse::<u64>()
    {
        return Some(parsed);
    }
    let size_hint = request.body().size_hint();
    size_hint.exact().or_else(|| size_hint.upper()).or(Some(
        u64::try_from(MAX_HTTP_BODY_BYTES).expect("max body bytes fits u64"),
    ))
}

fn is_write_method(method: &Method) -> bool {
    matches!(
        *method,
        Method::POST | Method::PUT | Method::PATCH | Method::DELETE
    )
}

fn retry_after_json(error: &'static str) -> Response {
    let mut headers = HeaderMap::new();
    insert_default_response_headers(&mut headers);
    headers.insert(
        axum::http::header::RETRY_AFTER,
        HeaderValue::from_static("1"),
    );
    headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
    (
        StatusCode::SERVICE_UNAVAILABLE,
        headers,
        serde_json::json!({ "error": error }).to_string(),
    )
        .into_response()
}

/// Builds a `200`-style JSON response with the `application/json` content type.
/// Centralizes the content-type wiring so handlers returning a JSON body share
/// one consistent style.
fn json_response(status: StatusCode, body: String) -> Response {
    (status, [(CONTENT_TYPE, "application/json")], body).into_response()
}

/// Path of the cluster egress-health probe (M2). Peers POST a payload here over
/// the cluster plane; the round-trip time exposes loss/delay on the sender's
/// egress, which a small heartbeat-sized request would mask.
pub(crate) const CLUSTER_PROBE_PATH: &str = "/__ursula/cluster-probe";
pub(crate) const LEADERSHIP_SHED_PATH: &str = "/__ursula/leadership-shed";
pub(crate) const READINESS_PATH: &str = "/__ursula/ready";

/// Probe target: drain the body (so the sender's full egress traverses the
/// cluster plane) and answer 200. Bypasses ingress admission.
async fn cluster_probe(_body: Bytes) -> StatusCode {
    StatusCode::OK
}

async fn readiness(State(state): State<HttpState>) -> Response {
    let disk = state.wal_disk.snapshot();
    let status = if disk.pressure {
        StatusCode::SERVICE_UNAVAILABLE
    } else {
        StatusCode::OK
    };
    json_response(
        status,
        serde_json::json!({
            "ready": !disk.pressure,
            "wal_disk_pressure": disk.pressure,
            "wal_available_bytes": disk.available_bytes,
            "wal_min_available_bytes": disk.min_available_bytes,
            "wal_resume_available_bytes": disk.resume_available_bytes,
            "wal_disk_stat_errors": disk.stat_errors,
        })
        .to_string(),
    )
}

async fn leadership_shed_status(State(state): State<HttpState>) -> Response {
    let shed_state = state
        .raft_registry()
        .map(RaftGroupHandleRegistry::leadership_shed_state)
        .unwrap_or_default();
    let body = serde_json::json!({
        "bits": shed_state.bits(),
        "state": shed_state.to_string(),
        "should_accept_transfer": shed_state.should_accept_transfer(),
        "should_campaign": shed_state.should_campaign(),
        "should_shed_current_leaders": shed_state.should_shed_current_leaders(),
    })
    .to_string();
    json_response(StatusCode::OK, body)
}

async fn mark_maintenance_drain(State(state): State<HttpState>) -> Response {
    let Some(registry) = state.raft_registry() else {
        return (
            StatusCode::BAD_REQUEST,
            "raft registry is not configured for this server",
        )
            .into_response();
    };
    registry.mark_leadership_shed(LeadershipShedReason::MaintenanceDrain);
    leadership_shed_status(State(state)).await
}

async fn clear_maintenance_drain(State(state): State<HttpState>) -> Response {
    let Some(registry) = state.raft_registry() else {
        return (
            StatusCode::BAD_REQUEST,
            "raft registry is not configured for this server",
        )
            .into_response();
    };
    registry.clear_leadership_shed(LeadershipShedReason::MaintenanceDrain);
    leadership_shed_status(State(state)).await
}

async fn quiesce_raft_for_restart(State(state): State<HttpState>) -> Response {
    let Some(registry) = state.raft_registry() else {
        return (
            StatusCode::BAD_REQUEST,
            "raft registry is not configured for this server",
        )
            .into_response();
    };
    if !registry
        .leadership_shed_state()
        .contains(ursula_raft::LeadershipShedState::MAINTENANCE_DRAIN)
    {
        return (
            StatusCode::CONFLICT,
            "maintenance drain must be active before Raft restart quiescence",
        )
            .into_response();
    }
    let groups = registry.metrics_snapshot();
    let Some(node_id) = groups.first().map(|group| group.node_id) else {
        return (StatusCode::CONFLICT, "no local Raft groups are registered").into_response();
    };
    let led_groups = groups
        .into_iter()
        .filter(|group| group.current_leader == Some(node_id))
        .map(|group| group.raft_group_id)
        .collect::<Vec<_>>();
    if !led_groups.is_empty() {
        return json_response(
            StatusCode::CONFLICT,
            serde_json::json!({
                "quiesced": false,
                "node_id": node_id,
                "reason": "node still leads raft groups",
                "raft_group_ids": led_groups,
            })
            .to_string(),
        );
    }
    match registry.quiesce_for_restart().await {
        Ok(group_count) => json_response(
            StatusCode::OK,
            serde_json::json!({
                "quiesced": true,
                "node_id": node_id,
                "raft_group_count": group_count,
            })
            .to_string(),
        ),
        Err(err) => (
            StatusCode::INTERNAL_SERVER_ERROR,
            format!("quiesce raft groups for restart: {err}"),
        )
            .into_response(),
    }
}

pub fn client_router_with_admission(state: HttpState, admission: IngressAdmission) -> Router {
    let finite_record_response =
        |_status: StatusCode,
         _version: Version,
         headers: &HeaderMap,
         _extensions: &axum::http::Extensions| {
            headers
                .get(CONTENT_TYPE)
                .and_then(|value| value.to_str().ok())
                .is_some_and(|content_type| {
                    let media_type = content_type
                        .split(';')
                        .next()
                        .unwrap_or(content_type)
                        .trim();
                    media_type == "application/json"
                        || media_type == "application/x-ndjson"
                        || media_type == "application/vnd.durable-stream-records+ndjson"
                })
        };
    let response_compression = CompressionLayer::new()
        .gzip(true)
        .quality(CompressionLevel::Fastest)
        .compress_when(SizeAbove::new(256).and(finite_record_response));

    Router::new()
        .route("/__ursula/metrics", get(metrics))
        .route(READINESS_PATH, get(readiness))
        .route("/__ursula/usage", get(bucket_usage))
        .route(
            "/__ursula/purge/{bucket}",
            axum::routing::delete(purge_bucket),
        )
        .route("/__ursula/quota/{bucket}", put(set_bucket_quota))
        .route(CLUSTER_PROBE_PATH, post(cluster_probe))
        .route("/{bucket}", put(create_bucket))
        .route(
            "/{bucket}/{stream}/snapshot",
            get(read_latest_snapshot).put(publish_snapshot_at_record),
        )
        .route(
            "/{bucket}/{stream}/snapshot/{snapshot_offset}",
            put(publish_snapshot)
                .get(read_snapshot)
                .delete(delete_snapshot),
        )
        .route(
            "/{bucket}/{stream}/retention",
            put(advance_retention_at_record),
        )
        .route(
            "/{bucket}/{stream}/retention/{retained_offset}",
            put(advance_retention),
        )
        .route("/{bucket}/{stream}/bootstrap", get(bootstrap_stream))
        .route(
            "/{bucket}/{stream}/attrs",
            put(update_stream_attrs).get(get_stream_attrs),
        )
        .route(
            "/{bucket}/{stream}",
            put(create_stream)
                .post(append_stream)
                .get(read_stream)
                .delete(delete_stream)
                .head(head_stream),
        )
        .route("/{bucket}/{stream}/append-batch", post(append_batch))
        .route(
            "/{bucket}/{affinity}/$transaction",
            post(append_transaction),
        )
        .route(
            "/{bucket}/{affinity}/{stream}/snapshot",
            get(read_latest_snapshot).put(publish_snapshot_at_record),
        )
        .route(
            "/{bucket}/{affinity}/{stream}/snapshot/{snapshot_offset}",
            put(publish_snapshot)
                .get(read_snapshot)
                .delete(delete_snapshot),
        )
        .route(
            "/{bucket}/{affinity}/{stream}/retention",
            put(advance_retention_at_record),
        )
        .route(
            "/{bucket}/{affinity}/{stream}/retention/{retained_offset}",
            put(advance_retention),
        )
        .route(
            "/{bucket}/{affinity}/{stream}/bootstrap",
            get(bootstrap_stream),
        )
        .route(
            "/{bucket}/{affinity}/{stream}/attrs",
            put(update_stream_attrs).get(get_stream_attrs),
        )
        .route(
            "/{bucket}/{affinity}/{stream}",
            put(create_stream)
                .post(append_stream)
                .get(read_stream)
                .delete(delete_stream)
                .head(head_stream),
        )
        .route(
            "/{bucket}/{affinity}/{stream}/append-batch",
            post(append_batch),
        )
        .layer(DefaultBodyLimit::max(MAX_HTTP_BODY_BYTES))
        .layer(middleware::from_fn(path_affinity_extension_middleware))
        .layer(middleware::from_fn_with_state(
            admission,
            ingress_admission_middleware,
        ))
        .layer(response_compression)
        .with_state(state)
}

pub(crate) fn should_externalize_payload(
    state: &HttpState,
    payload_len: usize,
    allowed: bool,
) -> bool {
    allowed
        && payload_len > 0
        && state.runtime.has_cold_store()
        && payload_len >= state.external_payload_min_bytes
}

pub(crate) async fn stage_external_payload(
    state: &HttpState,
    stream_id: &BucketStreamId,
    payload: &[u8],
) -> Result<ExternalPayloadRef, Response> {
    let Some(cold_store) = state.runtime.cold_store() else {
        return Err((
            StatusCode::SERVICE_UNAVAILABLE,
            "cold backend must be configured before externalizing payloads",
        )
            .into_response());
    };
    let s3_path = new_external_payload_path(stream_id);
    let object_size = cold_store
        .write_chunk(&s3_path, payload)
        .await
        .map_err(|err| {
            (
                StatusCode::BAD_GATEWAY,
                format!("write external payload object: {err}"),
            )
                .into_response()
        })?;
    Ok(ExternalPayloadRef {
        s3_path,
        payload_len: u64::try_from(payload.len()).expect("payload len fits u64"),
        object_size,
    })
}

pub(crate) async fn cleanup_external_payload(state: &HttpState, s3_path: &str) {
    let Some(cold_store) = state.runtime.cold_store() else {
        return;
    };
    let _ = cold_store.delete_chunk(s3_path).await;
}

pub(crate) fn create_stream_http_response(input: CreateStreamHttpResponseInput<'_>) -> Response {
    let CreateStreamHttpResponseInput {
        response,
        stream_id,
        content_type,
        stream_ttl_seconds,
        stream_expires_at_ms,
        producer,
    } = input;
    let mut headers = HeaderMap::new();
    insert_default_response_headers(&mut headers);
    insert_content_type(&mut headers, content_type);
    insert_offset(&mut headers, response.next_offset);
    insert_location(&mut headers, stream_id);
    insert_lifetime_headers(&mut headers, stream_ttl_seconds, stream_expires_at_ms);
    insert_producer_ack(&mut headers, producer);
    if let Some(record_range) = response.record_range {
        insert_record_operation_headers(&mut headers, record_range);
    }
    if response.closed {
        insert_static(&mut headers, HEADER_STREAM_CLOSED, "true");
    }
    let status = if response.already_exists {
        StatusCode::OK
    } else {
        StatusCode::CREATED
    };
    (status, headers).into_response()
}

pub(crate) fn append_http_response(response: AppendResponse) -> Response {
    let mut headers = HeaderMap::new();
    insert_default_response_headers(&mut headers);
    insert_offset(&mut headers, response.next_offset);
    insert_producer_ack(&mut headers, response.producer.as_ref());
    if let Some(record_range) = response.record_range {
        insert_record_operation_headers(&mut headers, record_range);
    }
    if response.closed {
        insert_static(&mut headers, HEADER_STREAM_CLOSED, "true");
    }
    let status = if response.producer.is_some() && !response.deduplicated {
        StatusCode::OK
    } else {
        StatusCode::NO_CONTENT
    };
    (status, headers).into_response()
}

/// Administrator-triggered tenant offboarding (#150): purges the bucket from
/// every Raft group, then runs one cold-GC pass so the enqueued cold-object
/// prefixes are reclaimed before the report returns. Idempotent — purging an
/// absent bucket returns the same report shape with zero counts, and a
/// crashed purge converges on re-run because cold reclamation is
/// list-then-delete over object prefixes.
pub(crate) async fn purge_bucket(
    State(state): State<HttpState>,
    Path(bucket): Path<String>,
) -> Response {
    // Packs written before bucket erasure domains may contain several
    // tenants. Rewrite a bounded number of their live slices on every retry
    // and do not remove the target bucket until the global legacy debt reaches
    // zero. Control retries this idempotent request, so large migrations
    // converge without exceeding its claim lease. This compatibility pass may
    // be removed only after every supported snapshot has zero shared chunks
    // outside `{bucket}/_packs/` and the oldest deployable writer uses the
    // bucket-scoped pack path.
    let legacy = match state
        .runtime
        .migrate_legacy_shared_cold_once(LEGACY_SHARED_MIGRATION_MAX_CHUNKS, 0)
        .await
    {
        Ok(report) => report,
        Err(err) => {
            let target = format!("/__ursula/purge/{bucket}");
            return runtime_error_or_leader_redirect_async(&state, err, &target).await;
        }
    };
    if legacy.pending_chunks > 0 {
        return axum::Json(serde_json::json!({
            "bucket": bucket,
            "removed_streams": 0,
            "groups_with_streams": [],
            "cold_gc_entries_reclaimed": 0,
            "cold_gc_pending_entries": legacy.pending_chunks,
            "cold_gc_complete": false,
            "cold_gc_error": null,
            "bucket_prefix_absent": false,
            "legacy_shared_chunks_pending": legacy.pending_chunks,
        }))
        .into_response();
    }
    let report = match state.runtime.purge_bucket_all_groups(&bucket).await {
        Ok(report) => report,
        Err(err) => {
            let target = format!("/__ursula/purge/{bucket}");
            return runtime_error_or_leader_redirect_async(&state, err, &target).await;
        }
    };
    // Reclaim the just-enqueued cold prefixes now instead of waiting for the
    // background worker's next pass. Failures leave entries queued for the
    // worker; the purge itself is already durable.
    let (cold_gc_reclaimed, mut cold_gc_error) = match state
        .runtime
        .run_cold_gc_all_groups_once(COLD_GC_PURGE_BATCH_MAX_ENTRIES)
        .await
    {
        Ok(reclaimed) => (reclaimed, None),
        Err(err) => {
            tracing::warn!(
                bucket = %bucket,
                error = %err,
                "cold GC pass after purge failed; background worker will finish reclamation"
            );
            (0, Some(err.to_string()))
        }
    };
    // A second idempotent purge is a linearized read of every group's durable
    // queue after reclamation. The number reclaimed by one pass is only
    // diagnostic: delayed entries, a partial failure, or another leader's
    // background worker can all make it zero without proving cold absence.
    let proof = match state.runtime.purge_bucket_all_groups(&bucket).await {
        Ok(report) => report,
        Err(err) => {
            let target = format!("/__ursula/purge/{bucket}");
            return runtime_error_or_leader_redirect_async(&state, err, &target).await;
        }
    };
    let bucket_prefix_absent = if proof.pending_cold_gc_entries == 0 && cold_gc_error.is_none() {
        match state
            .runtime
            .erase_bucket_cold_prefix_and_prove(&bucket)
            .await
        {
            Ok(()) => true,
            Err(err) => {
                cold_gc_error = Some(err.to_string());
                false
            }
        }
    } else {
        false
    };
    let cold_gc_complete =
        proof.pending_cold_gc_entries == 0 && cold_gc_error.is_none() && bucket_prefix_absent;
    axum::Json(serde_json::json!({
        "bucket": bucket,
        "removed_streams": report.removed_streams,
        "groups_with_streams": report.groups_with_streams,
        "cold_gc_entries_reclaimed": cold_gc_reclaimed,
        "cold_gc_pending_entries": proof.pending_cold_gc_entries,
        "cold_gc_complete": cold_gc_complete,
        "cold_gc_error": cold_gc_error,
        "bucket_prefix_absent": bucket_prefix_absent,
    }))
    .into_response()
}

const COLD_GC_PURGE_BATCH_MAX_ENTRIES: usize = 4096;
const LEGACY_SHARED_MIGRATION_MAX_CHUNKS: usize = 32;

pub(crate) async fn create_bucket(Path(_bucket): Path<String>) -> Response {
    StatusCode::CREATED.into_response()
}

/// Versioned, self-described per-bucket usage summed across this node's Raft
/// groups. Served from local replica state; consumers tolerate replication
/// lag and validate the contract before interpreting derived counters.
pub(crate) async fn bucket_usage(State(state): State<HttpState>) -> Response {
    match state.runtime.bucket_usage_all_groups().await {
        Ok(report) => {
            let buckets = report
                .into_iter()
                .map(|entry| {
                    (
                        entry.bucket_id,
                        serde_json::json!({
                            "committed_append_bytes": entry.usage.committed_append_bytes,
                            "committed_records": entry.usage.committed_records,
                            "committed_write_units": entry.usage.committed_write_units,
                            "retained_bytes": entry.usage.retained_bytes,
                            "stream_count": entry.usage.stream_count,
                        }),
                    )
                })
                .collect::<serde_json::Map<_, _>>();
            axum::Json(serde_json::json!({
                "version": 1,
                "write_unit_bytes": ursula_runtime::COMMITTED_WRITE_UNIT_BYTES,
                "buckets": buckets,
            }))
            .into_response()
        }
        Err(err) => (
            StatusCode::INTERNAL_SERVER_ERROR,
            format!("bucket usage read failed: {err}"),
        )
            .into_response(),
    }
}

#[derive(Debug, Default, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub(crate) struct BucketQuotaBody {
    #[serde(default)]
    max_streams: Option<u64>,
    #[serde(default)]
    max_retained_bytes: Option<u64>,
}

/// Sets or clears one bucket's data-plane quota, replicated to every Raft
/// group. An empty or omitted body clears the record. Limits are per-group
/// backstops: the cluster-wide bound is `limit × group_count`; exact
/// tenant-level enforcement lives at the gateway.
pub(crate) async fn set_bucket_quota(
    State(state): State<HttpState>,
    Path(bucket): Path<String>,
    body: Option<axum::Json<BucketQuotaBody>>,
) -> Response {
    let body = body.map(|axum::Json(body)| body).unwrap_or_default();
    match state
        .runtime
        .set_bucket_quota_all_groups(&bucket, body.max_streams, body.max_retained_bytes)
        .await
    {
        Ok(()) => StatusCode::NO_CONTENT.into_response(),
        Err(err) => {
            let status = crate::render::runtime_error_status(&err);
            (status, format!("bucket quota update failed: {err}")).into_response()
        }
    }
}

pub(crate) async fn metrics(State(state): State<HttpState>) -> Response {
    let raft_groups = state
        .raft_registry()
        .map(RaftGroupHandleRegistry::metrics_snapshot)
        .unwrap_or_default();
    let mut body = render_metrics(
        state.runtime.metrics().snapshot(),
        state.runtime.mailbox_snapshot(),
        state.http_metrics.snapshot(),
        &raft_groups,
        state.runtime.cold_store_info().as_ref(),
    );
    // Splice process-level memory observability onto the metrics JSON so
    // a chaos node's RSS trajectory is visible from any HTTP client (the
    // status-publishing pipeline survives SSM exec failures).
    let rss = state.node_memory.last_rss_bytes();
    let cap = state.node_memory.abort_cap_bytes().unwrap_or_default();
    if let Some(object) = body.as_object_mut() {
        object.insert("process_rss_bytes".to_owned(), serde_json::json!(rss));
        object.insert(
            "node_memory_abort_cap_bytes".to_owned(),
            serde_json::json!(cap),
        );
        object.insert(
            "wal_backend".to_owned(),
            serde_json::json!(state.wal_backend),
        );
        let wal_disk = state.wal_disk.snapshot();
        object.insert(
            "wal_available_bytes".to_owned(),
            serde_json::json!(wal_disk.available_bytes),
        );
        object.insert(
            "wal_min_available_bytes".to_owned(),
            serde_json::json!(wal_disk.min_available_bytes),
        );
        object.insert(
            "wal_resume_available_bytes".to_owned(),
            serde_json::json!(wal_disk.resume_available_bytes),
        );
        object.insert(
            "wal_disk_pressure".to_owned(),
            serde_json::json!(wal_disk.pressure),
        );
        object.insert(
            "wal_disk_stat_errors".to_owned(),
            serde_json::json!(wal_disk.stat_errors),
        );
    }
    json_response(StatusCode::OK, body.to_string())
}

#[cfg(feature = "jemalloc-prof")]
pub(crate) async fn heap_profile(headers: HeaderMap) -> Response {
    if let Err(response) = authorize_debug_endpoint(&headers) {
        return *response;
    }

    let profile = tokio::task::spawn_blocking(dump_jemalloc_heap_profile).await;
    let bytes = match profile {
        Ok(Ok(bytes)) => bytes,
        Ok(Err(HeapProfileError::Disabled(message))) => {
            return json_response(
                StatusCode::CONFLICT,
                serde_json::json!({
                    "error": "heap_profile_unavailable",
                    "message": message,
                    "required_build_feature": "jemalloc-prof",
                    "required_malloc_conf":
                        format!("{MALLOC_CONF_ENV_VAR}=prof:true,prof_active:true,lg_prof_sample:19"),
                })
                .to_string(),
            );
        }
        Ok(Err(HeapProfileError::Io(message))) => {
            return json_response(
                StatusCode::INTERNAL_SERVER_ERROR,
                serde_json::json!({
                    "error": "heap_profile_io_failed",
                    "message": message,
                })
                .to_string(),
            );
        }
        Err(err) => {
            return json_response(
                StatusCode::INTERNAL_SERVER_ERROR,
                serde_json::json!({
                    "error": "heap_profile_task_failed",
                    "message": err.to_string(),
                })
                .to_string(),
            );
        }
    };

    let mut response_headers = HeaderMap::new();
    insert_default_response_headers(&mut response_headers);
    response_headers.insert(
        CONTENT_TYPE,
        HeaderValue::from_static("application/octet-stream"),
    );
    response_headers.insert(
        CONTENT_DISPOSITION,
        HeaderValue::from_static("attachment; filename=\"ursula-heap.heap\""),
    );
    (StatusCode::OK, response_headers, bytes).into_response()
}

#[cfg(feature = "jemalloc-prof")]
fn authorize_debug_endpoint(headers: &HeaderMap) -> Result<(), BoxResponse> {
    // Any failure answers with the router's plain 404 so unauthenticated
    // probes cannot tell this endpoint apart from an unknown path.
    let expected = match std::env::var("URSULA_DEBUG_TOKEN") {
        Ok(token) if !token.is_empty() => token,
        _ => return Err(Box::new(StatusCode::NOT_FOUND.into_response())),
    };

    let authorized = headers
        .get(HEADER_URSULA_DEBUG_TOKEN)
        .and_then(|value| value.to_str().ok())
        .is_some_and(|actual| constant_time_str_eq(actual, &expected));
    if authorized {
        Ok(())
    } else {
        Err(Box::new(StatusCode::NOT_FOUND.into_response()))
    }
}

// Token comparison must not short-circuit on the first mismatching byte;
// this route is reachable through the gateway, so response timing is
// attacker-observable. Only the length may leak.
#[cfg(feature = "jemalloc-prof")]
fn constant_time_str_eq(a: &str, b: &str) -> bool {
    let a = a.as_bytes();
    let b = b.as_bytes();
    if a.len() != b.len() {
        return false;
    }
    a.iter()
        .zip(b.iter())
        .fold(0u8, |acc, (x, y)| acc | (x ^ y))
        == 0
}

#[cfg(feature = "jemalloc-prof")]
enum HeapProfileError {
    /// Profiling cannot produce data under the current build or runtime
    /// configuration; the response carries remediation hints.
    Disabled(String),
    /// The dump itself failed for reasons unrelated to configuration.
    Io(String),
}

#[cfg(feature = "jemalloc-prof")]
fn dump_jemalloc_heap_profile() -> Result<Vec<u8>, HeapProfileError> {
    let profiling_enabled = tikv_jemalloc_ctl::profiling::prof::read()
        .map_err(|err| HeapProfileError::Io(format!("read jemalloc opt.prof: {err}")))?;
    if !profiling_enabled {
        return Err(HeapProfileError::Disabled(format!(
            "jemalloc profiling is disabled; restart with \
             {MALLOC_CONF_ENV_VAR}=prof:true,prof_active:true"
        )));
    }

    // Dump into a fresh mode-0700 temp directory: a fixed path in
    // world-writable /tmp would be symlink-attackable, readable by other
    // local users, and shared between co-located nodes.
    let dump_dir = tempfile::tempdir()
        .map_err(|err| HeapProfileError::Io(format!("create heap profile dir: {err}")))?;
    let dump_path = dump_dir.path().join("ursula-heap.heap");
    let dump_path = dump_path
        .to_str()
        .ok_or_else(|| HeapProfileError::Io("heap profile path is not UTF-8".to_owned()))?;
    // `raw::write_str` only accepts a `'static` value; leaking the short
    // path string on each authorized dump is the price of staying on the
    // safe mallctl API.
    let dump_path_nul: &'static [u8] =
        Box::leak(format!("{dump_path}\0").into_bytes().into_boxed_slice());
    tikv_jemalloc_ctl::raw::write_str(b"prof.dump\0", dump_path_nul).map_err(|err| {
        HeapProfileError::Io(format!("dump jemalloc heap profile to {dump_path}: {err}"))
    })?;
    let bytes = std::fs::read(dump_path).map_err(|err| {
        HeapProfileError::Io(format!("read jemalloc heap profile {dump_path}: {err}"))
    })?;
    if !heap_profile_has_samples(&bytes) {
        return Err(HeapProfileError::Disabled(format!(
            "heap profile contains no samples; ensure {MALLOC_CONF_ENV_VAR} sets \
             prof_active:true and that jemalloc is this process's allocator"
        )));
    }
    Ok(bytes)
}

// A jemalloc `heap_v2` dump aggregates its totals on the first `t*:` line as
// `t*: <live count>: <live bytes> [...]`. Zero live samples means sampling is
// not actually recording (prof_active:false, or jemalloc is linked but not
// the process's global allocator), which must surface as an error instead of
// an empty-but-200 profile.
#[cfg(feature = "jemalloc-prof")]
fn heap_profile_has_samples(profile: &[u8]) -> bool {
    let text = String::from_utf8_lossy(profile);
    for line in text.lines() {
        let Some(totals) = line.trim_start().strip_prefix("t*:") else {
            continue;
        };
        return totals
            .split(':')
            .next()
            .and_then(|count| count.trim().parse::<u64>().ok())
            .is_none_or(|count| count > 0);
    }
    true
}

pub(crate) async fn flush_cold_stream(
    State(state): State<HttpState>,
    OriginalUri(uri): OriginalUri,
    Path(path): Path<StreamPath>,
    RawQuery(raw_query): RawQuery,
) -> Response {
    let query = match parse_query(raw_query.as_deref()) {
        Ok(query) => query,
        Err(response) => return *response,
    };
    let min_hot_bytes = query
        .get("min_hot_bytes")
        .and_then(|raw| raw.parse::<usize>().ok())
        .unwrap_or(1);
    let max_flush_bytes = query
        .get("max_bytes")
        .and_then(|raw| raw.parse::<usize>().ok())
        .unwrap_or(8 * 1024 * 1024);
    let stream_id = path.into_stream_id();
    match state
        .runtime
        .flush_cold_once(PlanColdFlushRequest {
            stream_id,
            min_hot_bytes,
            max_flush_bytes,
        })
        .await
    {
        Ok(Some(response)) => json_response(
            StatusCode::OK,
            serde_json::json!({
                "hot_start_offset": response.hot_start_offset,
                "group_commit_index": response.group_commit_index,
            })
            .to_string(),
        ),
        Ok(None) => StatusCode::NO_CONTENT.into_response(),
        Err(err) => {
            runtime_error_or_leader_redirect_async(&state, err, &request_target(&uri)).await
        }
    }
}

pub(crate) const BACKUP_FORMAT_VERSION: u32 = 1;
pub(crate) const HEADER_BACKUP_FORMAT: &str = "x-ursula-backup-format";
pub(crate) const HEADER_BACKUP_BLAKE3: &str = "x-ursula-backup-blake3";
pub(crate) const HEADER_BACKUP_COMMIT_INDEX: &str = "x-ursula-backup-commit-index";

/// Cluster shape a backup client needs before iterating groups.
pub(crate) async fn backup_info(State(state): State<HttpState>) -> Response {
    json_response(
        StatusCode::OK,
        serde_json::json!({
            "format_version": BACKUP_FORMAT_VERSION,
            "raft_group_count": state.runtime.raft_group_count(),
        })
        .to_string(),
    )
}

/// Exports one group's complete stream state as a MessagePack document.
///
/// The export is the same deterministic `StreamSnapshot` the raft snapshot
/// path persists, so it is internally consistent per group while writes
/// continue; cross-group consistency is intentionally not promised (the
/// recovery boundary is per stream).
pub(crate) async fn export_backup_group(
    State(state): State<HttpState>,
    OriginalUri(uri): OriginalUri,
    Path(raft_group_id): Path<u64>,
) -> Response {
    let group_count = u64::from(state.runtime.raft_group_count());
    if raft_group_id >= group_count {
        return (
            StatusCode::BAD_REQUEST,
            format!("raft group {raft_group_id} out of range 0..{group_count}"),
        )
            .into_response();
    }
    let Ok(raft_group_id) = parse_raft_group_id(raft_group_id) else {
        return (StatusCode::BAD_REQUEST, "invalid raft group id").into_response();
    };
    let snapshot = match state.runtime.snapshot_group(raft_group_id).await {
        Ok(snapshot) => snapshot,
        Err(err) => {
            return runtime_error_or_leader_redirect_async(&state, err, &request_target(&uri))
                .await;
        }
    };
    let body = match rmp_serde::to_vec_named(&snapshot.stream_snapshot) {
        Ok(body) => body,
        Err(err) => {
            return (
                StatusCode::INTERNAL_SERVER_ERROR,
                format!("encode backup snapshot: {err}"),
            )
                .into_response();
        }
    };
    let checksum = blake3::hash(&body).to_hex().to_string();
    let mut response = (StatusCode::OK, body).into_response();
    let headers = response.headers_mut();
    headers.insert(HEADER_BACKUP_FORMAT, HeaderValue::from_static("1"));
    if let Ok(value) = HeaderValue::from_str(&checksum) {
        headers.insert(HEADER_BACKUP_BLAKE3, value);
    }
    if let Ok(value) = HeaderValue::from_str(&snapshot.group_commit_index.to_string()) {
        headers.insert(HEADER_BACKUP_COMMIT_INDEX, value);
    }
    headers.insert(
        axum::http::header::CONTENT_TYPE,
        HeaderValue::from_static("application/x-msgpack"),
    );
    response
}

/// Imports one group's backup snapshot into an empty group as a replicated
/// write. Non-empty groups fail closed with `409`; invalid payloads with
/// `400`. The restored cluster keeps its own raft identity and membership.
pub(crate) async fn import_backup_group(
    State(state): State<HttpState>,
    OriginalUri(uri): OriginalUri,
    Path(raft_group_id): Path<u64>,
    body: axum::body::Bytes,
) -> Response {
    let group_count = u64::from(state.runtime.raft_group_count());
    if raft_group_id >= group_count {
        return (
            StatusCode::BAD_REQUEST,
            format!("raft group {raft_group_id} out of range 0..{group_count}"),
        )
            .into_response();
    }
    let Ok(raft_group_id) = parse_raft_group_id(raft_group_id) else {
        return (StatusCode::BAD_REQUEST, "invalid raft group id").into_response();
    };
    let snapshot: ursula_runtime::StreamSnapshot = match rmp_serde::from_slice(&body) {
        Ok(snapshot) => snapshot,
        Err(err) => {
            return (
                StatusCode::BAD_REQUEST,
                format!("decode backup snapshot: {err}"),
            )
                .into_response();
        }
    };
    match state
        .runtime
        .import_group_state(raft_group_id, ImportGroupStateRequest {
            snapshot: Box::new(snapshot),
        })
        .await
    {
        Ok(response) => json_response(
            StatusCode::OK,
            serde_json::json!({
                "buckets": response.buckets,
                "streams": response.streams,
                "group_commit_index": response.group_commit_index,
            })
            .to_string(),
        ),
        Err(err) => {
            runtime_error_or_leader_redirect_async(&state, err, &request_target(&uri)).await
        }
    }
}

pub(crate) async fn trigger_raft_snapshot(
    State(state): State<HttpState>,
    Path(raft_group_id): Path<u64>,
) -> Response {
    let (raft_group_id, raft) = match resolve_raft_group(&state, raft_group_id) {
        Ok(resolved) => resolved,
        Err(response) => return *response,
    };
    let snapshot_log_id = raft.metrics().borrow_watched().last_applied;
    if let Err(err) = raft.trigger().snapshot().await {
        return (
            StatusCode::INTERNAL_SERVER_ERROR,
            format!("trigger raft snapshot: {err}"),
        )
            .into_response();
    }
    if let Some(snapshot_log_id) = snapshot_log_id
        && let Err(err) = raft
            .wait(Some(Duration::from_secs(10)))
            .metrics(
                |metrics| {
                    metrics
                        .snapshot
                        .as_ref()
                        .is_some_and(|snapshot| snapshot >= &snapshot_log_id)
                },
                format!("admin snapshot trigger .snapshot >= {snapshot_log_id}"),
            )
            .await
    {
        return (
            StatusCode::INTERNAL_SERVER_ERROR,
            format!("wait for raft snapshot: {err}"),
        )
            .into_response();
    }

    let metrics = raft.metrics().borrow_watched().clone();
    json_response(
        StatusCode::OK,
        serde_json::json!({
            "raft_group_id": raft_group_id.0,
            "snapshot_index": metrics.snapshot.map(|log_id| log_id.index),
        })
        .to_string(),
    )
}

pub(crate) async fn trigger_raft_purge(
    State(state): State<HttpState>,
    Path(raft_group_id): Path<u64>,
    RawQuery(raw_query): RawQuery,
) -> Response {
    let query = match parse_query(raw_query.as_deref()) {
        Ok(query) => query,
        Err(response) => return *response,
    };
    let Some(upto) = query
        .get("upto")
        .and_then(|value| value.parse::<u64>().ok())
    else {
        return (StatusCode::BAD_REQUEST, "upto query parameter is required").into_response();
    };
    let (raft_group_id, raft) = match resolve_raft_group(&state, raft_group_id) {
        Ok(resolved) => resolved,
        Err(response) => return *response,
    };
    if let Err(err) = raft.trigger().purge_log(upto).await {
        return (
            StatusCode::INTERNAL_SERVER_ERROR,
            format!("trigger raft purge: {err}"),
        )
            .into_response();
    }
    if let Err(err) = raft
        .wait(Some(Duration::from_secs(10)))
        .metrics(
            |metrics| metrics.purged.map(|log_id| log_id.index) >= Some(upto),
            format!("admin purge to index {upto}"),
        )
        .await
    {
        return (
            StatusCode::INTERNAL_SERVER_ERROR,
            format!("wait for raft purge: {err}"),
        )
            .into_response();
    }
    let metrics = raft.metrics().borrow_watched().clone();
    json_response(
        StatusCode::OK,
        serde_json::json!({
            "raft_group_id": raft_group_id.0,
            "purged_index": metrics.purged.map(|log_id| log_id.index),
        })
        .to_string(),
    )
}

pub(crate) async fn add_raft_learner(
    State(state): State<HttpState>,
    Path((raft_group_id, node_id)): Path<(u64, u64)>,
    RawQuery(raw_query): RawQuery,
) -> Response {
    let query = match parse_query(raw_query.as_deref()) {
        Ok(query) => query,
        Err(response) => return *response,
    };
    let Some(address) = query.get("addr").filter(|value| !value.trim().is_empty()) else {
        return (StatusCode::BAD_REQUEST, "addr query parameter is required").into_response();
    };
    let blocking = match query.get("blocking") {
        Some(raw) => match raw.parse::<bool>() {
            Ok(blocking) => blocking,
            Err(error) => {
                return (
                    StatusCode::BAD_REQUEST,
                    format!("invalid blocking query parameter '{raw}': {error}"),
                )
                    .into_response();
            }
        },
        None => true,
    };
    let (raft_group_id, raft) = match resolve_raft_group(&state, raft_group_id) {
        Ok(resolved) => resolved,
        Err(response) => return *response,
    };
    match raft
        .add_learner(node_id, BasicNode::new(address.clone()), blocking)
        .await
    {
        Ok(response) => json_response(
            StatusCode::OK,
            serde_json::json!({
                "raft_group_id": raft_group_id.0,
                "node_id": node_id,
                "log_index": response.log_id.index(),
            })
            .to_string(),
        ),
        Err(err) => (
            StatusCode::INTERNAL_SERVER_ERROR,
            format!("add raft learner: {err}"),
        )
            .into_response(),
    }
}

pub(crate) async fn change_raft_membership(
    State(state): State<HttpState>,
    Path(raft_group_id): Path<u64>,
    RawQuery(raw_query): RawQuery,
) -> Response {
    let query = match parse_query(raw_query.as_deref()) {
        Ok(query) => query,
        Err(response) => return *response,
    };
    let Some(raw_voters) = query.get("voters").filter(|value| !value.trim().is_empty()) else {
        return (
            StatusCode::BAD_REQUEST,
            "voters query parameter is required",
        )
            .into_response();
    };
    let voters = match parse_voter_ids(raw_voters) {
        Ok(voters) => voters,
        Err(message) => return (StatusCode::BAD_REQUEST, message).into_response(),
    };
    let (raft_group_id, raft) = match resolve_raft_group(&state, raft_group_id) {
        Ok(resolved) => resolved,
        Err(response) => return *response,
    };
    let metrics = raft.metrics().borrow_watched().clone();
    if metrics.current_leader != Some(metrics.id) {
        return json_response(
            StatusCode::CONFLICT,
            serde_json::json!({
                "raft_group_id": raft_group_id.0,
                "current_leader": metrics.current_leader,
                "changed": false,
                "reason": "not leader",
            })
            .to_string(),
        );
    }

    match raft.change_membership(voters.clone(), false).await {
        Ok(response) => json_response(
            StatusCode::OK,
            serde_json::json!({
                "raft_group_id": raft_group_id.0,
                "voter_ids": voters,
                "log_index": response.log_id.index(),
                "changed": true,
            })
            .to_string(),
        ),
        Err(err) => (
            StatusCode::INTERNAL_SERVER_ERROR,
            format!("change raft membership: {err}"),
        )
            .into_response(),
    }
}

pub(crate) fn parse_voter_ids(raw: &str) -> Result<BTreeSet<u64>, String> {
    let mut voters = BTreeSet::new();
    for part in raw.split(',') {
        let part = part.trim();
        if part.is_empty() {
            return Err("voters contains an empty node id".to_owned());
        }
        let node_id = part
            .parse::<u64>()
            .map_err(|err| format!("invalid voter id '{part}': {err}"))?;
        voters.insert(node_id);
    }
    if voters.is_empty() {
        return Err("voters must not be empty".to_owned());
    }
    Ok(voters)
}

pub(crate) async fn transfer_raft_leader(
    State(state): State<HttpState>,
    Path((raft_group_id, node_id)): Path<(u64, u64)>,
) -> Response {
    let (raft_group_id, raft) = match resolve_raft_group(&state, raft_group_id) {
        Ok(resolved) => resolved,
        Err(response) => return *response,
    };
    let metrics_before = raft.metrics().borrow_watched().clone();
    let current_leader = metrics_before.current_leader;
    let self_id = metrics_before.id;
    if current_leader != Some(self_id) {
        return json_response(
            StatusCode::CONFLICT,
            serde_json::json!({
                "raft_group_id": raft_group_id.0,
                "current_leader": current_leader,
                "transferred": false,
                "reason": "not leader",
            })
            .to_string(),
        );
    }
    if node_id == self_id {
        return (
            StatusCode::BAD_REQUEST,
            "target node_id is the current leader",
        )
            .into_response();
    }
    if !metrics_before
        .membership_config
        .voter_ids()
        .any(|voter| voter == node_id)
    {
        return (
            StatusCode::BAD_REQUEST,
            "target node_id is not a voter in this raft group",
        )
            .into_response();
    }
    if let Err(err) = raft.trigger().transfer_leader(node_id).await {
        return (
            StatusCode::INTERNAL_SERVER_ERROR,
            format!("trigger raft transfer leader: {err}"),
        )
            .into_response();
    }
    json_response(
        StatusCode::OK,
        serde_json::json!({
            "raft_group_id": raft_group_id.0,
            "from": self_id,
            "to": node_id,
            "transferred": true,
        })
        .to_string(),
    )
}

pub(crate) fn parse_raft_group_id(raw: u64) -> Result<RaftGroupId, std::num::TryFromIntError> {
    u32::try_from(raw).map(RaftGroupId)
}

/// Resolves the raft registry, parses the group id, and looks up the live group
/// handle — the preamble shared by every raft admin endpoint. Returns the
/// appropriate error response (`400`/`404`) when any step fails, so handlers can
/// `?`-style early-return and focus on their actual operation.
fn resolve_raft_group(
    state: &HttpState,
    raft_group_id: u64,
) -> Result<(RaftGroupId, RaftGroupHandle), Box<Response>> {
    let Some(registry) = state.raft_registry() else {
        return Err(Box::new(
            (
                StatusCode::BAD_REQUEST,
                "raft registry is not configured for this server",
            )
                .into_response(),
        ));
    };
    let Ok(raft_group_id) = parse_raft_group_id(raft_group_id) else {
        return Err(Box::new(
            (StatusCode::BAD_REQUEST, "invalid raft group id").into_response(),
        ));
    };
    let Some(raft) = registry.get(raft_group_id) else {
        return Err(Box::new(
            (StatusCode::NOT_FOUND, "raft group is not registered").into_response(),
        ));
    };
    Ok((raft_group_id, raft))
}

pub(crate) async fn create_stream(
    State(state): State<HttpState>,
    OriginalUri(uri): OriginalUri,
    Path(path): Path<StreamPath>,
    headers: HeaderMap,
    body: Bytes,
) -> Response {
    let stream_id = path.into_stream_id();
    create_stream_by_id(state, request_target(&uri), stream_id, headers, body).await
}

pub(crate) async fn create_stream_by_id(
    state: HttpState,
    request_target: String,
    stream_id: BucketStreamId,
    request_headers: HeaderMap,
    body: Bytes,
) -> Response {
    let content_type_explicit = has_content_type(&request_headers);
    let content_type = request_content_type(&request_headers);
    let (stream_ttl_seconds, stream_expires_at_ms) = match stream_lifetime(&request_headers) {
        Ok(lifetime) => lifetime,
        Err(response) => return *response,
    };
    let attrs = match stream_attrs(&request_headers) {
        Ok(attrs) => attrs,
        Err(response) => return *response,
    };
    let mut request = CreateStreamRequest::new(stream_id.clone(), content_type.clone());
    request.content_type_explicit = content_type_explicit;
    request.now_ms = state.unix_time_ms();
    request.initial_payload = match normalize_http_write_payload(&content_type, body.clone(), true)
    {
        Ok(payload) => payload,
        Err(message) => return (StatusCode::BAD_REQUEST, message).into_response(),
    };
    request.close_after = stream_closed(&request_headers);
    request.stream_seq = stream_seq(&request_headers);
    request.stream_ttl_seconds = stream_ttl_seconds;
    request.stream_expires_at_ms = stream_expires_at_ms;
    request.attrs = attrs;
    let producer = match producer_request(&request_headers) {
        Ok(producer) => producer,
        Err(message) => return (StatusCode::BAD_REQUEST, message).into_response(),
    };
    request.producer = producer.clone();
    if should_externalize_payload(&state, request.initial_payload.len(), true) {
        return create_stream_external_by_id(state, request_target, request, producer).await;
    }

    match state.runtime.create_stream(request).await {
        Ok(response) => create_stream_http_response(CreateStreamHttpResponseInput {
            response,
            stream_id: &stream_id,
            content_type: &content_type,
            stream_ttl_seconds,
            stream_expires_at_ms,
            producer: producer.as_ref(),
        }),
        Err(err) => runtime_error_or_leader_redirect_async(&state, err, &request_target).await,
    }
}

pub(crate) async fn create_stream_external_by_id(
    state: HttpState,
    request_target: String,
    mut request: CreateStreamRequest,
    producer: Option<ProducerRequest>,
) -> Response {
    let stream_id = request.stream_id.clone();
    let content_type = request.content_type.clone();
    let stream_ttl_seconds = request.stream_ttl_seconds;
    let stream_expires_at_ms = request.stream_expires_at_ms;
    let record_ends = request.canonical_record_ends();
    let payload = std::mem::take(&mut request.initial_payload);
    let external_payload = match stage_external_payload(&state, &stream_id, &payload).await {
        Ok(payload) => payload,
        Err(response) => return response,
    };
    let external_path = external_payload.s3_path.clone();
    let external_request =
        CreateStreamExternalRequest::from_create_request(request, external_payload, record_ends);

    match state.runtime.create_stream_external(external_request).await {
        Ok(response) => create_stream_http_response(CreateStreamHttpResponseInput {
            response,
            stream_id: &stream_id,
            content_type: &content_type,
            stream_ttl_seconds,
            stream_expires_at_ms,
            producer: producer.as_ref(),
        }),
        Err(err) => {
            cleanup_external_payload(&state, &external_path).await;
            runtime_error_or_leader_redirect_async(&state, err, &request_target).await
        }
    }
}

pub(crate) async fn append_stream(
    State(state): State<HttpState>,
    OriginalUri(uri): OriginalUri,
    Path(path): Path<StreamPath>,
    headers: HeaderMap,
    body: Bytes,
) -> Response {
    let stream_id = path.into_stream_id();
    append_stream_by_id(state, request_target(&uri), stream_id, headers, body).await
}

#[tracing::instrument(
    name = "http.append",
    skip_all,
    fields(bucket = %stream_id.bucket_id, stream = %stream_id.stream_id),
)]
pub(crate) async fn append_stream_by_id(
    state: HttpState,
    request_target: String,
    stream_id: BucketStreamId,
    headers: HeaderMap,
    body: Bytes,
) -> Response {
    let close_after = stream_closed(&headers);

    if body.is_empty() && close_after {
        let producer = match producer_request(&headers) {
            Ok(producer) => producer,
            Err(message) => return (StatusCode::BAD_REQUEST, message).into_response(),
        };
        return match state
            .runtime
            .close_stream(CloseStreamRequest {
                stream_id,
                stream_seq: stream_seq(&headers),
                producer: producer.clone(),
                now_ms: state.unix_time_ms(),
            })
            .await
        {
            Ok(response) => {
                let mut headers = HeaderMap::new();
                insert_default_response_headers(&mut headers);
                insert_offset(&mut headers, response.next_offset);
                insert_producer_ack(&mut headers, producer.as_ref());
                if let Some(record_range) = response.record_range {
                    insert_record_operation_headers(&mut headers, record_range);
                }
                insert_static(&mut headers, HEADER_STREAM_CLOSED, "true");
                (StatusCode::NO_CONTENT, headers).into_response()
            }
            Err(err) => runtime_error_or_leader_redirect_async(&state, err, &request_target).await,
        };
    }
    if !body.is_empty() && !has_content_type(&headers) {
        return (
            StatusCode::BAD_REQUEST,
            "append with a body must include content type",
        )
            .into_response();
    }

    let content_type = request_content_type(&headers);
    let payload = match normalize_http_write_payload(&content_type, body.clone(), false) {
        Ok(payload) => payload,
        Err(message) => return (StatusCode::BAD_REQUEST, message).into_response(),
    };
    let mut request = AppendRequest::from_bytes(stream_id, payload);
    request.content_type = content_type;
    request.close_after = close_after;
    request.stream_seq = stream_seq(&headers);
    request.now_ms = state.unix_time_ms();
    let producer = match producer_request(&headers) {
        Ok(producer) => producer,
        Err(message) => return (StatusCode::BAD_REQUEST, message).into_response(),
    };
    request.producer = producer.clone();
    request.record_match = match stream_record_match(&headers) {
        Ok(record_match) => record_match,
        Err(response) => return *response,
    };

    if should_externalize_payload(&state, request.payload.len(), true) {
        return append_stream_external_by_id(state, request_target, request).await;
    }

    match state.runtime.append(request).await {
        Ok(response) => append_http_response(response),
        Err(err) => runtime_error_or_leader_redirect_async(&state, err, &request_target).await,
    }
}

pub(crate) async fn append_stream_external_by_id(
    state: HttpState,
    request_target: String,
    mut request: AppendRequest,
) -> Response {
    let stream_id = request.stream_id.clone();
    let record_ends = request.canonical_record_ends();
    let payload = std::mem::take(&mut request.payload);
    let external_payload = match stage_external_payload(&state, &stream_id, &payload).await {
        Ok(payload) => payload,
        Err(response) => return response,
    };
    let external_path = external_payload.s3_path.clone();
    let external_request =
        AppendExternalRequest::from_append_request(request, external_payload, record_ends);
    match state.runtime.append_external(external_request).await {
        Ok(response) => append_http_response(response),
        Err(err) => {
            cleanup_external_payload(&state, &external_path).await;
            runtime_error_or_leader_redirect_async(&state, err, &request_target).await
        }
    }
}

#[tracing::instrument(
    name = "http.append_batch",
    skip_all,
    fields(bucket = %path.bucket, affinity = ?path.affinity, stream = %path.stream, bytes = body.len(), payloads = tracing::field::Empty),
)]
pub(crate) async fn append_batch(
    State(state): State<HttpState>,
    OriginalUri(uri): OriginalUri,
    Path(path): Path<StreamPath>,
    headers: HeaderMap,
    body: Bytes,
) -> Response {
    if body.len() > APPEND_BATCH_MAX_BYTES {
        return (StatusCode::PAYLOAD_TOO_LARGE, "append batch is too large").into_response();
    }
    let producer = match producer_request(&headers) {
        Ok(producer) => producer,
        Err(message) => return (StatusCode::BAD_REQUEST, message).into_response(),
    };
    let minimal_ack = prefers_minimal_response(&headers);
    let payloads = match parse_append_batch(&body) {
        Ok(payloads) => payloads,
        Err(message) => return (StatusCode::BAD_REQUEST, message).into_response(),
    };
    tracing::Span::current().record("payloads", payloads.len());
    if payloads.len() > APPEND_BATCH_MAX_ITEMS {
        return (
            StatusCode::PAYLOAD_TOO_LARGE,
            "append batch contains too many items",
        )
            .into_response();
    }

    let stream_id = path.into_stream_id();
    let content_type = request_content_type(&headers);
    let payloads = match payloads
        .into_iter()
        .map(|payload| normalize_http_write_payload(&content_type, payload, false))
        .collect::<Result<Vec<_>, _>>()
    {
        Ok(payloads) => payloads,
        Err(message) => return (StatusCode::BAD_REQUEST, message).into_response(),
    };
    let mut request = AppendBatchRequest::new(stream_id, payloads);
    request.content_type = content_type;
    request.producer = producer.clone();
    request.now_ms = state.unix_time_ms();
    let response = match state.runtime.append_batch(request).await {
        Ok(response) => response,
        Err(err) => {
            return runtime_error_or_leader_redirect_async(&state, err, &request_target(&uri))
                .await;
        }
    };

    let mut headers = HeaderMap::new();
    insert_default_response_headers(&mut headers);
    insert_producer_ack(&mut headers, producer.as_ref());
    let has_record_ranges = response.items.iter().any(|item| {
        item.as_ref()
            .is_ok_and(|response| response.record_range.is_some())
    });
    if has_record_ranges {
        insert_record_extension(&mut headers);
    }
    if minimal_ack && response.items.iter().all(Result::is_ok) && !has_record_ranges {
        return (StatusCode::NO_CONTENT, headers).into_response();
    }

    insert_content_type(&mut headers, "application/json");
    let body = render_batch_results(&response.items);
    (StatusCode::OK, headers, body).into_response()
}

#[tracing::instrument(
    name = "http.append_transaction",
    skip_all,
    fields(bucket = %path.bucket, affinity = %path.affinity, bytes = body.len()),
)]
pub(crate) async fn append_transaction(
    State(state): State<HttpState>,
    OriginalUri(uri): OriginalUri,
    Path(path): Path<AffinityPath>,
    headers: HeaderMap,
    body: Bytes,
) -> Response {
    if !render::is_json_content_type(&request_content_type(&headers)) {
        return (
            StatusCode::BAD_REQUEST,
            "append transaction body must be application/json",
        )
            .into_response();
    }
    let transaction = match serde_json::from_slice::<AppendTransactionHttpRequest>(&body) {
        Ok(transaction) => transaction,
        Err(err) => {
            return (
                StatusCode::BAD_REQUEST,
                format!("invalid append transaction JSON: {err}"),
            )
                .into_response();
        }
    };
    let now_ms = state.unix_time_ms();
    let mut operations = Vec::with_capacity(transaction.operations.len());
    for operation in transaction.operations {
        let payload = match BASE64_STANDARD.decode(operation.payload_base64) {
            Ok(payload) => Bytes::from(payload),
            Err(err) => {
                return (
                    StatusCode::BAD_REQUEST,
                    format!("invalid payload_base64 for '{}': {err}", operation.stream),
                )
                    .into_response();
            }
        };
        let payload = match normalize_http_write_payload(&operation.content_type, payload, false) {
            Ok(payload) => payload,
            Err(message) => return (StatusCode::BAD_REQUEST, message).into_response(),
        };
        operations.push(AppendRequest {
            stream_id: BucketStreamId::with_affinity(
                path.bucket.clone(),
                path.affinity.clone(),
                operation.stream,
            ),
            content_type: operation.content_type,
            payload,
            close_after: operation.close_after,
            stream_seq: operation.stream_seq,
            producer: operation.producer,
            now_ms,
            record_match: operation.record_match,
        });
    }
    let response = match state
        .runtime
        .append_transaction(AppendTransactionRequest { operations })
        .await
    {
        Ok(response) => response,
        Err(err) => {
            return runtime_error_or_leader_redirect_async(&state, err, &request_target(&uri))
                .await;
        }
    };
    let body = match serde_json::to_vec(&response.items) {
        Ok(body) => body,
        Err(err) => {
            return (
                StatusCode::INTERNAL_SERVER_ERROR,
                format!("render append transaction JSON: {err}"),
            )
                .into_response();
        }
    };
    let mut response_headers = HeaderMap::new();
    insert_default_response_headers(&mut response_headers);
    insert_content_type(&mut response_headers, "application/json");
    (StatusCode::OK, response_headers, body).into_response()
}

pub(crate) async fn delete_stream(
    State(state): State<HttpState>,
    OriginalUri(uri): OriginalUri,
    Path(path): Path<StreamPath>,
) -> Response {
    let stream_id = path.into_stream_id();
    delete_stream_by_id(state, request_target(&uri), stream_id).await
}

pub(crate) async fn delete_stream_by_id(
    state: HttpState,
    request_target: String,
    stream_id: BucketStreamId,
) -> Response {
    match state
        .runtime
        .delete_stream(DeleteStreamRequest { stream_id })
        .await
    {
        Ok(_) => {
            let mut headers = HeaderMap::new();
            insert_default_response_headers(&mut headers);
            (StatusCode::NO_CONTENT, headers).into_response()
        }
        Err(err) => runtime_error_or_leader_redirect_async(&state, err, &request_target).await,
    }
}

pub(crate) async fn update_stream_attrs(
    State(state): State<HttpState>,
    OriginalUri(uri): OriginalUri,
    Path(path): Path<StreamPath>,
    headers: HeaderMap,
    body: Bytes,
) -> Response {
    if !has_content_type(&headers) {
        return (
            StatusCode::BAD_REQUEST,
            "stream attrs update must include content type",
        )
            .into_response();
    }
    let content_type = request_content_type(&headers);
    if !render::is_json_content_type(&content_type) {
        return (
            StatusCode::BAD_REQUEST,
            "stream attrs update body must be application/json",
        )
            .into_response();
    }
    let attrs = match serde_json::from_slice::<StreamAttrs>(&body) {
        Ok(attrs) => attrs,
        Err(err) => {
            return (
                StatusCode::BAD_REQUEST,
                format!("invalid stream attrs JSON: {err}"),
            )
                .into_response();
        }
    };
    let stream_id = path.into_stream_id();
    match state
        .runtime
        .update_stream_attrs(UpdateStreamAttrsRequest {
            stream_id,
            attrs: Some(attrs),
            now_ms: state.unix_time_ms(),
        })
        .await
    {
        Ok(_) => {
            let mut headers = HeaderMap::new();
            insert_default_response_headers(&mut headers);
            (StatusCode::NO_CONTENT, headers).into_response()
        }
        Err(err) => {
            runtime_error_or_leader_redirect_async(&state, err, &request_target(&uri)).await
        }
    }
}

pub(crate) async fn get_stream_attrs(
    State(state): State<HttpState>,
    OriginalUri(uri): OriginalUri,
    Path(path): Path<StreamPath>,
) -> Response {
    let stream_id = path.into_stream_id();
    match state
        .runtime
        .get_stream_attrs(GetStreamAttrsRequest {
            stream_id,
            now_ms: state.unix_time_ms(),
        })
        .await
    {
        Ok(response) => {
            let attrs = response.attrs.unwrap_or_default();
            let body = match serde_json::to_vec(&attrs) {
                Ok(body) => body,
                Err(err) => {
                    return (
                        StatusCode::INTERNAL_SERVER_ERROR,
                        format!("render stream attrs JSON: {err}"),
                    )
                        .into_response();
                }
            };
            let mut headers = HeaderMap::new();
            insert_default_response_headers(&mut headers);
            insert_content_type(&mut headers, "application/json");
            insert_cache_control(&mut headers, "no-store");
            (StatusCode::OK, headers, body).into_response()
        }
        Err(err) => {
            runtime_error_or_leader_redirect_async(&state, err, &request_target(&uri)).await
        }
    }
}

pub(crate) async fn head_stream(
    State(state): State<HttpState>,
    OriginalUri(uri): OriginalUri,
    Path(path): Path<StreamPath>,
) -> Response {
    let stream_id = path.into_stream_id();
    head_stream_by_id(state, request_target(&uri), stream_id).await
}

#[tracing::instrument(
    name = "http.head",
    skip_all,
    fields(bucket = %stream_id.bucket_id, stream = %stream_id.stream_id),
)]
pub(crate) async fn head_stream_by_id(
    state: HttpState,
    request_target: String,
    stream_id: BucketStreamId,
) -> Response {
    match state
        .runtime
        .head_stream(HeadStreamRequest {
            stream_id,
            now_ms: state.unix_time_ms(),
        })
        .await
    {
        Ok(response) => {
            let mut headers = HeaderMap::new();
            insert_default_response_headers(&mut headers);
            insert_content_type(&mut headers, &response.content_type);
            insert_offset(&mut headers, response.tail_offset);
            insert_u64_header(
                &mut headers,
                HEADER_STREAM_COLD_HOT_START_OFFSET,
                response.cold_hot_start_offset,
            );
            insert_static(&mut headers, HEADER_STREAM_UP_TO_DATE, "true");
            insert_cache_control(&mut headers, "no-store");
            insert_lifetime_headers(
                &mut headers,
                response.stream_ttl_seconds,
                response.stream_expires_at_ms,
            );
            insert_header_str(
                &mut headers,
                HEADER_STREAM_INTEGRITY_LIVE_SETSUM,
                &response.integrity.live_setsum,
            );
            insert_header_str(
                &mut headers,
                HEADER_STREAM_INTEGRITY_EVICTED_SETSUM,
                &response.integrity.evicted_setsum,
            );
            insert_header_str(
                &mut headers,
                HEADER_STREAM_INTEGRITY_TOTAL_SETSUM,
                &response.integrity.total_setsum,
            );
            insert_u64_header(
                &mut headers,
                HEADER_STREAM_INTEGRITY_LIVE_START_OFFSET,
                response.integrity.live_start_offset,
            );
            insert_u64_header(
                &mut headers,
                HEADER_STREAM_INTEGRITY_LIVE_RECORDS,
                response.integrity.live_records,
            );
            insert_u64_header(
                &mut headers,
                HEADER_STREAM_INTEGRITY_EVICTED_RECORDS,
                response.integrity.evicted_records,
            );
            insert_u64_header(
                &mut headers,
                HEADER_STREAM_INTEGRITY_TOTAL_RECORDS,
                response.integrity.total_records,
            );
            if let Some(snapshot_offset) = response.snapshot_offset {
                insert_snapshot_offset(&mut headers, snapshot_offset);
            }
            if let Some(snapshot_digest) = response.snapshot_digest {
                insert_snapshot_digest(&mut headers, &snapshot_digest);
            }
            insert_u64_header(
                &mut headers,
                HEADER_STREAM_RETAINED_OFFSET,
                response.retained_offset,
            );
            if let Some(record_range) = response.record_range {
                insert_record_head_headers(&mut headers, record_range);
            }
            if response.closed {
                insert_static(&mut headers, HEADER_STREAM_CLOSED, "true");
            }
            (StatusCode::OK, headers).into_response()
        }
        Err(err) => runtime_error_or_leader_redirect_async(&state, err, &request_target).await,
    }
}

fn insert_record_extension(headers: &mut HeaderMap) {
    insert_extension_token(headers, JSON_RECORD_COORDINATES_EXTENSION);
}

fn insert_extension_token(headers: &mut HeaderMap, token: &'static str) {
    let value = match headers
        .get(HEADER_STREAM_EXTENSIONS)
        .and_then(|value| value.to_str().ok())
    {
        Some(existing) if existing.split(',').any(|item| item.trim() == token) => return,
        Some(existing) => format!("{existing}, {token}"),
        None => token.to_owned(),
    };
    if let Ok(value) = HeaderValue::from_str(&value) {
        headers.insert(HEADER_STREAM_EXTENSIONS, value);
    }
}

fn insert_record_operation_headers(
    headers: &mut HeaderMap,
    record_range: ursula_runtime::StreamRecordRange,
) {
    insert_record_extension(headers);
    insert_u64_header(
        headers,
        HEADER_STREAM_RECORD_START,
        record_range.first_record,
    );
    insert_u64_header(headers, HEADER_STREAM_RECORD_NEXT, record_range.next_record);
}

fn insert_record_head_headers(
    headers: &mut HeaderMap,
    record_range: ursula_runtime::StreamRecordRange,
) {
    insert_record_extension(headers);
    insert_u64_header(
        headers,
        HEADER_STREAM_RECORD_FIRST,
        record_range.first_record,
    );
    insert_u64_header(headers, HEADER_STREAM_RECORD_NEXT, record_range.next_record);
}

pub(crate) async fn read_stream(
    State(state): State<HttpState>,
    OriginalUri(uri): OriginalUri,
    Path(path): Path<StreamPath>,
    headers: HeaderMap,
    RawQuery(raw_query): RawQuery,
) -> Response {
    let stream_id = path.into_stream_id();
    read_stream_by_id(state, request_target(&uri), stream_id, headers, raw_query).await
}

#[tracing::instrument(
    name = "http.read",
    skip_all,
    fields(bucket = %stream_id.bucket_id, stream = %stream_id.stream_id),
)]
pub(crate) async fn read_stream_by_id(
    state: HttpState,
    request_target: String,
    stream_id: BucketStreamId,
    headers: HeaderMap,
    raw_query: Option<String>,
) -> Response {
    let query = match parse_query(raw_query.as_deref()) {
        Ok(query) => query,
        Err(response) => return *response,
    };
    let live_mode = query.get("live").map(String::as_str);
    let leader_only = match query.get("consistency").map(String::as_str) {
        None | Some("local") => false,
        Some("leader") => true,
        Some(_) => return (StatusCode::BAD_REQUEST, "invalid consistency").into_response(),
    };
    if leader_only && live_mode.is_some() {
        return (
            StatusCode::BAD_REQUEST,
            "leader consistency is available only for catch-up reads",
        )
            .into_response();
    }
    let offset_is_now = query.get("offset").is_some_and(|offset| offset == "now");
    let record_aware = query.contains_key("record") || query.contains_key("tail_records");
    let envelope_view = match query.get("record_view").map(String::as_str) {
        None => false,
        Some("envelope") if record_aware => true,
        Some("envelope") => {
            return (
                StatusCode::BAD_REQUEST,
                "record_view requires record or tail_records",
            )
                .into_response();
        }
        Some(_) => return (StatusCode::BAD_REQUEST, "invalid record_view").into_response(),
    };
    if query.contains_key("record") && query.contains_key("tail_records")
        || record_aware && query.contains_key("offset")
    {
        return (
            StatusCode::BAD_REQUEST,
            "record, tail_records, and offset are mutually exclusive",
        )
            .into_response();
    }
    if query.contains_key("max_records") && !record_aware {
        return (
            StatusCode::BAD_REQUEST,
            "max_records requires record or tail_records",
        )
            .into_response();
    }
    if record_aware && query.contains_key("max_bytes") {
        return (
            StatusCode::BAD_REQUEST,
            "record-aware reads do not support max_bytes",
        )
            .into_response();
    }
    if live_mode.is_some() && !query.contains_key("offset") && !record_aware {
        return (
            StatusCode::BAD_REQUEST,
            "live reads require a start position",
        )
            .into_response();
    }
    if matches!(live_mode, Some("sse" | "long-poll"))
        && let Err(err) = state
            .runtime
            .require_local_live_read_owner(&stream_id)
            .await
    {
        return runtime_error_or_leader_redirect_async(&state, err, &request_target).await;
    }
    let record = if record_aware {
        match read_record_start(&state, &stream_id, &query, &request_target).await {
            Ok(record) => Some(record),
            Err(response) => return *response,
        }
    } else {
        None
    };
    let max_records = match query.get("max_records") {
        Some(raw) => match raw.parse::<u64>() {
            Ok(value) if value > 0 => Some(value),
            _ => {
                return (StatusCode::BAD_REQUEST, "max_records must be positive").into_response();
            }
        },
        None => None,
    };
    let offset = if record_aware {
        0
    } else {
        match read_offset(
            &state,
            &stream_id,
            query.get("offset").map(String::as_str),
            &request_target,
        )
        .await
        {
            Ok(offset) => offset,
            Err(response) => return *response,
        }
    };
    let max_len = query
        .get("max_bytes")
        .and_then(|raw| raw.parse::<usize>().ok())
        .unwrap_or(usize::MAX);

    match live_mode {
        Some("sse") => {
            return sse_stream(
                state,
                request_target,
                stream_id,
                offset,
                max_len,
                record,
                max_records,
                envelope_view,
                &query,
            )
            .await;
        }
        Some("long-poll") => {
            return long_poll_stream(
                state,
                request_target,
                stream_id,
                offset,
                max_len,
                record,
                max_records,
                envelope_view,
                &query,
                headers,
            )
            .await;
        }
        Some(_) => return (StatusCode::BAD_REQUEST, "invalid live mode").into_response(),
        None => {}
    }

    match state
        .runtime
        .read_stream(ReadStreamRequest {
            stream_id,
            offset,
            max_len,
            now_ms: state.unix_time_ms(),
            record,
            max_records,
            leader_only,
        })
        .await
    {
        Ok(response) if offset_is_now => offset_now_response(response),
        Ok(response) if envelope_view => record_envelope_response(response, &headers, None),
        Ok(response) => read_response(response, &headers, None),
        Err(err) => runtime_error_or_leader_redirect_async(&state, err, &request_target).await,
    }
}

async fn read_record_start(
    state: &HttpState,
    stream_id: &BucketStreamId,
    query: &HashMap<String, String>,
    request_target: &str,
) -> Result<u64, BoxResponse> {
    let head = match state
        .runtime
        .head_stream(HeadStreamRequest {
            stream_id: stream_id.clone(),
            now_ms: state.unix_time_ms(),
        })
        .await
    {
        Ok(head) => head,
        Err(err) => {
            return Err(Box::new(
                runtime_error_or_leader_redirect_async(state, err, request_target).await,
            ));
        }
    };
    let Some(range) = head.record_range else {
        return Err(Box::new(
            (
                StatusCode::BAD_REQUEST,
                "record coordinates are inactive for this stream",
            )
                .into_response(),
        ));
    };
    let record = if let Some(raw) = query.get("record") {
        if raw == "now" {
            range.next_record
        } else {
            raw.parse::<u64>().map_err(|_| {
                Box::new((StatusCode::BAD_REQUEST, "invalid record").into_response())
            })?
        }
    } else {
        let count = query
            .get("tail_records")
            .and_then(|raw| raw.parse::<u64>().ok())
            .ok_or_else(|| {
                Box::new((StatusCode::BAD_REQUEST, "invalid tail_records").into_response())
            })?;
        range
            .next_record
            .saturating_sub(count)
            .max(range.first_record)
    };
    if record < range.first_record || record > range.next_record {
        let status = if record < range.first_record {
            StatusCode::GONE
        } else {
            StatusCode::BAD_REQUEST
        };
        let mut headers = HeaderMap::new();
        insert_default_response_headers(&mut headers);
        insert_record_head_headers(&mut headers, range);
        return Err(Box::new((status, headers).into_response()));
    }
    Ok(record)
}

#[tracing::instrument(
    name = "http.snapshot_publish",
    skip_all,
    fields(bucket = %path.bucket, affinity = ?path.affinity, stream = %path.stream, snapshot_offset = %path.snapshot_offset, bytes = body.len()),
)]
pub(crate) async fn publish_snapshot(
    State(state): State<HttpState>,
    OriginalUri(uri): OriginalUri,
    Path(path): Path<SnapshotPath>,
    headers: HeaderMap,
    body: Bytes,
) -> Response {
    let (stream_id, snapshot_offset) = path.into_parts();
    let snapshot_offset = match parse_snapshot_offset(&snapshot_offset) {
        Ok(offset) => offset,
        Err(response) => return *response,
    };
    publish_snapshot_by_offset(
        state,
        request_target(&uri),
        stream_id,
        snapshot_offset,
        headers,
        body,
    )
    .await
}

async fn publish_snapshot_by_offset(
    state: HttpState,
    request_target: String,
    stream_id: BucketStreamId,
    snapshot_offset: u64,
    headers: HeaderMap,
    body: Bytes,
) -> Response {
    let expected_digest = match headers.get(HEADER_STREAM_SNAPSHOT_MATCH) {
        Some(value) => match value.to_str() {
            Ok(value) if !value.trim().is_empty() => Some(value.to_owned()),
            _ => {
                return (StatusCode::BAD_REQUEST, "invalid Stream-Snapshot-Match").into_response();
            }
        },
        None => None,
    };
    let request = PublishSnapshotRequest {
        stream_id,
        snapshot_offset,
        content_type: request_content_type(&headers),
        payload: body,
        expected_digest,
        now_ms: state.unix_time_ms(),
    };
    match state.runtime.publish_snapshot(request).await {
        Ok(response) => {
            let mut headers = HeaderMap::new();
            insert_default_response_headers(&mut headers);
            insert_snapshot_offset(&mut headers, response.snapshot_offset);
            insert_snapshot_digest(&mut headers, &response.snapshot_digest);
            if let Some(record_range) = response.record_range {
                insert_record_head_headers(&mut headers, record_range);
            }
            (StatusCode::NO_CONTENT, headers).into_response()
        }
        Err(err) => runtime_error_or_leader_redirect_async(&state, err, &request_target).await,
    }
}

pub(crate) async fn publish_snapshot_at_record(
    State(state): State<HttpState>,
    OriginalUri(uri): OriginalUri,
    Path(path): Path<StreamPath>,
    headers: HeaderMap,
    RawQuery(raw_query): RawQuery,
    body: Bytes,
) -> Response {
    let query = match parse_query(raw_query.as_deref()) {
        Ok(query) => query,
        Err(response) => return *response,
    };
    let Some(record) = query.get("record") else {
        return (
            StatusCode::BAD_REQUEST,
            "record query parameter is required",
        )
            .into_response();
    };
    let record = match record.parse::<u64>() {
        Ok(record) => record,
        Err(_) => return (StatusCode::BAD_REQUEST, "invalid record").into_response(),
    };
    let stream_id = path.into_stream_id();
    let request_target = request_target(&uri);
    let snapshot_offset =
        match resolve_record_offset(&state, &stream_id, record, &request_target).await {
            Ok(offset) => offset,
            Err(response) => return response,
        };
    publish_snapshot_by_offset(
        state,
        request_target,
        stream_id,
        snapshot_offset,
        headers,
        body,
    )
    .await
}

async fn resolve_record_offset(
    state: &HttpState,
    stream_id: &BucketStreamId,
    record: u64,
    request_target: &str,
) -> Result<u64, Response> {
    match state
        .runtime
        .read_stream(ReadStreamRequest {
            stream_id: stream_id.clone(),
            offset: 0,
            max_len: 1,
            now_ms: state.unix_time_ms(),
            record: Some(record),
            max_records: Some(1),
            leader_only: false,
        })
        .await
    {
        Ok(response) => Ok(response.offset),
        Err(err) => Err(runtime_error_or_leader_redirect_async(state, err, request_target).await),
    }
}

pub(crate) async fn advance_retention(
    State(state): State<HttpState>,
    OriginalUri(uri): OriginalUri,
    Path(path): Path<RetentionPath>,
) -> Response {
    let (stream_id, retained_offset) = path.into_parts();
    let retained_offset = match parse_snapshot_offset(&retained_offset) {
        Ok(offset) => offset,
        Err(response) => return *response,
    };
    advance_retention_by_offset(state, request_target(&uri), stream_id, retained_offset).await
}

pub(crate) async fn advance_retention_at_record(
    State(state): State<HttpState>,
    OriginalUri(uri): OriginalUri,
    Path(path): Path<StreamPath>,
    RawQuery(raw_query): RawQuery,
) -> Response {
    let query = match parse_query(raw_query.as_deref()) {
        Ok(query) => query,
        Err(response) => return *response,
    };
    let Some(record) = query.get("record") else {
        return (
            StatusCode::BAD_REQUEST,
            "record query parameter is required",
        )
            .into_response();
    };
    let record = match record.parse::<u64>() {
        Ok(record) => record,
        Err(_) => return (StatusCode::BAD_REQUEST, "invalid record").into_response(),
    };
    let stream_id = path.into_stream_id();
    let request_target = request_target(&uri);
    let retained_offset =
        match resolve_record_offset(&state, &stream_id, record, &request_target).await {
            Ok(offset) => offset,
            Err(response) => return response,
        };
    advance_retention_by_offset(state, request_target, stream_id, retained_offset).await
}

async fn advance_retention_by_offset(
    state: HttpState,
    request_target: String,
    stream_id: BucketStreamId,
    retained_offset: u64,
) -> Response {
    match state
        .runtime
        .advance_retention(AdvanceRetentionRequest {
            stream_id,
            retained_offset,
            now_ms: state.unix_time_ms(),
        })
        .await
    {
        Ok(response) => {
            let mut headers = HeaderMap::new();
            insert_default_response_headers(&mut headers);
            insert_u64_header(
                &mut headers,
                HEADER_STREAM_RETAINED_OFFSET,
                response.retained_offset,
            );
            if let Some(record_range) = response.record_range {
                insert_record_head_headers(&mut headers, record_range);
            }
            (StatusCode::NO_CONTENT, headers).into_response()
        }
        Err(err) => runtime_error_or_leader_redirect_async(&state, err, &request_target).await,
    }
}

#[tracing::instrument(
    name = "http.snapshot_read_latest",
    skip_all,
    fields(bucket = %path.bucket, affinity = ?path.affinity, stream = %path.stream),
)]
pub(crate) async fn read_latest_snapshot(
    State(state): State<HttpState>,
    OriginalUri(uri): OriginalUri,
    Path(path): Path<StreamPath>,
    headers: HeaderMap,
) -> Response {
    let stream_id = path.stream_id();
    let head = match state
        .runtime
        .head_stream(HeadStreamRequest {
            stream_id,
            now_ms: state.unix_time_ms(),
        })
        .await
    {
        Ok(head) => head,
        Err(err) => {
            return runtime_error_or_leader_redirect_async(&state, err, &request_target(&uri))
                .await;
        }
    };
    let Some(snapshot_offset) = head.snapshot_offset else {
        return StatusCode::NOT_FOUND.into_response();
    };
    let mut response_headers = HeaderMap::new();
    insert_default_response_headers(&mut response_headers);
    insert_snapshot_offset(&mut response_headers, snapshot_offset);
    if let Some(snapshot_digest) = head.snapshot_digest {
        insert_snapshot_digest(&mut response_headers, &snapshot_digest);
    }
    if let Some(record_range) = head.record_range {
        insert_record_head_headers(&mut response_headers, record_range);
    }
    let snapshot_path = match &path.affinity {
        Some(affinity) => format!(
            "/{}/{affinity}/{}/snapshot/{snapshot_offset:020}",
            path.bucket, path.stream
        ),
        None => format!(
            "/{}/{}/snapshot/{snapshot_offset:020}",
            path.bucket, path.stream
        ),
    };
    insert_public_location(&mut response_headers, &headers, &snapshot_path);
    (StatusCode::TEMPORARY_REDIRECT, response_headers).into_response()
}

#[tracing::instrument(
    name = "http.snapshot_read",
    skip_all,
    fields(bucket = %path.bucket, affinity = ?path.affinity, stream = %path.stream, snapshot_offset = %path.snapshot_offset),
)]
pub(crate) async fn read_snapshot(
    State(state): State<HttpState>,
    OriginalUri(uri): OriginalUri,
    Path(path): Path<SnapshotPath>,
) -> Response {
    let (stream_id, snapshot_offset) = path.into_parts();
    let snapshot_offset = match parse_snapshot_offset(&snapshot_offset) {
        Ok(offset) => offset,
        Err(response) => return *response,
    };
    match state
        .runtime
        .read_snapshot(ReadSnapshotRequest {
            stream_id,
            snapshot_offset: Some(snapshot_offset),
            now_ms: state.unix_time_ms(),
        })
        .await
    {
        Ok(response) => snapshot_response(response),
        Err(err) => {
            runtime_error_or_leader_redirect_async(&state, err, &request_target(&uri)).await
        }
    }
}

#[tracing::instrument(
    name = "http.snapshot_delete",
    skip_all,
    fields(bucket = %path.bucket, affinity = ?path.affinity, stream = %path.stream, snapshot_offset = %path.snapshot_offset),
)]
pub(crate) async fn delete_snapshot(
    State(state): State<HttpState>,
    OriginalUri(uri): OriginalUri,
    Path(path): Path<SnapshotPath>,
) -> Response {
    let (stream_id, snapshot_offset) = path.into_parts();
    let snapshot_offset = match parse_snapshot_offset(&snapshot_offset) {
        Ok(offset) => offset,
        Err(response) => return *response,
    };
    match state
        .runtime
        .delete_snapshot(DeleteSnapshotRequest {
            stream_id,
            snapshot_offset,
            now_ms: state.unix_time_ms(),
        })
        .await
    {
        Ok(()) => {
            let mut headers = HeaderMap::new();
            insert_default_response_headers(&mut headers);
            (StatusCode::NO_CONTENT, headers).into_response()
        }
        Err(err) => {
            runtime_error_or_leader_redirect_async(&state, err, &request_target(&uri)).await
        }
    }
}

pub(crate) async fn bootstrap_stream(
    State(state): State<HttpState>,
    OriginalUri(uri): OriginalUri,
    Path(path): Path<StreamPath>,
    RawQuery(raw_query): RawQuery,
) -> Response {
    let query = match parse_query(raw_query.as_deref()) {
        Ok(query) => query,
        Err(response) => return *response,
    };
    if query.contains_key("live") {
        return (
            StatusCode::BAD_REQUEST,
            "bootstrap does not support live reads",
        )
            .into_response();
    }
    let stream_id = path.into_stream_id();
    match state
        .runtime
        .bootstrap_stream(BootstrapStreamRequest {
            stream_id,
            now_ms: state.unix_time_ms(),
        })
        .await
    {
        Ok(response) => bootstrap_response(response),
        Err(err) => {
            runtime_error_or_leader_redirect_async(&state, err, &request_target(&uri)).await
        }
    }
}

fn parse_snapshot_offset(raw: &str) -> Result<u64, BoxResponse> {
    if raw == "-1" {
        return Err(Box::new(
            (StatusCode::BAD_REQUEST, "invalid snapshot offset").into_response(),
        ));
    }
    raw.parse::<u64>()
        .map_err(|_| Box::new((StatusCode::BAD_REQUEST, "invalid snapshot offset").into_response()))
}

pub(crate) async fn read_offset(
    state: &HttpState,
    stream_id: &BucketStreamId,
    raw: Option<&str>,
    request_target: &str,
) -> Result<u64, BoxResponse> {
    match raw {
        Some("-1") => Ok(0),
        Some("now") => match state
            .runtime
            .head_stream(HeadStreamRequest {
                stream_id: stream_id.clone(),
                now_ms: state.unix_time_ms(),
            })
            .await
        {
            Ok(head) => Ok(head.tail_offset),
            Err(err) => {
                let response =
                    runtime_error_or_leader_redirect_async(state, err, request_target).await;
                Err(Box::new(response))
            }
        },
        Some(raw) => raw
            .parse::<u64>()
            .map_err(|_| Box::new((StatusCode::BAD_REQUEST, "invalid offset").into_response())),
        None => Ok(0),
    }
}

pub(crate) async fn long_poll_stream(
    state: HttpState,
    request_target: String,
    stream_id: BucketStreamId,
    offset: u64,
    max_len: usize,
    record: Option<u64>,
    max_records: Option<u64>,
    envelope_view: bool,
    query: &HashMap<String, String>,
    headers: HeaderMap,
) -> Response {
    let timeout_ms = long_poll_timeout_ms(query);
    let read = state.runtime.wait_read_stream(ReadStreamRequest {
        stream_id: stream_id.clone(),
        offset,
        max_len: max_len.max(1),
        now_ms: state.unix_time_ms(),
        record,
        max_records,
        leader_only: false,
    });
    match http_time::timeout(Duration::from_millis(timeout_ms), read).await {
        Ok(Ok(response)) if response.payload.is_empty() && response.up_to_date => {
            long_poll_no_content_response(&response, query.get("cursor").map(String::as_str))
        }
        Ok(Ok(response)) if envelope_view => record_envelope_response(
            response,
            &headers,
            Some(query.get("cursor").map(String::as_str).unwrap_or("")),
        ),
        Ok(Ok(response)) => read_response(
            response,
            &headers,
            Some(query.get("cursor").map(String::as_str).unwrap_or("")),
        ),
        Ok(Err(err)) => runtime_error_or_leader_redirect_async(&state, err, &request_target).await,
        Err(_) => match state
            .runtime
            .head_stream(HeadStreamRequest {
                stream_id: stream_id.clone(),
                now_ms: state.unix_time_ms(),
            })
            .await
        {
            Ok(head) => {
                let mut headers = HeaderMap::new();
                insert_default_response_headers(&mut headers);
                insert_offset(&mut headers, head.tail_offset);
                insert_static(&mut headers, HEADER_STREAM_UP_TO_DATE, "true");
                if let (Some(record), Some(record_range)) = (record, head.record_range) {
                    insert_record_head_headers(&mut headers, record_range);
                    insert_record_operation_headers(
                        &mut headers,
                        ursula_runtime::StreamRecordRange {
                            first_record: record,
                            next_record: record,
                        },
                    );
                }
                if head.closed {
                    insert_static(&mut headers, HEADER_STREAM_CLOSED, "true");
                } else {
                    insert_cursor(
                        &mut headers,
                        response_cursor(head.tail_offset, query.get("cursor").map(String::as_str)),
                    );
                }
                (StatusCode::NO_CONTENT, headers).into_response()
            }
            Err(err) => runtime_error_or_leader_redirect_async(&state, err, &request_target).await,
        },
    }
}

#[derive(Clone)]
struct SseState {
    runtime: ShardRuntime,
    http_metrics: Arc<HttpMetrics>,
    wall_clock: Arc<dyn WallClock>,
    stream_id: BucketStreamId,
    offset: u64,
    max_len: usize,
    encode_base64: bool,
    cursor: Option<String>,
    initial_read: bool,
    record: Option<u64>,
    max_records: Option<u64>,
    envelope_view: bool,
}

pub(crate) async fn sse_stream(
    state: HttpState,
    request_target: String,
    stream_id: BucketStreamId,
    offset: u64,
    max_len: usize,
    record: Option<u64>,
    max_records: Option<u64>,
    envelope_view: bool,
    query: &HashMap<String, String>,
) -> Response {
    let head = match state
        .runtime
        .head_stream(HeadStreamRequest {
            stream_id: stream_id.clone(),
            now_ms: state.unix_time_ms(),
        })
        .await
    {
        Ok(head) => head,
        Err(err) => {
            return runtime_error_or_leader_redirect_async(&state, err, &request_target).await;
        }
    };

    let encode_base64 = !envelope_view && should_base64_encode_sse_data(&head.content_type);
    state
        .http_metrics
        .sse_streams_opened
        .fetch_add(1, Ordering::Relaxed);
    let sse_max_len = if encode_base64 {
        max_len.max(1)
    } else {
        max_len.max(4)
    };
    let sse_state = SseState {
        runtime: state.runtime,
        http_metrics: state.http_metrics,
        wall_clock: state.wall_clock,
        stream_id,
        offset,
        max_len: sse_max_len,
        encode_base64,
        cursor: query.get("cursor").cloned(),
        initial_read: true,
        record,
        max_records,
        envelope_view,
    };
    let body_stream = stream::unfold(Some(sse_state), |state| async move {
        let mut state = match state {
            Some(state) => state,
            None => return None,
        };
        state
            .http_metrics
            .sse_read_iterations
            .fetch_add(1, Ordering::Relaxed);
        let read_request = ReadStreamRequest {
            stream_id: state.stream_id.clone(),
            offset: state.offset,
            max_len: state.max_len,
            now_ms: state.wall_clock.unix_time_ms(),
            record: state.record,
            max_records: if state.envelope_view {
                Some(1)
            } else {
                state.max_records
            },
            leader_only: false,
        };
        let read = if state.initial_read {
            state.initial_read = false;
            state.runtime.read_stream(read_request).await
        } else {
            state.runtime.wait_read_stream(read_request).await
        };
        let mut read = match read {
            Ok(read) => read,
            Err(err) => {
                state
                    .http_metrics
                    .sse_error_events
                    .fetch_add(1, Ordering::Relaxed);
                let event = format!("event: error\ndata:{}\n\n", sse_safe_line(&err.to_string()));
                return Some((Ok::<Bytes, Infallible>(Bytes::from(event)), None));
            }
        };
        if state.envelope_view
            && let Err(err) = apply_record_envelope(&mut read)
        {
            let event = format!("event: error\ndata:{}\n\n", sse_safe_line(&err));
            return Some((Ok::<Bytes, Infallible>(Bytes::from(event)), None));
        }
        clamp_sse_text_read(&mut read, state.encode_base64);

        state.offset = read.next_offset;
        state.record = read.record_range.map(|range| range.next_record);
        let done = read.closed && read.up_to_date;
        if !read.payload.is_empty() {
            state
                .http_metrics
                .sse_data_events
                .fetch_add(1, Ordering::Relaxed);
        }
        state
            .http_metrics
            .sse_control_events
            .fetch_add(1, Ordering::Relaxed);
        let event = render_sse_read(&read, state.encode_base64, state.cursor.as_deref());
        let next = if done { None } else { Some(state) };
        Some((Ok::<Bytes, Infallible>(Bytes::from(event)), next))
    });

    let mut headers = HeaderMap::new();
    insert_default_response_headers(&mut headers);
    insert_content_type(&mut headers, "text/event-stream");
    insert_header_str(
        &mut headers,
        HEADER_STREAM_DATA_CONTENT_TYPE,
        if envelope_view {
            "application/vnd.durable-stream-record+json"
        } else {
            http_read_content_type(&head.content_type)
        },
    );
    insert_cache_control(&mut headers, "no-cache");
    if head.record_range.is_some() {
        insert_record_extension(&mut headers);
    }
    if encode_base64 {
        insert_static(&mut headers, HEADER_STREAM_SSE_DATA_ENCODING, "base64");
    }
    (StatusCode::OK, headers, Body::from_stream(body_stream)).into_response()
}

pub(crate) fn long_poll_timeout_ms(query: &HashMap<String, String>) -> u64 {
    query
        .get("timeout_ms")
        .and_then(|raw| raw.parse::<u64>().ok())
        .unwrap_or(DEFAULT_LONG_POLL_TIMEOUT_MS)
        .clamp(1, MAX_LONG_POLL_TIMEOUT_MS)
}

#[cfg(not(madsim))]
pub(crate) fn unix_time_ms() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|duration| u64::try_from(duration.as_millis()).unwrap_or(u64::MAX))
        .unwrap_or(0)
}

#[cfg(madsim)]
pub(crate) fn unix_time_ms() -> u64 {
    panic!(
        "unix_time_ms() / SystemWallClock is non-deterministic under cfg(madsim); \
         inject a deterministic WallClock via HttpState::with_wall_clock (or _handle)"
    );
}

pub(crate) fn parse_query(raw: Option<&str>) -> Result<HashMap<String, String>, BoxResponse> {
    let mut query = HashMap::new();
    let Some(raw) = raw else {
        return Ok(query);
    };
    for (key, value) in url::form_urlencoded::parse(raw.as_bytes()) {
        if key == "offset" && query.contains_key("offset") {
            return Err(Box::new(
                (StatusCode::BAD_REQUEST, "multiple offset parameters").into_response(),
            ));
        }
        query.insert(key.into_owned(), value.into_owned());
    }
    Ok(query)
}

pub(crate) fn request_content_type(headers: &HeaderMap) -> String {
    headers
        .get(CONTENT_TYPE)
        .and_then(|value| value.to_str().ok())
        .filter(|value| !value.trim().is_empty())
        .map(normalize_content_type)
        .unwrap_or_else(|| DEFAULT_CONTENT_TYPE.to_owned())
}

pub(crate) fn stream_attrs(headers: &HeaderMap) -> Result<Option<StreamAttrs>, BoxResponse> {
    let Some(raw) = header_value(headers, HEADER_STREAM_ATTRS) else {
        return Ok(None);
    };
    serde_json::from_str::<StreamAttrs>(raw)
        .map(Some)
        .map_err(|err| {
            Box::new(
                (
                    StatusCode::BAD_REQUEST,
                    format!("invalid stream-attrs JSON: {err}"),
                )
                    .into_response(),
            )
        })
}

pub(crate) fn has_content_type(headers: &HeaderMap) -> bool {
    headers
        .get(CONTENT_TYPE)
        .and_then(|value| value.to_str().ok())
        .is_some_and(|value| !value.trim().is_empty())
}

pub(crate) fn normalize_content_type(value: &str) -> String {
    value
        .split(';')
        .map(str::trim)
        .filter(|part| !part.is_empty())
        .map(str::to_ascii_lowercase)
        .collect::<Vec<_>>()
        .join("; ")
}

pub(crate) fn stream_lifetime(
    headers: &HeaderMap,
) -> Result<(Option<u64>, Option<u64>), BoxResponse> {
    let ttl = header_value(headers, HEADER_STREAM_TTL)
        .map(parse_stream_ttl)
        .transpose()
        .map_err(|message| Box::new((StatusCode::BAD_REQUEST, message).into_response()))?;
    let expires_at = header_value(headers, HEADER_STREAM_EXPIRES_AT)
        .map(parse_stream_expires_at)
        .transpose()
        .map_err(|message| Box::new((StatusCode::BAD_REQUEST, message).into_response()))?;
    if ttl.is_some() && expires_at.is_some() {
        return Err(Box::new(
            (
                StatusCode::BAD_REQUEST,
                "stream-ttl and stream-expires-at cannot be provided together",
            )
                .into_response(),
        ));
    }
    Ok((ttl, expires_at))
}

pub(crate) fn parse_stream_ttl(raw: &str) -> Result<u64, String> {
    if raw.is_empty() {
        return Err("stream-ttl must not be empty".to_owned());
    }
    if raw.len() > 1 && raw.starts_with('0') {
        return Err("stream-ttl must not contain leading zeros".to_owned());
    }
    if !raw.bytes().all(|byte| byte.is_ascii_digit()) {
        return Err("stream-ttl must be a non-negative decimal integer".to_owned());
    }
    raw.parse::<u64>()
        .map_err(|_| "stream-ttl is too large".to_owned())
}

pub(crate) fn parse_stream_expires_at(raw: &str) -> Result<u64, String> {
    let expires_at = DateTime::parse_from_rfc3339(raw)
        .map_err(|_| "stream-expires-at must be an RFC3339 timestamp".to_owned())?;
    u64::try_from(expires_at.timestamp_millis())
        .map_err(|_| "stream-expires-at must not be before the Unix epoch".to_owned())
}

pub(crate) fn stream_closed(headers: &HeaderMap) -> bool {
    headers
        .get(HEADER_STREAM_CLOSED)
        .and_then(|value| value.to_str().ok())
        .is_some_and(|value| value.eq_ignore_ascii_case("true"))
}

pub(crate) fn stream_seq(headers: &HeaderMap) -> Option<String> {
    headers
        .get(HEADER_STREAM_SEQ)
        .and_then(|value| value.to_str().ok())
        .filter(|value| !value.trim().is_empty())
        .map(str::to_owned)
}

fn stream_record_match(headers: &HeaderMap) -> Result<Option<u64>, BoxResponse> {
    header_value(headers, HEADER_STREAM_RECORD_MATCH)
        .map(|raw| {
            raw.parse::<u64>().map_err(|_| {
                Box::new((StatusCode::BAD_REQUEST, "invalid Stream-Record-Match").into_response())
            })
        })
        .transpose()
}

pub(crate) fn producer_request(headers: &HeaderMap) -> Result<Option<ProducerRequest>, String> {
    let producer_id = header_value(headers, HEADER_PRODUCER_ID);
    let producer_epoch = header_value(headers, HEADER_PRODUCER_EPOCH);
    let producer_seq = header_value(headers, HEADER_PRODUCER_SEQ);
    let present = [
        producer_id.is_some(),
        producer_epoch.is_some(),
        producer_seq.is_some(),
    ];
    if present.iter().all(|value| !*value) {
        return Ok(None);
    }
    if !present.iter().all(|value| *value) {
        return Err(
            "producer-id, producer-epoch, and producer-seq must be provided together".to_owned(),
        );
    }

    let producer_id = producer_id.expect("checked present");
    if producer_id.trim().is_empty() {
        return Err("producer-id must not be empty".to_owned());
    }
    Ok(Some(ProducerRequest {
        producer_id: producer_id.to_owned(),
        producer_epoch: parse_producer_integer(
            HEADER_PRODUCER_EPOCH,
            producer_epoch.expect("checked present"),
        )?,
        producer_seq: parse_producer_integer(
            HEADER_PRODUCER_SEQ,
            producer_seq.expect("checked present"),
        )?,
    }))
}

pub(crate) fn prefers_minimal_response(headers: &HeaderMap) -> bool {
    headers
        .get(HEADER_PREFER)
        .and_then(|value| value.to_str().ok())
        .is_some_and(|value| {
            value
                .split(',')
                .any(|part| part.trim().eq_ignore_ascii_case("return=minimal"))
        })
}

pub(crate) fn header_value<'a>(headers: &'a HeaderMap, name: &str) -> Option<&'a str> {
    headers
        .get(name)
        .and_then(|value| value.to_str().ok())
        .map(str::trim)
}

pub(crate) fn parse_producer_integer(name: &str, raw: &str) -> Result<u64, String> {
    const MAX_JS_SAFE_INTEGER: u64 = 9_007_199_254_740_991;
    let value = raw
        .parse::<u64>()
        .map_err(|_| format!("{name} must be a non-negative integer"))?;
    if value > MAX_JS_SAFE_INTEGER {
        return Err(format!("{name} must be <= {MAX_JS_SAFE_INTEGER}"));
    }
    Ok(value)
}

fn runtime_error_response(err: RuntimeError) -> Response {
    let status = runtime_error_status(&err);
    if status.is_server_error() {
        tracing::warn!(%status, error = %err, "runtime request failed");
    }
    let mut headers = HeaderMap::new();
    insert_default_response_headers(&mut headers);
    insert_retry_after_for_temporary(&mut headers, &err);
    insert_producer_error_headers(&mut headers, &err);
    insert_stream_error_headers(&mut headers, &err);
    insert_stream_error_offset(&mut headers, &err);
    (status, headers, err.to_string()).into_response()
}

fn insert_retry_after_for_temporary(headers: &mut HeaderMap, err: &RuntimeError) {
    if err.status() == ErrorStatus::Temporary {
        headers.insert(
            axum::http::header::RETRY_AFTER,
            HeaderValue::from_static("1"),
        );
    }
}

pub(crate) async fn runtime_error_or_leader_redirect_async(
    state: &HttpState,
    err: RuntimeError,
    request_target: &str,
) -> Response {
    let Some(router) = state.client_write_router() else {
        return runtime_error_response(err);
    };
    // Peer URLs are the configured client-reachable leader addresses
    // (`server.listen` when shared, or `server.cluster_listen` when split), so
    // they are valid redirect targets for reads and writes alike. 307 preserves
    // the method and body, so a redirected POST/PUT re-runs as a write on the
    // leader. Writes go through the leader's raft client_write exactly as a
    // local write would; redirecting only moves the leader hop to the client.
    if let Some(redirect) = router.redirect_response(&err, request_target) {
        return redirect;
    }
    // Forward-to-leader error whose leader is currently unknown (election in
    // progress): tell the client to retry rather than failing hard.
    if is_forward_to_leader(&err) {
        return leader_unknown_retry_response(err);
    }
    runtime_error_response(err)
}

/// True when `err` is a group-engine error asking the caller to forward to the
/// leader (carries a leader hint), regardless of whether the leader is yet
/// known.
fn is_forward_to_leader(err: &RuntimeError) -> bool {
    err.leader_hint().is_some()
}

/// 503 + `Retry-After: 1` for a write that hit a non-leader while the group has
/// no known leader. Retryable: a new leader should be elected shortly.
fn leader_unknown_retry_response(err: RuntimeError) -> Response {
    let mut headers = HeaderMap::new();
    insert_default_response_headers(&mut headers);
    headers.insert(
        axum::http::header::RETRY_AFTER,
        HeaderValue::from_static("1"),
    );
    (StatusCode::SERVICE_UNAVAILABLE, headers, err.to_string()).into_response()
}

fn request_target(uri: &Uri) -> String {
    uri.path_and_query()
        .map(|path_and_query| path_and_query.as_str().to_owned())
        .unwrap_or_else(|| uri.path().to_owned())
}

#[cfg(test)]
mod tests;