fraiseql-server 2.3.0

//! Axum router integration for the REST transport.
//!
//! [`rest_router`] builds an axum [`Router`] from a [`RestRouteTable`] and
//! mounts it with middleware (compression, `X-Request-Id`).  Auth, rate
//! limiting, and CORS are applied at the server level and inherited.

pub mod helpers;

#[cfg(test)]
mod tests;

use std::sync::Arc;

use axum::{
    Router,
    body::Body,
    extract::{Request, State},
    http::StatusCode,
    response::Response,
    routing::{delete, get, patch, post, put},
};
use fraiseql_core::{
    db::traits::{DatabaseAdapter, SupportsMutations},
    runtime::Executor,
};
use helpers::{
    error_response, parse_query_pairs, rest_result_to_response, strip_base_path, to_axum_path,
};
use serde_json::json;
use tower_http::compression::{CompressionLayer, predicate::SizeAbove};
use tracing::info;

use super::{
    handler::{RestError, RestHandler, RestResponse},
    resource::{HttpMethod, RestRouteTable, RouteSource},
};
use crate::{extractors::OptionalSecurityContext, routes::graphql::AppState};

// ---------------------------------------------------------------------------
// Public API
// ---------------------------------------------------------------------------

// ---------------------------------------------------------------------------

/// Derive REST configuration, route table, and shared state from the compiled schema.
///
/// Returns `None` if `rest_config` is absent or `enabled` is `false`, or if
/// route derivation fails. This shared helper is used by both [`rest_query_router`]
/// and [`rest_mutation_router`].
///
/// # Errors
///
/// Returns `None` (with a warning log) if the route table cannot be derived.
fn derive_rest_context<A>(
    state: &AppState<A>,
) -> Option<(String, Arc<RestRouteTable>, RestState<A>)>
where
    A: DatabaseAdapter + Clone + Send + Sync + 'static,
{
    let executor = state.executor();
    let schema = executor.schema();

    let config = match &schema.rest_config {
        Some(cfg) if cfg.enabled => cfg.clone(),
        Some(_) => {
            info!("REST transport disabled (rest.enabled = false)");
            return None;
        },
        None => {
            return None;
        },
    };

    let route_table = match RestRouteTable::from_compiled_schema(schema) {
        Ok(rt) => Arc::new(rt),
        Err(e) => {
            tracing::warn!(error = %e, "REST route derivation failed — REST transport disabled");
            return None;
        },
    };

    // Log diagnostics from derivation.
    for diag in &route_table.diagnostics {
        match diag.level {
            super::resource::DiagnosticLevel::Info => {
                tracing::debug!(message = %diag.message, "REST derivation");
            },
            super::resource::DiagnosticLevel::Warning => {
                tracing::warn!(message = %diag.message, "REST derivation");
            },
            super::resource::DiagnosticLevel::Error => {
                tracing::error!(message = %diag.message, "REST derivation");
            },
        }
    }

    let base_path = config.path;
    let idempotency_store = super::idempotency::create_store(config.idempotency_ttl_seconds);

    let rest_state = RestState {
        executor: state.executor.load_full(),
        route_table: route_table.clone(),
        idempotency_store,
        #[cfg(feature = "observers")]
        event_transport: None,
    };

    Some((base_path, route_table, rest_state))
}

