ff-server 0.6.0

//! REST API layer — thin axum handlers over Server methods.

use std::collections::HashMap;
use std::fmt;
use std::sync::Arc;

use axum::{
    extract::{DefaultBodyLimit, MatchedPath, Path, Query, Request, State},
    http::StatusCode,
    middleware,
    response::{IntoResponse, Response},
    routing::{get, post, put, MethodRouter},
    Json, Router,
};
use serde::{Deserialize, Serialize};
use axum::http::Method;
use tower_http::cors::{AllowOrigin, CorsLayer};
use tower_http::trace::TraceLayer;

use ff_core::contracts::*;
use ff_core::state::PublicState;
use ff_core::types::*;

use crate::config::ConfigError;
use crate::server::{Server, ServerError};

// ── Per-route body-size limits (#97) ──
//
// Axum's out-of-the-box `DefaultBodyLimit` is 2 MiB and was applied silently
// with no per-route override, meaning a misrouted multi-GB JSON would still
// be buffered up to 2 MiB before rejection and mega-byte payloads on
// control-plane endpoints (cancel, priority, claim, …) were accepted with
// no protection beyond that global cap.
//
// We apply three categories, ordered by generosity:
//
//   * `BODY_LIMIT_LARGE_PAYLOAD` (1 MiB) — carries an application
//     `input_payload` (`POST /v1/executions`). FlowFabric has no Lua-side
//     cap on `input_payload` (`execution.lua` SETs whatever is passed);
//     1 MiB matches common workflow-engine conventions (e.g. AWS Step
//     Functions input size limit is 256 KB, Temporal's default is 2 MiB)
//     and is an order of magnitude above typical JSON control envelopes.
//
//   * `BODY_LIMIT_MEDIUM_PAYLOAD` (256 KiB) — signal delivery
//     (`POST /v1/executions/{id}/signal`) carries a `DeliverSignalArgs`
//     with an optional `Vec<u8> payload`. Signals are notification-shaped
//     events (webhooks, human approvals), not bulk-data transfers; 256 KiB
//     is the same ceiling Step Functions imposes on task inputs and is
//     comfortably above any real correlation-id / approval-blob.
//
//   * `BODY_LIMIT_CONTROL` (64 KiB) — everything else. Control-plane JSON
//     bodies (cancel args, priority changes, claim requests, flow
//     members, budget/quota definitions, rotate-secret keys) are all tiny
//     fixed-shape records; 64 KiB leaves plenty of headroom for
//     metadata-heavy bodies while capping DoS amplification by ~32× vs
//     the previous axum default.
//
// Cross-checked against `lua/*.lua` — the only payload-size caps in the
// Lua layer are `CAPS_MAX_BYTES=4096` / `CAPS_MAX_TOKENS=256` for
// capability lists (`lua/helpers.lua`). No other Lua path enforces a
// byte-length limit on incoming payloads, so HTTP is the right layer to
// draw the line.
//
// Future admin routes that need larger bodies (e.g. bulk import) should
// introduce a new `BODY_LIMIT_ADMIN_BULK` const and opt in per-route
// rather than bumping `BODY_LIMIT_CONTROL`.

const BODY_LIMIT_LARGE_PAYLOAD: usize = 1024 * 1024;     // 1 MiB
const BODY_LIMIT_MEDIUM_PAYLOAD: usize = 256 * 1024;     // 256 KiB
const BODY_LIMIT_CONTROL: usize = 64 * 1024;             // 64 KiB

/// Limit metadata attached to a request via extension so the 413 error body
/// can report the route's configured cap. The `DefaultBodyLimit::max(bytes)`
/// layer enforces; this struct only exists for response shape.
#[derive(Clone, Copy)]
struct BodyLimit {
    bytes: usize,
}

/// Apply per-route body-size cap + attach [`BodyLimit`] extension so the
/// JSON rejection handler can report `limit_bytes` in the 413 response.
///
/// Three layers, each guarding a different failure mode:
///
///   1. `from_fn(enforce_body_limit)` — inspects `Content-Length` up front
///      and rejects with our structured 413 body BEFORE the handler runs.
///      This is the primary enforcement path and works regardless of
///      whether the handler consumes the body. Also attaches the
///      [`BodyLimit`] extension so `AppJson`'s 413 branch (below, case 2)
///      can report `limit_bytes` even when rejection happens during
///      streaming.
///   2. `DefaultBodyLimit::max(bytes)` — hint consumed by `AppJson`
///      (`Bytes::from_request` internally) for the case where a client
///      lies in `Content-Length` or uses `Transfer-Encoding: chunked`:
///      the body extractor trips on a streaming-length check and we
///      rewrite the rejection into the structured 413 shape in
///      `AppJson::from_request`.
///   3. `tower_http::limit::RequestBodyLimitLayer::new(bytes)` —
///      transport-level body cap that also fires on chunked transfers
///      even when the handler never consumes the body. Returns a plain
///      413 (no JSON body) in that edge case; the structured path in
///      (1) covers the overwhelmingly common case (clients sending
///      Content-Length).
fn with_body_limit(route: MethodRouter<Arc<Server>>, bytes: usize) -> MethodRouter<Arc<Server>> {
    let r: MethodRouter<Arc<Server>> =
        route.layer(tower_http::limit::RequestBodyLimitLayer::new(bytes));
    let r: MethodRouter<Arc<Server>> = r.layer(DefaultBodyLimit::max(bytes));
    r.layer(middleware::from_fn(
        move |req: Request, next: middleware::Next| enforce_body_limit(bytes, req, next),
    ))
}

/// Content-Length-based body-size enforcement + [`BodyLimit`] extension.
///
/// Runs before the handler so routes whose handlers never consume the
/// request body (`replay_execution`, `revoke_lease`, `reset_budget` — see
/// Copilot review on PR#100) are still protected against large uploads.
async fn enforce_body_limit(bytes: usize, mut req: Request, next: middleware::Next) -> Response {
    if let Some(content_length) = req
        .headers()
        .get(axum::http::header::CONTENT_LENGTH)
        .and_then(|v| v.to_str().ok())
        .and_then(|s| s.parse::<usize>().ok())
        && content_length > bytes
    {
        let route = req
            .extensions()
            .get::<MatchedPath>()
            .map(|m| m.as_str().to_owned())
            .unwrap_or_default();
        let body = PayloadTooLargeBody {
            error: "payload_too_large",
            limit_bytes: bytes,
            route,
        };
        return (StatusCode::PAYLOAD_TOO_LARGE, Json(body)).into_response();
    }
    req.extensions_mut().insert(BodyLimit { bytes });
    next.run(req).await
}

/// Shape for the 413 response body.
#[derive(Serialize)]
struct PayloadTooLargeBody {
    error: &'static str,
    limit_bytes: usize,
    route: String,
}

// ── Custom JSON extractor (uniform JSON error on malformed body) ──

struct AppJson<T>(T);

