linkrs_server/http/

router.rs

use axum::{
    http::{HeaderValue, Method, StatusCode},
    middleware,
    routing::{delete, get, post},
    Router,
};
use std::time::Duration;
use tower_http::{
    cors::CorsLayer, limit::RequestBodyLimitLayer, timeout::TimeoutLayer, trace::TraceLayer,
};

use crate::storage::UndoTarget;
use crate::storage::{
    StorageClient, StorageOperationContextOps, StorageSchemaContextOps, StorageSnapshotOps,
    StorageSyncContextOps,
};

#[cfg(feature = "vector")]
use super::handlers::vector;

#[cfg(feature = "fulltext")]
use super::handlers::rebuild as rebuild_handlers;

use super::{
    handlers::{
        auth::{login, logout, me},
        batch::{
            add_items, cancel as cancel_batch, create as create_batch, delete as delete_batch,
            execute as execute_batch, status as batch_status,
        },
        config::{get as get_config, get_key, reset_key, update as update_config, update_key},
        function::{info as function_info, list, register, unregister},
        health, query, schema,
        session::{create as create_session, delete_session, get_session, list_sessions},
        statistics::{
            database, freeze_stats, overview, queries, query_profile_detail,
            search as search_stats, session, system, trigger_freeze,
        },
        stream::execute_stream,
        sync, transaction,
    },
    middleware::{auth::auth_middleware, error, logging},
    state::AppState,
};

/// Creating a router
///
/// Routing structure:
/// /v1/health – Health check (public)
/// – /v1/metrics – Prometheus scrape endpoint (public)
/// – /v1/auth/* – Related to authentication (public information)
/// – /v1/sessions/* – Session management (authentication required)
/// /v1/query – Execution of a query (authentication required)
/// /v1/transactions/* – Transaction management (authentication required)
/// – /v1/schema/* – Schema management (requires authentication)
/// – /api/* – Web management APIs (authentication required)
pub fn create_router<
    S: StorageClient
        + StorageSchemaContextOps
        + StorageSnapshotOps
        + StorageSyncContextOps
        + StorageOperationContextOps
        + crate::storage::AutoCommitBatchOps
        + crate::storage::AutoCommitGroupOps
        + UndoTarget
        + Clone
        + Send
        + Sync
        + 'static,