/// Build an axum [`Router`] for read-only REST endpoints (GET queries and SSE
/// streams) derived from the compiled schema.
///
/// Returns `None` if `rest_config` is absent or `enabled` is `false`, or if
/// route derivation fails.
///
/// Does **not** require `SupportsMutations` — suitable for read-only adapters such
/// as `FraiseWireAdapter` and `SqliteAdapter`.
///
/// The returned router is *not* nested — the caller must merge it into the
/// application router.  Middleware applied at the server level (auth, rate
/// limiting, CORS, tracing, body-size limit) is inherited automatically.
///
/// # Errors
///
/// Returns `None` (with a warning log) if the route table cannot be derived.
pub fn rest_query_router<A>(state: &AppState<A>, compression_enabled: bool) -> Option<Router>
where
    A: DatabaseAdapter + Clone + Send + Sync + 'static,
{
    let (base_path, route_table, rest_state) = derive_rest_context(state)?;
    let executor = state.executor();
    let schema = executor.schema();

    let mut router = Router::new();

    for resource in &route_table.resources {
        // Register GET routes for queries.
        for route in &resource.routes {
            if route.method == HttpMethod::Get {
                let axum_path = to_axum_path(&base_path, &route.path);
                router = router.route(&axum_path, get(rest_get_handler::<A>));
            }
        }

        // Register SSE stream route: GET /{resource}/stream
        let stream_path = to_axum_path(&base_path, &format!("/{}/stream", resource.name));
        router = router.route(&stream_path, get(rest_sse_handler::<A>));
    }

    // Finalize state; apply framework-level compression if enabled.
    let mut router = router.with_state(rest_state);
    if compression_enabled {
        router = router.layer(CompressionLayer::new().compress_when(SizeAbove::new(1024)));
    }

    // Serve OpenAPI specification at {base_path}/openapi.json.
    let openapi_path = format!("{}/openapi.json", base_path.trim_end_matches('/'));
    let openapi_spec = match super::openapi::generate_openapi(schema, &route_table) {
        Ok(spec) => Arc::new(spec),
        Err(e) => {
            tracing::warn!(error = %e, "OpenAPI spec generation failed");
            Arc::new(json!({"error": "OpenAPI generation failed"}))
        },
    };
    router = router.route(
        &openapi_path,
        get(move || {
            let spec = openapi_spec.clone();
            async move { axum::Json((*spec).clone()) }
        }),
    );

    // Log startup summary.
    let resource_count = route_table.resources.len();
    let get_route_count: usize = route_table
        .resources
        .iter()
        .flat_map(|r| &r.routes)
        .filter(|r| r.method == HttpMethod::Get)
        .count();
    let paths: Vec<String> = route_table
        .resources
        .iter()
        .map(|r| format!("{}/{}", base_path, r.name))
        .collect();
    info!(
        resources = resource_count,
        routes = get_route_count,
        base_path = %base_path,
        paths = ?paths,
        "REST query transport enabled (read-only)"
    );

    Some(router)
}