impl<S, T> axum::extract::FromRequest<S> for AppJson<T>
where
    T: serde::de::DeserializeOwned + Send,
    S: Send + Sync,
{
    type Rejection = Response;

    async fn from_request(
        req: axum::extract::Request,
        state: &S,
    ) -> Result<Self, Self::Rejection> {
        // Snapshot metadata for a potential 413 before `Json::from_request`
        // consumes the request. `BodyLimit` is `Copy` and `MatchedPath` is
        // cheap to clone (its payload is an `Arc<str>`), so the non-413
        // path pays at most two extension lookups plus a ref-count bump.
        let limit = req.extensions().get::<BodyLimit>().copied();
        let matched = req.extensions().get::<MatchedPath>().cloned();

        match Json::<T>::from_request(req, state).await {
            Ok(Json(value)) => Ok(AppJson(value)),
            Err(rejection) => {
                let status = rejection.status();
                tracing::debug!(detail = %rejection.body_text(), "JSON rejection");
                if status == StatusCode::PAYLOAD_TOO_LARGE {
                    let limit_bytes = limit.map(|l| l.bytes).unwrap_or(0);
                    let route = matched
                        .as_ref()
                        .map(|m| m.as_str().to_owned())
                        .unwrap_or_default();
                    let body = PayloadTooLargeBody {
                        error: "payload_too_large",
                        limit_bytes,
                        route,
                    };
                    return Err((status, Json(body)).into_response());
                }
                let body = ErrorBody::plain(format!(
                    "invalid JSON: {}",
                    status.canonical_reason().unwrap_or("bad request"),
                ));
                Err((status, Json(body)).into_response())
            }
        }
    }
}

// ── Error handling ──

struct ApiError(ServerError);

impl From<ServerError> for ApiError {
    fn from(e: ServerError) -> Self {
        Self(e)
    }
}

/// HTTP error body. `kind`/`retryable` are populated for 500s backed by
/// a backend transport fault (see `ff_core::BackendErrorKind`) so HTTP
/// clients (e.g. cairn-fabric) can make retry decisions without parsing
/// the `error` string.
#[derive(Serialize)]
struct ErrorBody {
    error: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    kind: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    retryable: Option<bool>,
}

impl ErrorBody {
    fn plain(error: String) -> Self {
        Self { error, kind: None, retryable: None }
    }
}

impl IntoResponse for ApiError {
    fn into_response(self) -> Response {
        let (status, body) = match &self.0 {
            ServerError::NotFound(msg) => {
                (StatusCode::NOT_FOUND, ErrorBody::plain(msg.clone()))
            }
            ServerError::InvalidInput(msg) => {
                (StatusCode::BAD_REQUEST, ErrorBody::plain(msg.clone()))
            }
            ServerError::OperationFailed(msg) => {
                (StatusCode::BAD_REQUEST, ErrorBody::plain(msg.clone()))
            }
            ServerError::ConcurrencyLimitExceeded(source, max) => (
                StatusCode::TOO_MANY_REQUESTS,
                ErrorBody {
                    error: format!(
                        "too many concurrent {source} calls (server max: {max}); retry with backoff"
                    ),
                    kind: None,
                    retryable: Some(true),
                },
            ),
            ServerError::Backend(be) => {
                let kind_str = be.kind().as_stable_str();
                tracing::error!(
                    kind = kind_str,
                    message = be.message(),
                    "backend error"
                );
                (
                    StatusCode::INTERNAL_SERVER_ERROR,
                    ErrorBody {
                        error: self.0.to_string(),
                        kind: Some(kind_str.to_owned()),
                        retryable: Some(self.0.is_retryable()),
                    },
                )
            }
            ServerError::BackendContext { source, context } => {
                let kind_str = source.kind().as_stable_str();
                tracing::error!(
                    kind = kind_str,
                    message = source.message(),
                    context = %context,
                    "backend error"
                );
                (
                    StatusCode::INTERNAL_SERVER_ERROR,
                    ErrorBody {
                        error: self.0.to_string(),
                        kind: Some(kind_str.to_owned()),
                        retryable: Some(self.0.is_retryable()),
                    },
                )
            }
            ServerError::LibraryLoad(load_err) => {
                let kind_str = load_err
                    .valkey_kind()
                    .map(ff_backend_valkey::classify_ferriskey_kind)
                    .map(|k| k.as_stable_str());
                tracing::error!(
                    kind = kind_str.unwrap_or(""),
                    error = %load_err,
                    "library load failure"
                );
                (
                    StatusCode::INTERNAL_SERVER_ERROR,
                    ErrorBody {
                        error: format!("library load: {load_err}"),
                        kind: kind_str.map(str::to_owned),
                        retryable: Some(self.0.is_retryable()),
                    },
                )
            }
            // Script / Config / PartitionMismatch — developer or deployment
            // errors. No Valkey ErrorKind to surface, but retryable=false is
            // informative: a client-side retry won't change the outcome.
            other => (
                StatusCode::INTERNAL_SERVER_ERROR,
                ErrorBody {
                    error: other.to_string(),
                    kind: None,
                    retryable: Some(false),
                },
            ),
        };
        (status, Json(body)).into_response()
    }
}

// ── Router ──

pub fn router(
    server: Arc<Server>,
    cors_origins: &[String],
    api_token: Option<String>,
) -> Result<Router, ConfigError> {
    router_with_metrics(server, cors_origins, api_token, None)
}