>(
    state: AppState<S>,
    web_router: Option<Router>,
) -> Router {
    // Public route (no authentication required)
    let public_routes = Router::new()
        .route("/health", get(health::check))
        .route("/metrics", get(super::handlers::metrics::metrics))
        .route("/auth/login", post(login))
        .route(
            // POST /sessions verifies credentials itself, so it lives in the
            // public group — auth is enforced inside the handler, not the
            // middleware. Sessions GET/DELETE stay protected.
            "/sessions",
            post(create_session),
        );

    // Routes that require authentication
    let protected_routes = Router::new()
        .route("/auth/logout", post(logout))
        .route("/auth/me", get(me))
        .route(
            "/users",
            get(super::handlers::users::list).post(super::handlers::users::create),
        )
        .route("/users/{name}", delete(super::handlers::users::drop_user))
        .route(
            "/users/{name}/password",
            post(super::handlers::users::reset_password),
        )
        .route("/users/{name}/enable", post(super::handlers::users::enable))
        .route(
            "/users/{name}/disable",
            post(super::handlers::users::disable),
        )
        .route("/users/{name}/grant", post(super::handlers::users::grant))
        .route("/users/{name}/revoke", post(super::handlers::users::revoke))
        .route("/sessions", post(create_session).get(list_sessions))
        .route("/sessions/{id}", get(get_session).delete(delete_session))
        .route("/query", post(query::execute))
        .route("/query/validate", post(query::validate))
        .route("/query/explain", post(query::explain))
        .route("/query/batch", post(query::execute_batch))
        .route(
            "/transactions",
            post(transaction::begin).get(transaction::list_transactions),
        )
        .route("/transactions/metrics", get(transaction::metrics))
        .route("/transactions/{id}/commit", post(transaction::commit))
        .route("/transactions/{id}/rollback", post(transaction::rollback))
        .route(
            "/transactions/{id}/kill",
            post(transaction::kill_transaction),
        )
        .route(
            "/transactions/{id}/outbox/retry",
            post(transaction::retry_transaction_outbox),
        )
        .route(
            "/transactions/{id}/savepoints",
            post(transaction::create_savepoint).get(transaction::get_savepoints),
        )
        .route(
            "/transactions/{id}/savepoints/{sid}/rollback",
            post(transaction::rollback_to_savepoint),
        )
        .route(
            "/transactions/{id}/savepoints/{sid}",
            delete(transaction::release_savepoint),
        )
        // Batch operation of routes
        .route("/batch", post(create_batch))
        .route("/batch/{id}", get(batch_status).delete(delete_batch))
        .route("/batch/{id}/items", post(add_items))
        .route("/batch/{id}/execute", post(execute_batch))
        .route("/batch/{id}/cancel", post(cancel_batch))
        // Import/Export routes
        .route("/import", post(super::handlers::import::import_file))
        .route("/import/{id}", get(super::handlers::import::import_status))
        .route("/export", get(super::handlers::export::export_data))
        // Statistical information routing
        .route("/statistics/sessions/{id}", get(session))
        .route("/statistics/queries", get(queries))
        .route("/statistics/queries/{trace_id}", get(query_profile_detail))
        .route("/statistics/overview", get(overview))
        .route("/statistics/database", get(database))
        .route("/statistics/system", get(system))
        .route("/statistics/search", get(search_stats))
        .route("/statistics/freeze", get(freeze_stats).post(trigger_freeze))
        .route(
            "/statistics/migration",
            get(super::handlers::statistics::migration),
        )
        // Configure management routing.
        .route("/config", get(get_config).put(update_config))
        .route(
            "/config/{section}/{key}",
            get(get_key).put(update_key).delete(reset_key),
        )
        // Custom function routing
        .route("/functions", post(register).get(list))
        .route("/functions/{name}", get(function_info).delete(unregister))
        // Streaming Query Routing
        .route("/query/stream", post(execute_stream))
        // Forward-only result cursor routing
        .route(
            "/query/cursor/open",
            post(super::handlers::cursor::open_cursor),
        )
        .route(
            "/query/cursor/fetch",
            post(super::handlers::cursor::fetch_cursor),
        )
        .route(
            "/query/cursor/close",
            post(super::handlers::cursor::close_cursor),
        )
        // Sync Management Routes
        .route("/sync/status", get(sync::status))
        .route("/sync/outbox/retry", post(sync::retry_outbox))
        .route("/sync/outbox/diagnostics", get(sync::diagnostics))
        .route("/sync/outbox/dead_letters", get(sync::dead_letters))
        .route("/sync/outbox/requeue", post(sync::requeue))
        .route("/sync/outbox/degraded_ranges", get(sync::degraded_ranges))
        .route("/sync/outbox/degraded/clear", post(sync::degraded_clear))
        .route("/sync/outbox/retention/run", post(sync::retention_run))
        .route("/sync/outbox/retention/status", get(sync::retention_status))
        // Schema Routes
        .route(
            "/schema/spaces",
            post(schema::create_space).get(schema::list_spaces),
        )
        .route(
            "/schema/spaces/{name}",
            get(schema::get_space).delete(schema::drop_space),
        )
        .route(
            "/schema/spaces/{name}/tags",
            post(schema::create_tag).get(schema::list_tags),
        )
        .route(
            "/schema/spaces/{name}/edge-types",
            post(schema::create_edge_type).get(schema::list_edge_types),
        )
        // Schema Versioning Routes
        .route(
            "/schema/versions/{space}/{label}",
            get(schema::get_version_history),
        )
        .route(
            "/schema/changes/{space}/{label}/{from_version}/{to_version}",
            get(schema::get_schema_changes),
        )
        .route(
            "/schema/breaking-changes/{space}/{label}/{from_version}/{to_version}",
            get(schema::detect_breaking_changes),
        )
        // Migration routes
        .route(
            "/migration/plan/{space}/{label}",
            post(schema::create_migration_plan),
        )
        .route("/migration/execute", post(schema::execute_migration))
        .route("/migration/rollback", post(schema::rollback_migration))
        .route("/migration/dry-run", post(schema::dry_run_migration))
        .route(
            "/migration/history/{space}/{label}",
            get(schema::migration_history),
        )
        .route(
            "/migration/status/{space}/{label}",
            get(schema::migration_status),
        )
        .route(
            "/migration/stream/{space}/{label}",
            get(crate::http::handlers::migration_progress::migration_progress_stream),
        )
        .layer(middleware::from_fn_with_state(
            state.clone(),
            auth_middleware,
        ));

    let protected_routes = add_vector_routes(protected_routes);
    let protected_routes = add_fulltext_routes(protected_routes);

    let router = Router::new()
        .nest("/v1", public_routes.merge(protected_routes))
        .layer(middleware::from_fn(logging::logging_middleware))
        .layer(middleware::from_fn(error::error_handling_middleware))
        .layer(TraceLayer::new_for_http())
        .layer(TimeoutLayer::with_status_code(
            StatusCode::REQUEST_TIMEOUT,
            Duration::from_secs(30),
        ))
        .layer(RequestBodyLimitLayer::new(1024 * 1024 * 10)) // Limit on the request body size: 10 MB
        .with_state(state.clone());

    // CORS is driven by the live HTTP config: disabled means no layer at
    // all, an empty whitelist keeps the permissive local behavior with a
    // warning, and a non-empty whitelist enforces strict origins.
    let http_config = state.server.get_config().server.http.clone();
    let router = match create_cors_layer(&http_config) {
        Some(cors) => router.layer(cors),
        None => {
            log::info!("CORS layer disabled by configuration");
            router
        }
    };

    // Add web management routes if web_router is provided
    let mut router = if let Some(wr) = web_router {
        router.nest("/api", wr)
    } else {
        router
    };

    // Debug-only self-hosted OpenAPI document; excluded from the contract.
    if cfg!(debug_assertions) {
        router = router.route("/api-docs/openapi.json", get(serve_openapi));
    }

    router
}