/// Build an axum [`Router`] for all REST endpoints — both read-only (GET, SSE)
/// and mutation (POST, PUT, PATCH, DELETE) routes — derived from the compiled
/// schema.
///
/// Returns `None` if `rest_config` is absent or `enabled` is `false`, or if
/// route derivation fails.
///
/// Requires `SupportsMutations` because mutation handlers call
/// `Executor::execute_mutation()` which has the same compile-time bound.
///
/// The returned router is *not* nested — the caller must merge it into the
/// application router.  Middleware applied at the server level (auth, rate
/// limiting, CORS, tracing, body-size limit) is inherited automatically.
///
/// # Errors
///
/// Returns `None` (with a warning log) if the route table cannot be derived.
pub fn rest_router<A>(state: &AppState<A>, compression_enabled: bool) -> Option<Router>
where
    A: DatabaseAdapter + SupportsMutations + Clone + Send + Sync + 'static,
{
    let (base_path, route_table, rest_state) = derive_rest_context(state)?;
    let executor = state.executor();
    let schema = executor.schema();

    let mut router = Router::new();

    // Track which collection paths already have PATCH/DELETE so we can add
    // bulk operation routes for resources that have update/delete mutations.
    let mut collection_patch_paths = std::collections::HashSet::new();
    let mut collection_delete_paths = std::collections::HashSet::new();

    for resource in &route_table.resources {
        for route in &resource.routes {
            let axum_path = to_axum_path(&base_path, &route.path);
            router = match route.method {
                HttpMethod::Get => router.route(&axum_path, get(rest_get_handler::<A>)),
                HttpMethod::Post => router.route(&axum_path, post(rest_post_handler::<A>)),
                HttpMethod::Put => router.route(&axum_path, put(rest_put_handler::<A>)),
                HttpMethod::Patch => {
                    let collection_path = to_axum_path(&base_path, &format!("/{}", resource.name));
                    collection_patch_paths.insert(collection_path);
                    router.route(&axum_path, patch(rest_patch_handler::<A>))
                },
                HttpMethod::Delete => {
                    let collection_path = to_axum_path(&base_path, &format!("/{}", resource.name));
                    collection_delete_paths.insert(collection_path);
                    router.route(&axum_path, delete(rest_delete_handler::<A>))
                },
            };
        }

        // Register collection-level PATCH route for bulk update if an update
        // mutation exists but no collection PATCH was derived.
        let collection_path = to_axum_path(&base_path, &format!("/{}", resource.name));
        let has_update = resource.routes.iter().any(|r| {
            matches!(&r.source, RouteSource::Mutation { name }
                if state.executor().schema().find_mutation(name)
                    .is_some_and(|m| matches!(m.operation,
                        fraiseql_core::schema::MutationOperation::Update { .. })))
        });
        if has_update && !collection_patch_paths.contains(&collection_path) {
            router = router.route(&collection_path, patch(rest_patch_handler::<A>));
        }

        // Register collection-level DELETE route for bulk delete.
        let has_delete = resource.routes.iter().any(|r| {
            matches!(&r.source, RouteSource::Mutation { name }
                if state.executor().schema().find_mutation(name)
                    .is_some_and(|m| matches!(m.operation,
                        fraiseql_core::schema::MutationOperation::Delete { .. })))
        });
        if has_delete && !collection_delete_paths.contains(&collection_path) {
            router = router.route(&collection_path, delete(rest_delete_handler::<A>));
        }

        // Register SSE stream route: GET /{resource}/stream
        let stream_path = to_axum_path(&base_path, &format!("/{}/stream", resource.name));
        router = router.route(&stream_path, get(rest_sse_handler::<A>));
    }

    // Finalize state; apply framework-level compression if enabled.
    let mut router = router.with_state(rest_state);
    if compression_enabled {
        router = router.layer(CompressionLayer::new().compress_when(SizeAbove::new(1024)));
    }

    // Serve OpenAPI specification at {base_path}/openapi.json.
    let openapi_path = format!("{}/openapi.json", base_path.trim_end_matches('/'));
    let openapi_spec = match super::openapi::generate_openapi(schema, &route_table) {
        Ok(spec) => Arc::new(spec),
        Err(e) => {
            tracing::warn!(error = %e, "OpenAPI spec generation failed");
            Arc::new(json!({"error": "OpenAPI generation failed"}))
        },
    };
    router = router.route(
        &openapi_path,
        get(move || {
            let spec = openapi_spec.clone();
            async move { axum::Json((*spec).clone()) }
        }),
    );

    // Log startup summary.
    let resource_count = route_table.resources.len();
    let route_count: usize = route_table.resources.iter().map(|r| r.routes.len()).sum();
    let paths: Vec<String> = route_table
        .resources
        .iter()
        .map(|r| format!("{}/{}", base_path, r.name))
        .collect();
    info!(
        resources = resource_count,
        routes = route_count,
        base_path = %base_path,
        paths = ?paths,
        "REST transport enabled"
    );

    Some(router)
}

// ---------------------------------------------------------------------------
// State
// ---------------------------------------------------------------------------

/// Shared state for REST handlers.
#[derive(Clone)]
struct RestState<A: DatabaseAdapter> {
    executor:          Arc<Executor<A>>,
    route_table:       Arc<RestRouteTable>,
    idempotency_store: Arc<dyn super::idempotency::IdempotencyStore>,
    /// Optional event transport for SSE streaming (requires `observers` feature).
    #[cfg(feature = "observers")]
    event_transport:   Option<Arc<dyn fraiseql_observers::transport::EventTransport>>,
}

// ---------------------------------------------------------------------------
// Axum handlers
// ---------------------------------------------------------------------------