/// Router entry point that also mounts `/metrics` and the HTTP metrics
/// middleware. Used by `main.rs` when the `observability` feature is on.
///
/// When `metrics` is `Some` AND the `observability` feature is compiled
/// in, `/metrics` is mounted on an un-authenticated nested router (auth
/// middleware does not apply — Prometheus convention). When the
/// feature is off OR `metrics` is `None`, no `/metrics` route exists
/// (returns 404) and no HTTP metrics middleware is installed.
pub fn router_with_metrics(
    server: Arc<Server>,
    cors_origins: &[String],
    api_token: Option<String>,
    #[cfg_attr(not(feature = "observability"), allow(unused_variables))]
    metrics: Option<Arc<crate::Metrics>>,
) -> Result<Router, ConfigError> {
    let auth_enabled = api_token.is_some();
    let cors = build_cors_layer(cors_origins, auth_enabled)?;

    // Per-route body-size caps (#97). See module-level `BODY_LIMIT_*` consts
    // for the category rationale; each route picks the tightest cap that
    // still accommodates expected real traffic.
    let mut app = Router::new()
        // Executions
        .route(
            "/v1/executions",
            with_body_limit(
                get(list_executions).post(create_execution),
                BODY_LIMIT_LARGE_PAYLOAD,
            ),
        )
        .route("/v1/executions/{id}", get(get_execution))
        .route("/v1/executions/{id}/state", get(get_execution_state))
        .route(
            "/v1/executions/{id}/pending-waitpoints",
            get(list_pending_waitpoints),
        )
        .route("/v1/executions/{id}/result", get(get_execution_result))
        .route(
            "/v1/executions/{id}/cancel",
            with_body_limit(post(cancel_execution), BODY_LIMIT_CONTROL),
        )
        .route(
            "/v1/executions/{id}/signal",
            with_body_limit(post(deliver_signal), BODY_LIMIT_MEDIUM_PAYLOAD),
        )
        .route(
            "/v1/executions/{id}/priority",
            with_body_limit(put(change_priority), BODY_LIMIT_CONTROL),
        )
        .route(
            "/v1/executions/{id}/replay",
            with_body_limit(post(replay_execution), BODY_LIMIT_CONTROL),
        )
        .route(
            "/v1/executions/{id}/revoke-lease",
            with_body_limit(post(revoke_lease), BODY_LIMIT_CONTROL),
        )
        // Scheduler-routed claim (Batch C item 2). Worker POSTs lane +
        // identity + capabilities; server runs budget/quota/capability
        // admission via ff-scheduler and returns a ClaimGrant on
        // success (204 No Content when no eligible execution).
        .route(
            "/v1/workers/{worker_id}/claim",
            with_body_limit(post(claim_for_worker), BODY_LIMIT_CONTROL),
        )
        // Stream read + tail (RFC-006 #2)
        .route(
            "/v1/executions/{id}/attempts/{idx}/stream",
            get(read_attempt_stream),
        )
        .route(
            "/v1/executions/{id}/attempts/{idx}/stream/tail",
            get(tail_attempt_stream),
        )
        // Flows
        .route(
            "/v1/flows",
            with_body_limit(post(create_flow), BODY_LIMIT_CONTROL),
        )
        .route(
            "/v1/flows/{id}/members",
            with_body_limit(post(add_execution_to_flow), BODY_LIMIT_CONTROL),
        )
        .route(
            "/v1/flows/{id}/cancel",
            with_body_limit(post(cancel_flow), BODY_LIMIT_CONTROL),
        )
        .route(
            "/v1/flows/{id}/edges",
            with_body_limit(post(stage_dependency_edge), BODY_LIMIT_CONTROL),
        )
        .route(
            "/v1/flows/{id}/edges/apply",
            with_body_limit(post(apply_dependency_to_child), BODY_LIMIT_CONTROL),
        )
        // Budgets
        .route(
            "/v1/budgets",
            with_body_limit(post(create_budget), BODY_LIMIT_CONTROL),
        )
        .route("/v1/budgets/{id}", get(get_budget_status))
        .route(
            "/v1/budgets/{id}/usage",
            with_body_limit(post(report_usage), BODY_LIMIT_CONTROL),
        )
        .route(
            "/v1/budgets/{id}/reset",
            with_body_limit(post(reset_budget), BODY_LIMIT_CONTROL),
        )
        // Quotas
        .route(
            "/v1/quotas",
            with_body_limit(post(create_quota_policy), BODY_LIMIT_CONTROL),
        )
        // Admin: rotate waitpoint HMAC secret (RFC-004 §Waitpoint Security)
        .route(
            "/v1/admin/rotate-waitpoint-secret",
            with_body_limit(post(rotate_waitpoint_secret), BODY_LIMIT_CONTROL),
        )
        // Health (always unauthenticated)
        .route("/healthz", get(healthz));

    if let Some(token) = api_token {
        let token = Arc::new(token);
        app = app.layer(middleware::from_fn(move |req, next| {
            let token = token.clone();
            auth_middleware(token, req, next)
        }));
    }

    // PR-94: HTTP metrics middleware. Recorded AFTER the auth layer
    // above so unauthorized 401s are still counted under their route,
    // and BEFORE trace/cors so the metric captures handler time
    // including the auth check itself. No-op when the
    // `observability` feature is off.
    #[cfg(feature = "observability")]
    if let Some(m) = metrics.as_ref() {
        let m = m.clone();
        app = app.layer(middleware::from_fn_with_state(
            m,
            crate::metrics::http_middleware,
        ));
    }

    #[cfg_attr(not(feature = "observability"), allow(unused_mut))]
    let mut app = app
        .layer(TraceLayer::new_for_http())
        .layer(cors)
        .with_state(server);

    // PR-94: `/metrics` — intentionally unauthenticated. Mounted on a
    // separate router with its own State so the main app's auth
    // middleware does not run for scrapes (Prometheus convention).
    // Network-layer auth (ingress ACL, service-mesh policy, or
    // metrics-only listen address) is the expected gate; FF does not
    // own auth for scrape endpoints.
    #[cfg(feature = "observability")]
    if let Some(m) = metrics {
        let metrics_router: Router = Router::new()
            .route("/metrics", get(crate::metrics::metrics_handler))
            .with_state(m);
        app = app.merge(metrics_router);
    }

    Ok(app)
}

async fn auth_middleware(
    token: Arc<String>,
    req: Request,
    next: middleware::Next,
) -> Response {
    if req.uri().path() == "/healthz" {
        return next.run(req).await;
    }

    let auth_header = req
        .headers()
        .get("authorization")
        .and_then(|v| v.to_str().ok());

    let authorized = auth_header
        .and_then(|v| v.strip_prefix("Bearer "))
        .is_some_and(|t| constant_time_eq(t.as_bytes(), token.as_bytes()));

    if authorized {
        next.run(req).await
    } else {
        (
            StatusCode::UNAUTHORIZED,
            Json(ErrorBody::plain(
                "missing or invalid Authorization header".to_owned(),
            )),
        )
            .into_response()
    }
}

fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
    if a.len() != b.len() {
        return false;
    }
    let mut diff = 0u8;
    for (x, y) in a.iter().zip(b.iter()) {
        diff |= x ^ y;
    }
    diff == 0
}

/// Build the CORS layer from the configured `FF_CORS_ORIGINS` list.
///
/// Fails closed (#71): if every entry fails to parse as a `HeaderValue`,
/// return `ConfigError` so startup aborts instead of silently falling back
/// to `CorsLayer::permissive()` (a typo would otherwise broaden browser
/// access to "allow any origin").
///
/// When `auth_enabled` is true (i.e. `FF_API_TOKEN` is set), `Authorization`
/// is added to `allow_headers` (#66) so browser preflights for
/// cross-origin authenticated requests succeed.
fn build_cors_layer(origins: &[String], auth_enabled: bool) -> Result<CorsLayer, ConfigError> {
    if origins.iter().any(|o| o == "*") {
        return Ok(CorsLayer::permissive());
    }

    let mut parsed = Vec::with_capacity(origins.len());
    let mut accepted = Vec::with_capacity(origins.len());
    let mut invalid = Vec::new();
    for o in origins {
        match o.parse() {
            Ok(v) => {
                parsed.push(v);
                accepted.push(o.as_str());
            }
            Err(_) => invalid.push(o.clone()),
        }
    }

    if parsed.is_empty() && !origins.is_empty() {
        return Err(ConfigError::InvalidValue {
            var: "FF_CORS_ORIGINS".to_owned(),
            message: format!(
                "all configured origins failed to parse as valid HTTP header values: {:?}; \
                 refusing to fall back to permissive CORS",
                origins
            ),
        });
    }

    if !invalid.is_empty() {
        // Collected `accepted` list during the single parse loop above to
        // avoid an O(N*M) filter-contains scan per log event.
        tracing::warn!(
            ?invalid,
            ?accepted,
            "some FF_CORS_ORIGINS entries failed to parse and were dropped"
        );
    }

    // Prefer typed `header::*` constants over stringly-typed `from_static`
    // so typos fail to compile rather than fail at runtime.
    let mut headers = vec![axum::http::header::CONTENT_TYPE];
    if auth_enabled {
        headers.push(axum::http::header::AUTHORIZATION);
    }

    Ok(CorsLayer::new()
        .allow_origin(AllowOrigin::list(parsed))
        .allow_methods([Method::GET, Method::POST, Method::PUT])
        .allow_headers(headers))
}

