bestool_alertd/http_server/endpoints/
tasks.rs

1use std::{collections::BTreeMap, sync::Arc};
2
3use axum::{
4	Json,
5	body::Body,
6	extract::{Path, Query, State},
7	http::{HeaderValue, StatusCode, header::CONTENT_TYPE},
8	response::{IntoResponse, Response},
9};
10use futures::StreamExt;
11use tracing::warn;
12
13use crate::{
14	http_server::state::ServerState,
15	tasks::{TaskContext, TaskEndpointResponse},
16};
17
18/// Route handler for `/tasks/:task/:endpoint`.
19///
20/// Looks up the handler the named background task exposed (via
21/// `BackgroundTask::http_endpoints`) and invokes it with a fresh
22/// `TaskContext` built from the daemon's internal resources. The handler's
23/// `TaskEndpointResponse` is serialised to JSON or NDJSON depending on its
24/// variant.
25pub async fn handle_task_endpoint(
26	State(state): State<Arc<ServerState>>,
27	Path((task, endpoint)): Path<(String, String)>,
28	Query(query): Query<BTreeMap<String, String>>,
29) -> Response {
30	let Some(handler) = state.task_endpoints.get(&(task.clone(), endpoint.clone())) else {
31		return (
32			StatusCode::NOT_FOUND,
33			format!("no endpoint at /tasks/{task}/{endpoint}"),
34		)
35			.into_response();
36	};
37
38	let mut ctx = TaskContext::from_internal(&state.internal_context);
39	ctx.query = query;
40	let response = handler(ctx).await;
41
42	match response {
43		TaskEndpointResponse::Json(value) => Json(value).into_response(),
44		TaskEndpointResponse::JsonLines(stream) => {
45			let body = Body::from_stream(stream.map(|value| {
46				// One JSON value per line, NDJSON style. Newline-on-end keeps
47				// the last record syntactically self-contained for readers
48				// using `read_line`-style framing.
49				let mut bytes = serde_json::to_vec(&value).unwrap_or_else(|err| {
50					warn!(%err, "could not serialise task endpoint stream value");
51					b"{}".to_vec()
52				});
53				bytes.push(b'\n');
54				Ok::<_, std::convert::Infallible>(bytes)
55			}));
56			let mut response = Response::new(body);
57			response.headers_mut().insert(
58				CONTENT_TYPE,
59				HeaderValue::from_static("application/x-ndjson"),
60			);
61			response
62		}
63		TaskEndpointResponse::Error { status, message } => (
64			StatusCode::from_u16(status).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR),
65			message,
66		)
67			.into_response(),
68	}
69}
bestool_alertd/http_server/endpoints/tasks.rs

bestool_alertd/http_server/endpoints/
tasks.rs