/// GET handler — query execution (single resource or collection).
///
/// Content negotiation:
/// - `Accept: application/x-ndjson` → NDJSON streaming (one JSON object per line)
/// - `Accept: application/json` (default) → standard envelope response
async fn rest_get_handler<A>(
    State(rest): State<RestState<A>>,
    OptionalSecurityContext(security_ctx): OptionalSecurityContext,
    request: Request<Body>,
) -> Response
where
    A: DatabaseAdapter + Clone + Send + Sync + 'static,
{
    let (parts, _body) = request.into_parts();
    let relative_path = strip_base_path(&rest.route_table.base_path, parts.uri.path());
    let query_string = parts.uri.query().unwrap_or("");
    let query_pairs = parse_query_pairs(query_string);
    let query_refs: Vec<(&str, &str)> =
        query_pairs.iter().map(|(k, v)| (k.as_str(), v.as_str())).collect();

    // NDJSON content negotiation
    if super::streaming::accepts_ndjson(&parts.headers) {
        let schema = rest.executor.schema();
        let config = schema.rest_config.as_ref().expect("REST config must exist: handler is only reached via a matched REST route, which requires rest_config to be present in the schema");
        let handler = RestHandler::new(&rest.executor, schema, config, &rest.route_table);
        let result = super::streaming::handle_ndjson_get(
            &handler,
            &relative_path,
            &query_refs,
            &parts.headers,
            security_ctx.as_ref(),
        )
        .await;

        return match result {
            Ok(ndjson) => {
                let mut builder = Response::builder().status(StatusCode::OK);
                for (key, value) in &ndjson.headers {
                    builder = builder.header(key, value);
                }
                builder.body(ndjson.body.into_body()).unwrap_or_else(|_| {
                    Response::builder()
                        .status(StatusCode::INTERNAL_SERVER_ERROR)
                        .body(Body::empty())
                        .expect("fallback response: Response::builder() with INTERNAL_SERVER_ERROR status and empty body is infallible")
                })
            },
            Err(rest_err) => rest_result_to_response(Err(rest_err)),
        };
    }

    let schema = rest.executor.schema();
    let config = schema.rest_config.as_ref().expect("REST config must exist: handler is only reached via a matched REST route, which requires rest_config to be present in the schema");
    let handler = RestHandler::new(&rest.executor, schema, config, &rest.route_table);

    let result = handler
        .handle_get(&relative_path, &query_refs, &parts.headers, security_ctx.as_ref())
        .await;

    rest_result_to_response(result)
}

/// POST handler — create mutation or custom action.
async fn rest_post_handler<A>(
    State(rest): State<RestState<A>>,
    OptionalSecurityContext(security_ctx): OptionalSecurityContext,
    request: Request<Body>,
) -> Response
where
    A: DatabaseAdapter + SupportsMutations + Clone + Send + Sync + 'static,
{
    let (parts, body) = request.into_parts();
    let relative_path = strip_base_path(&rest.route_table.base_path, parts.uri.path());

    let body_value = match read_json_body(body).await {
        Ok(v) => v,
        Err(resp) => return resp,
    };

    let schema = rest.executor.schema();
    let config = schema.rest_config.as_ref().expect("REST config must exist: handler is only reached via a matched REST route, which requires rest_config to be present in the schema");
    let handler = RestHandler::new(&rest.executor, schema, config, &rest.route_table)
        .with_idempotency_store(&rest.idempotency_store);

    let result = handler
        .handle_post(&relative_path, &body_value, &parts.headers, security_ctx.as_ref())
        .await;

    rest_result_to_response(result)
}

/// PUT handler — full update mutation.
async fn rest_put_handler<A>(
    State(rest): State<RestState<A>>,
    OptionalSecurityContext(security_ctx): OptionalSecurityContext,
    request: Request<Body>,
) -> Response
where
    A: DatabaseAdapter + SupportsMutations + Clone + Send + Sync + 'static,
{
    let (parts, body) = request.into_parts();
    let relative_path = strip_base_path(&rest.route_table.base_path, parts.uri.path());

    let body_value = match read_json_body(body).await {
        Ok(v) => v,
        Err(resp) => return resp,
    };

    let schema = rest.executor.schema();
    let config = schema.rest_config.as_ref().expect("REST config must exist: handler is only reached via a matched REST route, which requires rest_config to be present in the schema");
    let handler = RestHandler::new(&rest.executor, schema, config, &rest.route_table);

    let result = handler
        .handle_put(&relative_path, &body_value, &parts.headers, security_ctx.as_ref())
        .await;

    rest_result_to_response(result)
}