// ── Execution handlers ──

#[derive(Deserialize)]
struct ListExecutionsParams {
    /// Partition index (`u16`) to enumerate. Serves as the partition
    /// key for the forward-only cursor listing.
    partition: u16,
    /// Exclusive cursor: start listing strictly after this execution
    /// id. Omit for the first page.
    #[serde(default)]
    cursor: Option<String>,
    #[serde(default = "default_limit")]
    limit: u64,
}

fn default_limit() -> u64 { 50 }

/// List executions in one partition with forward-only cursor
/// pagination.
///
/// **Breaking change (unreleased HTTP surface, not on crates.io):** as
/// of issue #182 this endpoint is a thin forwarder onto
/// [`ff_core::engine_backend::EngineBackend::list_executions`]. The
/// previous offset + lane + state filter query parameters were
/// dropped; the endpoint now returns partition-scoped execution ids
/// with an opaque `next_cursor`.
///
/// Request: `GET /v1/executions?partition=<u16>&cursor=<eid>&limit=<usize>`.
/// Response: `{ "executions": ["<eid>", ...], "next_cursor": "<eid>" | null }`.
async fn list_executions(
    State(server): State<Arc<Server>>,
    Query(params): Query<ListExecutionsParams>,
) -> Result<Json<ListExecutionsPage>, ApiError> {
    let limit = params.limit.min(1000) as usize;
    let cursor = match params.cursor {
        Some(raw) if !raw.is_empty() => Some(
            ff_core::types::ExecutionId::parse(&raw).map_err(|e| {
                ApiError::from(ServerError::InvalidInput(format!(
                    "invalid cursor: {e}"
                )))
            })?,
        ),
        _ => None,
    };
    let result = server
        .list_executions_page(params.partition, cursor, limit)
        .await?;
    Ok(Json(result))
}

async fn create_execution(
    State(server): State<Arc<Server>>,
    AppJson(args): AppJson<CreateExecutionArgs>,
) -> Result<(StatusCode, Json<CreateExecutionResult>), ApiError> {
    let result = server.create_execution(&args).await?;
    let status = match &result {
        CreateExecutionResult::Created { .. } => StatusCode::CREATED,
        CreateExecutionResult::Duplicate { .. } => StatusCode::OK,
    };
    Ok((status, Json(result)))
}

async fn get_execution(
    State(server): State<Arc<Server>>,
    Path(id): Path<String>,
) -> Result<Json<ExecutionInfo>, ApiError> {
    let eid = parse_execution_id(&id)?;
    Ok(Json(server.get_execution(&eid).await?))
}

async fn get_execution_state(
    State(server): State<Arc<Server>>,
    Path(id): Path<String>,
) -> Result<Json<PublicState>, ApiError> {
    let eid = parse_execution_id(&id)?;
    Ok(Json(server.get_execution_state(&eid).await?))
}

/// Returns the actionable (`pending`/`active`) waitpoints for an
/// execution, including the HMAC `waitpoint_token` required to deliver
/// signals. Human reviewers use this to look up the token originally
/// returned only to the suspending worker's `SuspendOutcome`.
///
/// SECURITY: `waitpoint_token` is a bearer credential for signal
/// delivery; leaking it lets a third party forge authority to resume or
/// influence the execution. Gate the endpoint behind `FF_API_TOKEN` in
/// any deployment reachable from untrusted networks. The auth middleware
/// only mounts when `FF_API_TOKEN` is set; this endpoint is
/// unauthenticated without it, and the server logs a loud warning at
/// startup so operators notice.
async fn list_pending_waitpoints(
    State(server): State<Arc<Server>>,
    Path(id): Path<String>,
) -> Result<Json<Vec<PendingWaitpointInfo>>, ApiError> {
    let eid = parse_execution_id(&id)?;
    Ok(Json(server.list_pending_waitpoints(&eid).await?))
}

/// Returns the raw result payload bytes written by the worker's
/// `ff_complete_execution` call. 404 when the execution has no stored
/// result (missing entirely, still in-flight, or trimmed by retention —
/// see below).
///
/// # Ordering (required)
///
/// Callers MUST poll `GET /v1/executions/{id}/state` until it returns
/// `completed` before fetching `/result`. Early polls may return 404
/// because completion writes `public_state = completed` and the result
/// `SET` in the same atomic Lua; in the normal path the window is
/// effectively zero, but network round-trip ordering between a state
/// poll and a result fetch can make the result appear briefly absent
/// during replay (`ff_replay_execution`).
///
/// # Retention / 404 after completed
///
/// `get_execution_state == completed` is authoritative for completion.
/// This endpoint additionally depends on the result bytes not having
/// been trimmed — v1 sets no retention policy, so
/// `state = completed` should always pair with a 200 here. Any
/// future retention-policy feature must call this contract out in its
/// own docs.
///
/// CONTENT-TYPE: `application/octet-stream`. The server is payload-format
/// agnostic — workers choose the encoding via the SDK's `complete(bytes)`
/// call, and callers must know the contract. The media-pipeline example
/// uses JSON by convention (`serde_json::to_vec(&Result)`); adapters can
/// pick any binary format.
///
/// SECURITY: completion payloads can contain PII (e.g. LLM summaries of
/// user audio). Treat this endpoint like any other read — gate behind
/// `FF_API_TOKEN` in any deployment reachable from untrusted networks.
/// The auth middleware only mounts when `FF_API_TOKEN` is set.
async fn get_execution_result(
    State(server): State<Arc<Server>>,
    Path(id): Path<String>,
) -> Result<Response, ApiError> {
    let eid = parse_execution_id(&id)?;
    match server.get_execution_result(&eid).await? {
        Some(bytes) => Ok((
            StatusCode::OK,
            [(axum::http::header::CONTENT_TYPE, "application/octet-stream")],
            bytes,
        )
            .into_response()),
        None => Err(ApiError(ServerError::NotFound(format!(
            "execution result not found: {eid}"
        )))),
    }
}

async fn cancel_execution(
    State(server): State<Arc<Server>>,
    Path(id): Path<String>,
    AppJson(mut args): AppJson<CancelExecutionArgs>,
) -> Result<Json<CancelExecutionResult>, ApiError> {
    let path_eid = parse_execution_id(&id)?;
    check_id_match(&path_eid, &args.execution_id, "execution_id")?;
    args.execution_id = path_eid;
    Ok(Json(server.cancel_execution(&args).await?))
}

async fn deliver_signal(
    State(server): State<Arc<Server>>,
    Path(id): Path<String>,
    AppJson(mut args): AppJson<DeliverSignalArgs>,
) -> Result<Json<DeliverSignalResult>, ApiError> {
    let path_eid = parse_execution_id(&id)?;
    check_id_match(&path_eid, &args.execution_id, "execution_id")?;
    args.execution_id = path_eid;
    Ok(Json(server.deliver_signal(&args).await?))
}

// ── Admin: rotate waitpoint HMAC secret (RFC-004 §Waitpoint Security) ──

#[derive(Deserialize)]
struct RotateWaitpointSecretBody {
    new_kid: String,
    /// Hex-encoded new secret. Even-length, 0-9a-fA-F.
    new_secret_hex: String,
}