/// Serve the aggregated OpenAPI document in debug builds.
async fn serve_openapi() -> ([(&'static str, &'static str); 1], String) {
    (
        [("Content-Type", "application/json")],
        super::openapi::openapi_json(),
    )
}

/// Conditionally add fulltext rebuild/clear routes
#[cfg(feature = "fulltext")]
fn add_fulltext_routes<
    S: crate::storage::StorageClient
        + crate::storage::StorageSchemaContextOps
        + crate::storage::StorageSyncContextOps
        + crate::storage::StorageOperationContextOps
        + Clone
        + Send
        + Sync
        + 'static,
>(
    router: Router<super::state::AppState<S>>,
) -> Router<super::state::AppState<S>> {
    router
        .route(
            "/fulltext/indexes/rebuild",
            post(rebuild_handlers::rebuild_fulltext),
        )
        .route(
            "/fulltext/rebuilds/{id}",
            get(rebuild_handlers::fulltext_rebuild_status),
        )
        .route(
            "/fulltext/indexes/clear",
            post(rebuild_handlers::clear_fulltext),
        )
        .route(
            "/fulltext/indexes/inconsistent",
            get(rebuild_handlers::inconsistent_fulltext),
        )
}

#[cfg(not(feature = "fulltext"))]
fn add_fulltext_routes<
    S: crate::storage::StorageClient
        + crate::storage::StorageSchemaContextOps
        + crate::storage::StorageSyncContextOps
        + crate::storage::StorageOperationContextOps
        + Clone
        + Send
        + Sync
        + 'static,