/// PATCH handler — partial update mutation or bulk update.
async fn rest_patch_handler<A>(
    State(rest): State<RestState<A>>,
    OptionalSecurityContext(security_ctx): OptionalSecurityContext,
    request: Request<Body>,
) -> Response
where
    A: DatabaseAdapter + SupportsMutations + Clone + Send + Sync + 'static,
{
    let (parts, body) = request.into_parts();
    let relative_path = strip_base_path(&rest.route_table.base_path, parts.uri.path());
    let query_string = parts.uri.query().unwrap_or("");
    let query_pairs = parse_query_pairs(query_string);
    let query_refs: Vec<(&str, &str)> =
        query_pairs.iter().map(|(k, v)| (k.as_str(), v.as_str())).collect();

    let body_value = match read_json_body(body).await {
        Ok(v) => v,
        Err(resp) => return resp,
    };

    let schema = rest.executor.schema();
    let config = schema.rest_config.as_ref().expect("REST config must exist: handler is only reached via a matched REST route, which requires rest_config to be present in the schema");
    let handler = RestHandler::new(&rest.executor, schema, config, &rest.route_table);

    let result = handler
        .handle_patch(
            &relative_path,
            &body_value,
            &query_refs,
            &parts.headers,
            security_ctx.as_ref(),
        )
        .await;

    rest_result_to_response(result)
}

/// DELETE handler — single-resource delete or bulk delete.
async fn rest_delete_handler<A>(
    State(rest): State<RestState<A>>,
    OptionalSecurityContext(security_ctx): OptionalSecurityContext,
    request: Request<Body>,
) -> Response
where
    A: DatabaseAdapter + SupportsMutations + Clone + Send + Sync + 'static,
{
    let (parts, _body) = request.into_parts();
    let relative_path = strip_base_path(&rest.route_table.base_path, parts.uri.path());
    let query_string = parts.uri.query().unwrap_or("");
    let query_pairs = parse_query_pairs(query_string);
    let query_refs: Vec<(&str, &str)> =
        query_pairs.iter().map(|(k, v)| (k.as_str(), v.as_str())).collect();

    let schema = rest.executor.schema();
    let config = schema.rest_config.as_ref().expect("REST config must exist: handler is only reached via a matched REST route, which requires rest_config to be present in the schema");
    let handler = RestHandler::new(&rest.executor, schema, config, &rest.route_table);

    let result = handler
        .handle_delete(&relative_path, &query_refs, &parts.headers, security_ctx.as_ref())
        .await;

    rest_result_to_response(result)
}