/// Hard ceiling on how long the rotate endpoint runs before the HTTP
/// handler bails. Rotation touches every execution partition (up to 256)
/// with HGET+HMGET+HDEL+HSET per partition; 6 round-trips × 30ms cross-AZ
/// × 256 partitions ≈ 46s worst-case. The internal SETNX lock is 10s TTL
/// per partition, so 120s gives ample margin for contention + slow RTTs
/// while staying below common LB idle timeouts (ALB 60s default, but
/// typically bumped to 120s+ for admin endpoints).
///
/// On timeout: returns HTTP 504 immediately. Valkey-side work may still
/// finish (the per-partition locks and HSETs are already in flight). The
/// operator observes a 504 and retries; retry is SAFE — rotation is
/// idempotent per-partition (same new_kid + same secret → no-op on
/// already-rotated partitions).
const ROTATE_HTTP_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(120);

async fn rotate_waitpoint_secret(
    State(server): State<Arc<Server>>,
    AppJson(body): AppJson<RotateWaitpointSecretBody>,
) -> Result<Response, ApiError> {
    // Cap the whole endpoint end-to-end. If this trips, the caller's
    // retry is SAFE — per-partition rotation is idempotent on the same
    // (new_kid, secret_hex) and the per-partition SETNX lock prevents
    // double-rotation under concurrent retries.
    let rotate_fut = server.rotate_waitpoint_secret(&body.new_kid, &body.new_secret_hex);
    let result = match tokio::time::timeout(ROTATE_HTTP_TIMEOUT, rotate_fut).await {
        Ok(r) => r?,
        Err(_) => {
            tracing::error!(
                target: "audit",
                new_kid = %body.new_kid,
                timeout_s = ROTATE_HTTP_TIMEOUT.as_secs(),
                "waitpoint_hmac_rotation_timeout_http_504"
            );
            let body = ErrorBody::plain(format!(
                "rotation exceeded {}s server-side timeout; retry is safe \
                 (per-partition rotation is idempotent on the same new_kid + secret_hex)",
                ROTATE_HTTP_TIMEOUT.as_secs()
            ));
            return Ok((StatusCode::GATEWAY_TIMEOUT, Json(body)).into_response());
        }
    };
    // Operator action log at audit target (per-partition detail logged inside
    // rotate_waitpoint_secret). Returns 400 only on actionable fault states.
    //
    // Two distinct rotated==0 cases:
    //   - failed.is_empty() → no partitions at all (num_flow_partitions == 0;
    //     env_u16_positive rejects this at boot so this is mostly dead code
    //     for library/Default callers).
    //   - !failed.is_empty() → every partition attempt raised a real error.
    //     Operator investigates Valkey/auth/cluster health before retrying.
    if result.rotated == 0 && result.failed.is_empty() {
        return Err(ApiError::from(ServerError::OperationFailed(
            "rotation had no partitions to operate on \
             (num_flow_partitions is 0 — server misconfigured)"
                .to_owned(),
        )));
    }
    if result.rotated == 0 && !result.failed.is_empty() {
        return Err(ApiError::from(ServerError::OperationFailed(
            "rotation failed on all partitions (check Valkey connectivity)".to_owned(),
        )));
    }
    Ok(Json(result).into_response())
}

#[derive(Deserialize)]
struct ChangePriorityBody {
    new_priority: i32,
}

async fn change_priority(
    State(server): State<Arc<Server>>,
    Path(id): Path<String>,
    AppJson(body): AppJson<ChangePriorityBody>,
) -> Result<Json<ChangePriorityResult>, ApiError> {
    let eid = parse_execution_id(&id)?;
    Ok(Json(server.change_priority(&eid, body.new_priority).await?))
}

async fn replay_execution(
    State(server): State<Arc<Server>>,
    Path(id): Path<String>,
) -> Result<Json<ReplayExecutionResult>, ApiError> {
    let eid = parse_execution_id(&id)?;
    Ok(Json(server.replay_execution(&eid).await?))
}

async fn revoke_lease(
    State(server): State<Arc<Server>>,
    Path(id): Path<String>,
) -> Result<Json<RevokeLeaseResult>, ApiError> {
    let eid = parse_execution_id(&id)?;
    Ok(Json(server.revoke_lease(&eid).await?))
}

// ── Scheduler-routed claim (Batch C item 2 PR-B) ──
//
// The server exposes the scheduler's `claim_for_worker` cycle via
// HTTP so ff-sdk workers can acquire claim grants without enabling
// the `direct-valkey-claim` feature. The request body carries lane +
// identity + capabilities; the server returns a serialized
// `ClaimGrant` (or 204 No Content when no eligible execution exists).

/// Request body for `POST /v1/workers/{worker_id}/claim`.
#[derive(Deserialize)]
struct ClaimForWorkerBody {
    lane_id: String,
    worker_instance_id: String,
    /// Capability tokens this worker advertises. Sorted + validated
    /// on the scheduler side; any non-printable/CSV-breaking token
    /// surfaces as 400.
    #[serde(default)]
    capabilities: Vec<String>,
    /// Grant TTL in milliseconds. Bounded so a worker can't request a
    /// multi-hour grant and squat the execution.
    grant_ttl_ms: u64,
}

/// Wire shape for `ff_core::contracts::ClaimGrant`. Carries the
/// opaque [`ff_core::partition::PartitionKey`] on the wire; consumers
/// never see the internal `PartitionFamily` enum (issue #91).
#[derive(Serialize)]
struct ClaimGrantDto {
    execution_id: String,
    partition_key: ff_core::partition::PartitionKey,
    grant_key: String,
    expires_at_ms: u64,
}

impl From<ff_core::contracts::ClaimGrant> for ClaimGrantDto {
    fn from(g: ff_core::contracts::ClaimGrant) -> Self {
        Self {
            execution_id: g.execution_id.to_string(),
            partition_key: g.partition_key,
            grant_key: g.grant_key,
            expires_at_ms: g.expires_at_ms,
        }
    }
}

/// Maximum grant TTL accepted via HTTP. Mirrors the scheduler's
/// internal ceiling so a misconfigured worker can't squat an
/// execution on a multi-hour grant.
const CLAIM_GRANT_TTL_MS_MAX: u64 = 60_000;

/// Reject empty / whitespace / non-printable identifiers the way
/// [`LaneId::try_new`] does for lanes. WorkerId + WorkerInstanceId
/// feed into scheduler scan jitter + Valkey key construction; silent
/// acceptance of "" or "w\nork" would either mis-key or mis-hash.
fn validate_identifier(field: &str, value: &str) -> Result<(), ApiError> {
    if value.is_empty() {
        return Err(ApiError(ServerError::InvalidInput(format!(
            "{field}: must not be empty"
        ))));
    }
    if value.len() > 256 {
        return Err(ApiError(ServerError::InvalidInput(format!(
            "{field}: exceeds 256 bytes (got {})",
            value.len()
        ))));
    }
    if value.chars().any(|c| c.is_control() || c.is_whitespace()) {
        return Err(ApiError(ServerError::InvalidInput(format!(
            "{field}: must not contain whitespace or control characters"
        ))));
    }
    Ok(())
}

