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,
};
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 {
let public_routes = Router::new()
.route("/health", get(health::check))
.route("/metrics", get(super::handlers::metrics::metrics))
.route("/auth/login", post(login))
.route(
"/sessions",
post(create_session),
);
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),
)
.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))
.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))
.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),
)
.route("/config", get(get_config).put(update_config))
.route(
"/config/{section}/{key}",
get(get_key).put(update_key).delete(reset_key),
)
.route("/functions", post(register).get(list))
.route("/functions/{name}", get(function_info).delete(unregister))
.route("/query/stream", post(execute_stream))
.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),
)
.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))
.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),
)
.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),
)
.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)) .with_state(state.clone());
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
}
};
let mut router = if let Some(wr) = web_router {
router.nest("/api", wr)
} else {
router
};
if cfg!(debug_assertions) {
router = router.route("/api-docs/openapi.json", get(serve_openapi));
}
router
}
async fn serve_openapi() -> ([(&'static str, &'static str); 1], String) {
(
[("Content-Type", "application/json")],
super::openapi::openapi_json(),
)
}
#[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
}
#[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
}
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)),
)
}