/// SSE handler — stream entity change events in real-time.
///
/// Returns `501 Not Implemented` when the `observers` feature is disabled.
/// Otherwise, streams events for the given resource type via SSE.
async fn rest_sse_handler<A>(
    State(rest): State<RestState<A>>,
    OptionalSecurityContext(security_ctx): OptionalSecurityContext,
    request: Request<Body>,
) -> Response
where
    A: DatabaseAdapter + Clone + Send + Sync + 'static,
{
    let (parts, _body) = request.into_parts();
    let relative_path = strip_base_path(&rest.route_table.base_path, parts.uri.path());

    // Extract resource name from /{resource}/stream path
    let resource_name = match super::sse::extract_stream_resource(&relative_path) {
        Some(name) => name.to_string(),
        None => {
            return rest_result_to_response(Err(super::handler::RestError::not_found(
                "Stream endpoint not found",
            )));
        },
    };

    // Verify the resource exists
    let schema = rest.executor.schema();
    let has_resource = rest.route_table.resources.iter().any(|r| r.name == resource_name);

    if !has_resource {
        return rest_result_to_response(Err(super::handler::RestError::not_found(format!(
            "Resource not found: {resource_name}"
        ))));
    }

    // Check auth if required
    if let Some(config) = &schema.rest_config {
        if config.require_auth && security_ctx.is_none() {
            return rest_result_to_response(Err(super::handler::RestError {
                status:  StatusCode::UNAUTHORIZED,
                code:    "UNAUTHENTICATED",
                message: "Authentication required".to_string(),
                details: None,
            }));
        }
    }

    // Read heartbeat interval from REST config (or use default).
    let heartbeat_secs = schema
        .rest_config
        .as_ref()
        .map_or(super::sse::DEFAULT_SSE_HEARTBEAT_SECONDS, |c| c.sse_heartbeat_seconds);

    // Check if observers feature is available
    #[cfg(not(feature = "observers"))]
    {
        let _ = heartbeat_secs; // suppress unused warning
        rest_result_to_response(Err(super::sse::observers_not_available()))
    }

    // With observers feature: set up SSE stream with real event subscription.
    #[cfg(feature = "observers")]
    {
        let _last_event_id = super::sse::extract_last_event_id(&parts.headers);
        let heartbeat_interval = std::time::Duration::from_secs(heartbeat_secs);

        // If we have an event transport, subscribe to real entity events.
        if let Some(ref transport) = rest.event_transport {
            let filter = fraiseql_observers::transport::EventFilter {
                entity_type: Some(resource_name.clone()),
                ..Default::default()
            };

            match transport.subscribe(filter).await {
                Ok(event_stream) => {
                    use futures::StreamExt;

                    // Merge entity events with heartbeat ticks.
                    let heartbeat = futures::stream::unfold((), move |()| async move {
                        tokio::time::sleep(heartbeat_interval).await;
                        let event = axum::response::sse::Event::default().event("ping").data("");
                        Some((event, ()))
                    });

                    let entity_events = event_stream.filter_map(|result| async move {
                        match result {
                            Ok(entity_event) => {
                                let event_type = super::sse::event_kind_to_sse_type(
                                    entity_event.event_type.as_str(),
                                );
                                let event = axum::response::sse::Event::default()
                                    .event(event_type)
                                    .id(entity_event.id.to_string())
                                    .json_data(&entity_event.data)
                                    .ok()?;
                                Some(event)
                            },
                            Err(e) => {
                                tracing::warn!(error = %e, "SSE event stream error");
                                None
                            },
                        }
                    });

                    // Select between entity events and heartbeat pings.
                    let merged = futures::stream::select(entity_events, heartbeat)
                        .map(Ok::<_, std::convert::Infallible>);

                    let sse = axum::response::sse::Sse::new(merged).keep_alive(
                        axum::response::sse::KeepAlive::new().interval(heartbeat_interval).text(""),
                    );

                    return axum::response::IntoResponse::into_response(sse);
                },
                Err(e) => {
                    tracing::warn!(error = %e, resource = %resource_name, "Failed to subscribe to event stream");
                    return rest_result_to_response(Err(super::handler::RestError {
                        status:  StatusCode::SERVICE_UNAVAILABLE,
                        code:    "EVENT_STREAM_UNAVAILABLE",
                        message: "Could not connect to event stream".to_string(),
                        details: None,
                    }));
                },
            }
        }

        // Fallback: no event transport configured — heartbeat-only stream.
        let stream = futures::stream::unfold((), move |()| async move {
            tokio::time::sleep(heartbeat_interval).await;
            let event = axum::response::sse::Event::default().event("ping").data("");
            Some((Ok::<_, std::convert::Infallible>(event), ()))
        });

        let sse = axum::response::sse::Sse::new(stream).keep_alive(
            axum::response::sse::KeepAlive::new().interval(heartbeat_interval).text(""),
        );

        axum::response::IntoResponse::into_response(sse)
    }
}

// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------

/// Read and parse a JSON request body.
async fn read_json_body(body: Body) -> Result<serde_json::Value, Response> {
    let Ok(bytes) = axum::body::to_bytes(body, 1_048_576).await else {
        return Err(error_response(
            StatusCode::PAYLOAD_TOO_LARGE,
            "PAYLOAD_TOO_LARGE",
            "Request body too large",
        ));
    };

    if bytes.is_empty() {
        return Ok(serde_json::Value::Object(serde_json::Map::new()));
    }

    serde_json::from_slice(&bytes).map_err(|e| {
        error_response(StatusCode::BAD_REQUEST, "INVALID_JSON", &format!("Invalid JSON body: {e}"))
    })
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------