async fn claim_for_worker(
    State(server): State<Arc<Server>>,
    Path(worker_id): Path<String>,
    AppJson(body): AppJson<ClaimForWorkerBody>,
) -> Result<Response, ApiError> {
    validate_identifier("worker_id", &worker_id)?;
    validate_identifier("worker_instance_id", &body.worker_instance_id)?;
    let worker_id = WorkerId::new(worker_id);
    let worker_instance_id = WorkerInstanceId::new(body.worker_instance_id);
    let lane = LaneId::try_new(body.lane_id).map_err(|e| {
        ApiError(ServerError::InvalidInput(format!("lane_id: {e}")))
    })?;
    if body.grant_ttl_ms == 0 || body.grant_ttl_ms > CLAIM_GRANT_TTL_MS_MAX {
        return Err(ApiError(ServerError::InvalidInput(format!(
            "grant_ttl_ms must be in 1..={CLAIM_GRANT_TTL_MS_MAX}"
        ))));
    }
    let caps: std::collections::BTreeSet<String> =
        body.capabilities.into_iter().collect();

    match server
        .claim_for_worker(
            &lane,
            &worker_id,
            &worker_instance_id,
            &caps,
            body.grant_ttl_ms,
        )
        .await?
    {
        Some(grant) => Ok((StatusCode::OK, Json(ClaimGrantDto::from(grant))).into_response()),
        None => Ok(StatusCode::NO_CONTENT.into_response()),
    }
}

// ── Stream read + tail ──

#[derive(Deserialize)]
struct ReadStreamParams {
    #[serde(default = "ff_core::contracts::StreamCursor::start")]
    from: ff_core::contracts::StreamCursor,
    #[serde(default = "ff_core::contracts::StreamCursor::end")]
    to: ff_core::contracts::StreamCursor,
    #[serde(default = "default_read_limit")]
    limit: u64,
}

fn default_read_limit() -> u64 { 100 }

#[derive(Serialize)]
struct ReadStreamResponse {
    frames: Vec<StreamFrame>,
    count: usize,
    /// When set, the producer has closed this stream — consumer should
    /// stop polling. Absent when the stream is still open (or never
    /// existed, which is indistinguishable from "still open" at this
    /// layer).
    #[serde(skip_serializing_if = "Option::is_none")]
    closed_at: Option<i64>,
    /// Reason from the closing writer: `attempt_success`, `attempt_failure`,
    /// `attempt_cancelled`, `attempt_interrupted`. Absent iff still open.
    #[serde(skip_serializing_if = "Option::is_none")]
    closed_reason: Option<String>,
}

impl From<ff_core::contracts::StreamFrames> for ReadStreamResponse {
    fn from(sf: ff_core::contracts::StreamFrames) -> Self {
        let count = sf.frames.len();
        Self {
            frames: sf.frames,
            count,
            closed_at: sf.closed_at.map(|t| t.0),
            closed_reason: sf.closed_reason,
        }
    }
}

/// REST-layer ceiling on `limit` for stream read/tail responses. Lower
/// than the internal `STREAM_READ_HARD_CAP` (10_000) because an HTTP
/// response buffers the whole JSON body in memory in axum — a
/// `10_000 × max_payload_bytes (65_536)` body is ~640MB per call, which
/// is a DoS vector from a single client. Internal callers using FCALL or
/// the SDK directly still get the full 10_000 ceiling; REST clients must
/// paginate through `from`/`to` for larger spans.
///
/// v2 candidate: chunked-transfer / SSE when the caller wants > this bound.
const REST_STREAM_LIMIT_CEILING: u64 = 1_000;

async fn read_attempt_stream(
    State(server): State<Arc<Server>>,
    Path((id, idx)): Path<(String, u32)>,
    Query(params): Query<ReadStreamParams>,
) -> Result<Json<ReadStreamResponse>, ApiError> {
    if params.limit == 0 {
        return Err(ApiError(ServerError::InvalidInput(
            "limit must be >= 1".to_owned(),
        )));
    }
    if params.limit > REST_STREAM_LIMIT_CEILING {
        return Err(ApiError(ServerError::InvalidInput(format!(
            "limit exceeds REST ceiling {REST_STREAM_LIMIT_CEILING}; paginate via from/to for larger spans"
        ))));
    }
    let eid = parse_execution_id(&id)?;
    let attempt_index = AttemptIndex::new(idx);
    let result = server
        .read_attempt_stream(
            &eid,
            attempt_index,
            params.from.to_wire(),
            params.to.to_wire(),
            params.limit,
        )
        .await?;
    Ok(Json(result.into()))
}

#[derive(Deserialize)]
struct TailStreamParams {
    #[serde(default = "ff_core::contracts::StreamCursor::beginning")]
    after: ff_core::contracts::StreamCursor,
    #[serde(default)]
    block_ms: u64,
    #[serde(default = "default_tail_limit")]
    limit: u64,
}

fn default_tail_limit() -> u64 { 50 }

/// Ceiling on BLOCK duration for the tail endpoint. Kept below common LB
/// idle timeouts (ALB 60s, nginx 60s, Cloudflare 100s) so the HTTP response
/// can't be cut mid-block.
///
/// Note: ferriskey's client auto-extends its `request_timeout` for XREAD
/// BLOCK to `block_ms + 500ms`, so a blocking call with the full ceiling
/// never produces a spurious transport timeout. See
/// `ff_script::stream_tail` module docs for the exact ferriskey code path.
const MAX_TAIL_BLOCK_MS: u64 = 30_000;

async fn tail_attempt_stream(
    State(server): State<Arc<Server>>,
    Path((id, idx)): Path<(String, u32)>,
    Query(params): Query<TailStreamParams>,
) -> Result<Json<ReadStreamResponse>, ApiError> {
    if params.block_ms > MAX_TAIL_BLOCK_MS {
        return Err(ApiError(ServerError::InvalidInput(format!(
            "block_ms exceeds {MAX_TAIL_BLOCK_MS}ms ceiling"
        ))));
    }
    if params.limit == 0 {
        return Err(ApiError(ServerError::InvalidInput(
            "limit must be >= 1".to_owned(),
        )));
    }
    if params.limit > REST_STREAM_LIMIT_CEILING {
        return Err(ApiError(ServerError::InvalidInput(format!(
            "limit exceeds REST ceiling {REST_STREAM_LIMIT_CEILING}; paginate via after for larger spans"
        ))));
    }
    // XREAD cursor must be a concrete ID — `Start`/`End` are
    // XRANGE-only. The opaque-cursor deserializer already rejects
    // the bare `-`/`+` wire tokens; this boundary also rejects the
    // structured `start`/`end` keywords because XREAD treats them
    // as invalid ids. Uses [`StreamCursor::is_concrete`] so the
    // SDK + REST guards stay in lock-step.
    if !params.after.is_concrete() {
        return Err(ApiError(ServerError::InvalidInput(
            "after: XREAD cursor must be a concrete entry id; pass '0-0' to start from the beginning"
                .to_owned(),
        )));
    }

    let eid = parse_execution_id(&id)?;
    let attempt_index = AttemptIndex::new(idx);
    let result = server
        .tail_attempt_stream(
            &eid,
            attempt_index,
            params.after.to_wire(),
            params.block_ms,
            params.limit,
        )
        .await?;
    Ok(Json(result.into()))
}

