stuart/databuild
1
1
Fork 0
Code Issues 1 Pull requests Projects Releases Packages Wiki Activity Actions

implemented api phase 6

Browse source
This commit is contained in:
Stuart Axelbrooke 2025-11-22 22:02:49 +08:00
parent f14d93da7a
commit 556ccb8a4b
2 changed files with 118 additions and 49 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
databuild/BUILD.bazel
View file

@ -36,6 +36,7 @@ rust_library(
"@crates//:tokio",
"@crates//:tower",
"@crates//:tower-http",
"@crates//:tracing",
"@crates//:uuid",
],
)

166
databuild/http_server.rs
View file

@ -9,7 +9,7 @@ use crate::{
use axum::{
Json, Router,
extract::{Path, Query, Request, State},
http::StatusCode,
http::{HeaderValue, Method, StatusCode},
middleware::{self, Next},
response::{IntoResponse, Response},
routing::{delete, get, post},
@ -20,6 +20,7 @@ use std::sync::{
};
use std::time::{SystemTime, UNIX_EPOCH};
use tokio::sync::{broadcast, mpsc, oneshot};
use tower_http::cors::CorsLayer;
/// Shared application state for HTTP handlers
#[derive(Clone)]
@ -73,6 +74,15 @@ async fn update_last_request_time(
/// Create the Axum router with all endpoints
pub fn create_router(state: AppState) -> Router {
// Configure CORS for web app development
let cors = CorsLayer::new()
.allow_origin("http://localhost:3000".parse::<HeaderValue>().unwrap())
.allow_methods([Method::GET, Method::POST, Method::DELETE, Method::OPTIONS])
.allow_headers([
axum::http::header::CONTENT_TYPE,
axum::http::header::AUTHORIZATION,
]);
Router::new()
// Health check
.route("/health", get(health))
@ -85,6 +95,8 @@ pub fn create_router(state: AppState) -> Router {
.route("/api/partitions", get(list_partitions))
// Job run endpoints
.route("/api/job_runs", get(list_job_runs))
// Add CORS middleware
.layer(cors)
// Add middleware to track request time
.layer(middleware::from_fn_with_state(
state.clone(),
@ -93,6 +105,34 @@ pub fn create_router(state: AppState) -> Router {
.with_state(state)
}
// ============================================================================
// Error Handling
// ============================================================================
/// Standard error response structure
#[derive(serde::Serialize)]
struct ErrorResponse {
error: String,
#[serde(skip_serializing_if = "Option::is_none")]
details: Option<serde_json::Value>,
}
impl ErrorResponse {
fn new(error: impl Into<String>) -> Self {
Self {
error: error.into(),
details: None,
}
}
fn with_details(error: impl Into<String>, details: serde_json::Value) -> Self {
Self {
error: error.into(),
details: Some(details),
}
}
}
// ============================================================================
// Handlers
// ============================================================================
@ -111,11 +151,10 @@ async fn list_wants(
let events = match state.bel_storage.list_events(0, 100000) {
Ok(events) => events,
Err(e) => {
tracing::error!("Failed to read events from BEL storage: {}", e);
return (
StatusCode::INTERNAL_SERVER_ERROR,
Json(serde_json::json!({
"error": format!("Failed to read events: {}", e)
})),
Json(ErrorResponse::new(format!("Failed to read events: {}", e))),
)
.into_response();
}
@ -135,11 +174,10 @@ async fn get_want(State(state): State<AppState>, Path(want_id): Path<String>) ->
let events = match state.bel_storage.list_events(0, 100000) {
Ok(events) => events,
Err(e) => {
tracing::error!("Failed to read events from BEL storage: {}", e);
return (
StatusCode::INTERNAL_SERVER_ERROR,
Json(serde_json::json!({
"error": format!("Failed to read events: {}", e)
})),
Json(ErrorResponse::new(format!("Failed to read events: {}", e))),
)
.into_response();
}
@ -147,16 +185,24 @@ async fn get_want(State(state): State<AppState>, Path(want_id): Path<String>) ->
let build_state = BuildState::from_events(&events);
let req = GetWantRequest { want_id };
let req = GetWantRequest {
want_id: want_id.clone(),
};
let response = build_state.get_want(&req.want_id);
match response {
Some(want) => (StatusCode::OK, Json(GetWantResponse { data: Some(want) })).into_response(),
None => (
StatusCode::NOT_FOUND,
Json(serde_json::json!({"error": "Want not found"})),
)
.into_response(),
None => {
tracing::debug!("Want not found: {}", want_id);
(
StatusCode::NOT_FOUND,
Json(ErrorResponse::with_details(
"Want not found",
serde_json::json!({"want_id": want_id}),
)),
)
.into_response()
}
}
}
@ -186,21 +232,33 @@ async fn create_want(
// Wait for orchestrator reply
match reply_rx.await {
Ok(Ok(response)) => (StatusCode::OK, Json(response)).into_response(),
Ok(Err(e)) => (
StatusCode::INTERNAL_SERVER_ERROR,
Json(serde_json::json!({
"error": format!("Failed to create want: {}", e)
})),
)
.into_response(),
Err(_) => (
StatusCode::INTERNAL_SERVER_ERROR,
Json(serde_json::json!({
"error": "Orchestrator did not respond"
})),
)
.into_response(),
Ok(Ok(response)) => {
tracing::info!(
"Created want: {}",
response
.data
.as_ref()
.map(|w| &w.want_id)
.unwrap_or(&"unknown".to_string())
);
(StatusCode::OK, Json(response)).into_response()
}
Ok(Err(e)) => {
tracing::error!("Failed to create want: {}", e);
(
StatusCode::INTERNAL_SERVER_ERROR,
Json(ErrorResponse::new(format!("Failed to create want: {}", e))),
)
.into_response()
}
Err(_) => {
tracing::error!("Orchestrator did not respond to create want command");
(
StatusCode::INTERNAL_SERVER_ERROR,
Json(ErrorResponse::new("Orchestrator did not respond")),
)
.into_response()
}
}
}
@ -234,21 +292,33 @@ async fn cancel_want(
// Wait for orchestrator reply
match reply_rx.await {
Ok(Ok(response)) => (StatusCode::OK, Json(response)).into_response(),
Ok(Err(e)) => (
StatusCode::INTERNAL_SERVER_ERROR,
Json(serde_json::json!({
"error": format!("Failed to cancel want: {}", e)
})),
)
.into_response(),
Err(_) => (
StatusCode::INTERNAL_SERVER_ERROR,
Json(serde_json::json!({
"error": "Orchestrator did not respond"
})),
)
.into_response(),
Ok(Ok(response)) => {
tracing::info!(
"Cancelled want: {}",
response
.data
.as_ref()
.map(|w| &w.want_id)
.unwrap_or(&"unknown".to_string())
);
(StatusCode::OK, Json(response)).into_response()
}
Ok(Err(e)) => {
tracing::error!("Failed to cancel want: {}", e);
(
StatusCode::INTERNAL_SERVER_ERROR,
Json(ErrorResponse::new(format!("Failed to cancel want: {}", e))),
)
.into_response()
}
Err(_) => {
tracing::error!("Orchestrator did not respond to cancel want command");
(
StatusCode::INTERNAL_SERVER_ERROR,
Json(ErrorResponse::new("Orchestrator did not respond")),
)
.into_response()
}
}
}
@ -260,11 +330,10 @@ async fn list_partitions(
let events = match state.bel_storage.list_events(0, 100000) {
Ok(events) => events,
Err(e) => {
tracing::error!("Failed to read events from BEL storage: {}", e);
return (
StatusCode::INTERNAL_SERVER_ERROR,
Json(serde_json::json!({
"error": format!("Failed to read events: {}", e)
})),
Json(ErrorResponse::new(format!("Failed to read events: {}", e))),
)
.into_response();
}
@ -284,11 +353,10 @@ async fn list_job_runs(
let events = match state.bel_storage.list_events(0, 100000) {
Ok(events) => events,
Err(e) => {
tracing::error!("Failed to read events from BEL storage: {}", e);
return (
StatusCode::INTERNAL_SERVER_ERROR,
Json(serde_json::json!({
"error": format!("Failed to read events: {}", e)
})),
Json(ErrorResponse::new(format!("Failed to read events: {}", e))),
)
.into_response();
}