>(
    router: Router<super::state::AppState<S>>,
) -> Router<super::state::AppState<S>> {
    router
}

/// Conditionally add vector search routes
#[cfg(feature = "vector")]
fn add_vector_routes<
    S: crate::storage::StorageClient
        + crate::storage::StorageSchemaContextOps
        + crate::storage::StorageSyncContextOps
        + crate::storage::StorageOperationContextOps
        + Clone
        + Send
        + Sync
        + 'static,
>(
    router: Router<super::state::AppState<S>>,
) -> Router<super::state::AppState<S>> {
    router
        .route(
            "/vector/indexes",
            post(vector::create_index).get(vector::list_indexes),
        )
        .route(
            "/vector/indexes/{space_id}/{tag_name}/{field_name}",
            get(vector::get_index_info).delete(vector::drop_index),
        )
        .route("/vector/search", post(vector::search))
        .route(
            "/vector/{space_id}/{tag_name}/{field_name}/{point_id}",
            get(vector::get_vector),
        )
        .route(
            "/vector/{space_id}/{tag_name}/{field_name}/count",
            get(vector::count),
        )
        .route("/vector/payload", axum::routing::put(vector::set_payload))
        .route(
            "/vector/payload/fields",
            axum::routing::put(vector::set_payload_fields),
        )
        .route("/vector/payload/delete", post(vector::delete_payload))
        .route("/vector/scroll", post(vector::scroll))
        .route(
            "/vector/indexes/rebuild",
            post(super::handlers::rebuild::rebuild_vector),
        )
        .route(
            "/vector/rebuilds/{id}",
            get(super::handlers::rebuild::vector_rebuild_status),
        )
        .route(
            "/vector/indexes/clear",
            post(super::handlers::rebuild::clear_vector),
        )
}

#[cfg(not(feature = "vector"))]
fn add_vector_routes<
    S: crate::storage::StorageClient
        + crate::storage::StorageSchemaContextOps
        + crate::storage::StorageSyncContextOps
        + crate::storage::StorageOperationContextOps
        + Clone
        + Send
        + Sync
        + 'static,
>(
    router: Router<super::state::AppState<S>>,
) -> Router<super::state::AppState<S>> {
    router
}

/// Create a CORS configuration layer from the live HTTP config.
///
/// Returns `None` when CORS is disabled. An enabled switch with an empty
/// whitelist keeps the permissive local-development combination and logs a
/// warning; production must set exact origins to get the strict layer.
/// The session id travels in the `X-Session-ID` header rather than a
/// cookie, so credential mode stays off and never conflicts with origins.
fn create_cors_layer(config: &crate::config::HttpServerConfig) -> Option<CorsLayer> {
    use tower_http::cors::{AllowOrigin, Any};

    if !config.cors_enabled {
        return None;
    }
    if config.cors_allowed_origins.is_empty() {
        log::warn!(
            "CORS running in permissive mode (any origin); set server.http.cors_allowed_origins for production"
        );
        return Some(
            CorsLayer::new()
                .allow_origin(Any)
                .allow_methods(Any)
                .allow_headers(Any),
        );
    }
    let origins: Vec<HeaderValue> = config
        .cors_allowed_origins
        .iter()
        .filter_map(|origin| match origin.parse() {
            Ok(value) => Some(value),
            Err(e) => {
                log::warn!("Skipping invalid CORS origin '{}': {}", origin, e);
                None
            }
        })
        .collect();
    log::info!(
        "CORS running in strict mode for {} origin(s)",
        origins.len()
    );
    Some(
        CorsLayer::new()
            .allow_origin(AllowOrigin::list(origins))
            .allow_methods([
                Method::GET,
                Method::POST,
                Method::PUT,
                Method::DELETE,
                Method::OPTIONS,
            ])
            .allow_headers([
                axum::http::header::CONTENT_TYPE,
                axum::http::header::AUTHORIZATION,
                axum::http::header::HeaderName::from_static("x-session-id"),
            ])
            .max_age(Duration::from_secs(600)),
    )
}