// ── Flow handlers ──

async fn create_flow(
    State(server): State<Arc<Server>>,
    AppJson(args): AppJson<CreateFlowArgs>,
) -> Result<(StatusCode, Json<CreateFlowResult>), ApiError> {
    let result = server.create_flow(&args).await?;
    let status = match &result {
        CreateFlowResult::Created { .. } => StatusCode::CREATED,
        CreateFlowResult::AlreadySatisfied { .. } => StatusCode::OK,
    };
    Ok((status, Json(result)))
}

async fn add_execution_to_flow(
    State(server): State<Arc<Server>>,
    Path(id): Path<String>,
    AppJson(mut args): AppJson<AddExecutionToFlowArgs>,
) -> Result<(StatusCode, Json<AddExecutionToFlowResult>), ApiError> {
    let path_fid = parse_flow_id(&id)?;
    check_id_match(&path_fid, &args.flow_id, "flow_id")?;
    args.flow_id = path_fid;
    let result = server.add_execution_to_flow(&args).await?;
    let status = match &result {
        AddExecutionToFlowResult::Added { .. } => StatusCode::CREATED,
        AddExecutionToFlowResult::AlreadyMember { .. } => StatusCode::OK,
    };
    Ok((status, Json(result)))
}

/// Cancel a flow.
///
/// By default the handler returns immediately with
/// [`CancelFlowResult::CancellationScheduled`] (or `Cancelled` for flows
/// with no members / non-cancel_all policies), and the individual member
/// execution cancellations run in a background task on the server.
/// Clients can track per-member progress by polling
/// `GET /v1/executions/{id}/state` for each id in `member_execution_ids`.
///
/// Pass `?wait=true` to run the dispatch loop inline; the handler will not
/// return until every member has been cancelled. Useful for tests and
/// callers that need synchronous completion.
async fn cancel_flow(
    State(server): State<Arc<Server>>,
    Path(id): Path<String>,
    Query(params): Query<HashMap<String, String>>,
    AppJson(mut args): AppJson<CancelFlowArgs>,
) -> Result<Json<CancelFlowResult>, ApiError> {
    let path_fid = parse_flow_id(&id)?;
    check_id_match(&path_fid, &args.flow_id, "flow_id")?;
    args.flow_id = path_fid;
    let wait = params.get("wait").is_some_and(|v| v == "true" || v == "1");
    let result = if wait {
        server.cancel_flow_wait(&args).await?
    } else {
        server.cancel_flow(&args).await?
    };
    Ok(Json(result))
}

async fn stage_dependency_edge(
    State(server): State<Arc<Server>>,
    Path(id): Path<String>,
    AppJson(mut args): AppJson<StageDependencyEdgeArgs>,
) -> Result<(StatusCode, Json<StageDependencyEdgeResult>), ApiError> {
    let path_fid = parse_flow_id(&id)?;
    check_id_match(&path_fid, &args.flow_id, "flow_id")?;
    args.flow_id = path_fid;
    let result = server.stage_dependency_edge(&args).await?;
    Ok((StatusCode::CREATED, Json(result)))
}

async fn apply_dependency_to_child(
    State(server): State<Arc<Server>>,
    Path(id): Path<String>,
    AppJson(mut args): AppJson<ApplyDependencyToChildArgs>,
) -> Result<Json<ApplyDependencyToChildResult>, ApiError> {
    let path_fid = parse_flow_id(&id)?;
    check_id_match(&path_fid, &args.flow_id, "flow_id")?;
    args.flow_id = path_fid;
    Ok(Json(server.apply_dependency_to_child(&args).await?))
}

// ── Budget / Quota handlers ──

async fn create_budget(
    State(server): State<Arc<Server>>,
    AppJson(args): AppJson<CreateBudgetArgs>,
) -> Result<(StatusCode, Json<CreateBudgetResult>), ApiError> {
    let result = server.create_budget(&args).await?;
    let status = match &result {
        CreateBudgetResult::Created { .. } => StatusCode::CREATED,
        CreateBudgetResult::AlreadySatisfied { .. } => StatusCode::OK,
    };
    Ok((status, Json(result)))
}

async fn get_budget_status(
    State(server): State<Arc<Server>>,
    Path(id): Path<String>,
) -> Result<Json<BudgetStatus>, ApiError> {
    let bid = parse_budget_id(&id)?;
    Ok(Json(server.get_budget_status(&bid).await?))
}

#[derive(Deserialize)]
struct ReportUsageBody {
    dimensions: HashMap<String, u64>,
    now: ff_core::types::TimestampMs,
    #[serde(default)]
    dedup_key: Option<String>,
}

async fn report_usage(
    State(server): State<Arc<Server>>,
    Path(id): Path<String>,
    AppJson(body): AppJson<ReportUsageBody>,
) -> Result<Json<ReportUsageResult>, ApiError> {
    let bid = parse_budget_id(&id)?;
    let dims: Vec<String> = body.dimensions.keys().cloned().collect();
    let deltas: Vec<u64> = dims.iter().map(|d| body.dimensions[d]).collect();
    let args = ReportUsageArgs {
        dimensions: dims,
        deltas,
        now: body.now,
        dedup_key: body.dedup_key,
    };
    Ok(Json(server.report_usage(&bid, &args).await?))
}

async fn reset_budget(
    State(server): State<Arc<Server>>,
    Path(id): Path<String>,
) -> Result<Json<ResetBudgetResult>, ApiError> {
    let bid = parse_budget_id(&id)?;
    Ok(Json(server.reset_budget(&bid).await?))
}

async fn create_quota_policy(
    State(server): State<Arc<Server>>,
    AppJson(args): AppJson<CreateQuotaPolicyArgs>,
) -> Result<(StatusCode, Json<CreateQuotaPolicyResult>), ApiError> {
    let result = server.create_quota_policy(&args).await?;
    let status = match &result {
        CreateQuotaPolicyResult::Created { .. } => StatusCode::CREATED,
        CreateQuotaPolicyResult::AlreadySatisfied { .. } => StatusCode::OK,
    };
    Ok((status, Json(result)))
}

// ── Health check ──

#[derive(Serialize)]
struct HealthResponse {
    status: &'static str,
}

async fn healthz(
    State(server): State<Arc<Server>>,
) -> Result<Json<HealthResponse>, ApiError> {
    let _: String = server
        .client()
        .cmd("PING")
        .execute()
        .await
        .map_err(|e| ApiError(crate::server::backend_context(e, "healthz PING")))?;
    Ok(Json(HealthResponse { status: "ok" }))
}

// ── ID parsing helpers ──

/// Return 400 if the body contains an ID that differs from the path ID.
fn check_id_match<T: PartialEq + fmt::Display>(path_id: &T, body_id: &T, id_name: &str) -> Result<(), ApiError> {
    if body_id != path_id {
        return Err(ApiError(ServerError::InvalidInput(format!(
            "path {id_name} does not match body {id_name}"
        ))));
    }
    Ok(())
}

fn parse_execution_id(s: &str) -> Result<ExecutionId, ApiError> {
    ExecutionId::parse(s)
        .map_err(|e| ApiError(ServerError::InvalidInput(format!("invalid execution_id: {e}"))))
}

fn parse_flow_id(s: &str) -> Result<FlowId, ApiError> {
    FlowId::parse(s)
        .map_err(|e| ApiError(ServerError::InvalidInput(format!("invalid flow_id: {e}"))))
}

fn parse_budget_id(s: &str) -> Result<BudgetId, ApiError> {
    BudgetId::parse(s)
        .map_err(|e| ApiError(ServerError::InvalidInput(format!("invalid budget_id: {e}"))))
}

#[cfg(test)]
mod cors_tests {
    //! CORS + auth preflight tests (#66, #71).
    //!
    //! Exercised via `tower::ServiceExt::oneshot` against a minimal router
    //! that applies only the CORS layer to a noop `/healthz`-style route.
    //! This mirrors what browsers actually see (the `CorsLayer` is terminal
    //! for OPTIONS preflights) without needing a live `Server`.
    use super::*;
    use axum::body::Body;
    use axum::http::{Request, StatusCode};
    use axum::routing::get;
    use tower::ServiceExt;

    fn app_with_cors(origins: &[String], auth_enabled: bool) -> Router {
        let cors = build_cors_layer(origins, auth_enabled)
            .expect("build_cors_layer succeeds for valid inputs");
        Router::new().route("/noop", get(|| async { "ok" })).layer(cors)
    }

    #[test]
    fn all_origins_invalid_returns_config_error_instead_of_permissive() {
        // #71: invalid HeaderValue (contains a control character) should
        // fail closed rather than falling back to permissive CORS.
        //
        // Note: `HeaderValue` parsing is intentionally lax — it accepts
        // almost any printable ASCII, so "not a url" parses fine. We use
        // a string with an embedded NUL byte to exercise the actual parse
        // failure path.
        let err = build_cors_layer(&["bad\0origin".to_owned()], false).unwrap_err();
        let ConfigError::InvalidValue { var, message } = err;
        assert_eq!(var, "FF_CORS_ORIGINS");
        assert!(
            message.contains("all configured origins failed to parse"),
            "message was: {message}"
        );
    }

    #[test]
    fn wildcard_still_returns_permissive() {
        // Explicit `*` is still the documented permissive path.
        let layer = build_cors_layer(&["*".to_owned()], false);
        assert!(layer.is_ok());
    }

    #[test]
    fn empty_origins_returns_empty_allowlist_ok() {
        // Empty list (no origins configured) produces an empty allowlist;
        // no browser origin matches. This is NOT the fail-open case — that
        // only happens when origins are configured but all parse-invalid.
        let layer = build_cors_layer(&[], false);
        assert!(layer.is_ok());
    }

    #[test]
    fn mixed_valid_and_invalid_keeps_valid_entries() {
        // A partial typo should not fail the whole boot — we log a warning
        // and keep the parsable entries.
        let layer = build_cors_layer(
            &["https://ok.example.com".to_owned(), "bad\0origin".to_owned()],
            false,
        );
        assert!(layer.is_ok());
    }

    #[tokio::test]
    async fn preflight_allows_authorization_when_auth_enabled() {
        // #66: with token auth on, browsers need `Authorization` in the
        // preflight's `Access-Control-Allow-Headers` response.
        let app = app_with_cors(&["https://client.example.com".to_owned()], true);

        let req = Request::builder()
            .method("OPTIONS")
            .uri("/noop")
            .header("origin", "https://client.example.com")
            .header("access-control-request-method", "GET")
            .header("access-control-request-headers", "authorization,content-type")
            .body(Body::empty())
            .unwrap();

        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);

        let allow_headers = resp
            .headers()
            .get("access-control-allow-headers")
            .expect("access-control-allow-headers present")
            .to_str()
            .unwrap()
            .to_ascii_lowercase();
        assert!(
            allow_headers.contains("authorization"),
            "expected `authorization` in Access-Control-Allow-Headers, got: {allow_headers}"
        );
        assert!(
            allow_headers.contains("content-type"),
            "expected `content-type` in Access-Control-Allow-Headers, got: {allow_headers}"
        );
    }

    #[tokio::test]
    async fn preflight_omits_authorization_when_auth_disabled() {
        // Negative case: without auth, Authorization is not needed and
        // should not be advertised (principle of least privilege).
        let app = app_with_cors(&["https://client.example.com".to_owned()], false);

        let req = Request::builder()
            .method("OPTIONS")
            .uri("/noop")
            .header("origin", "https://client.example.com")
            .header("access-control-request-method", "GET")
            .header("access-control-request-headers", "content-type")
            .body(Body::empty())
            .unwrap();

        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);

        let allow_headers = resp
            .headers()
            .get("access-control-allow-headers")
            .expect("access-control-allow-headers present")
            .to_str()
            .unwrap()
            .to_ascii_lowercase();
        assert!(
            !allow_headers.contains("authorization"),
            "Authorization should not be advertised when auth is off; got: {allow_headers}"
        );
        assert!(allow_headers.contains("content-type"));
    }
}

#[cfg(test)]
mod claim_grant_dto_tests {
    //! Wire-shape tests for `ClaimGrantDto` (issue #91).
    //!
    //! Pins the exact JSON shape the server emits on the
    //! `POST /v1/workers/{id}/claim` 200 response so any drift shows
    //! up at compile-adjacent test time rather than on a live SDK.
    use super::*;
    use ff_core::contracts::ClaimGrant;
    use ff_core::partition::{Partition, PartitionFamily, PartitionKey};
    use ff_core::types::{ExecutionId, FlowId};

    fn sample_grant(family: PartitionFamily) -> ClaimGrant {
        let config = ff_core::partition::PartitionConfig::default();
        let fid = FlowId::new();
        let eid = ExecutionId::for_flow(&fid, &config);
        let p = Partition { family, index: 7 };
        ClaimGrant {
            execution_id: eid,
            partition_key: PartitionKey::from(&p),
            grant_key: "ff:exec:{fp:7}:deadbeef:claim_grant".to_owned(),
            expires_at_ms: 1_700_000_000_000,
        }
    }

    #[test]
    fn claim_grant_dto_emits_opaque_partition_key() {
        let dto = ClaimGrantDto::from(sample_grant(PartitionFamily::Flow));
        let json = serde_json::to_value(&dto).unwrap();
        // `partition_key` serialises transparent: a bare string, not
        // an object, not a `partition_family`/`partition_index` pair.
        assert_eq!(json["partition_key"], serde_json::json!("{fp:7}"));
        assert!(json.get("partition_family").is_none());
        assert!(json.get("partition_index").is_none());
        assert_eq!(json["expires_at_ms"], serde_json::json!(1_700_000_000_000u64));
    }

    #[test]
    fn claim_grant_dto_collapses_execution_alias_on_wire() {
        // RFC-011 §11 alias: Execution-family grants emit the same
        // `{fp:N}` hash tag as Flow. This test pins the wire-level
        // equivalence.
        let dto_flow = ClaimGrantDto::from(sample_grant(PartitionFamily::Flow));
        let dto_exec = ClaimGrantDto::from(sample_grant(PartitionFamily::Execution));
        let jf = serde_json::to_value(&dto_flow).unwrap();
        let je = serde_json::to_value(&dto_exec).unwrap();
        assert_eq!(jf["partition_key"], je["partition_key"]);
    }
}