faucet_cli/serve/handlers/
catalog.rs

1//! `GET /v1/catalog/*` — browse the Data Movement Catalog (#279): the
2//! accumulated cross-run picture of every dataset the server's pipelines have
3//! touched. Read-only; all three routes require the `CatalogRead` permission
4//! (granted to every role, `viewer` up), enforced by the auth middleware.
5
6use crate::serve::error::ServeError;
7use crate::serve::history::catalog::{
8    CatalogDatasetDetail, CatalogDatasetPage, CatalogLineageEdge, CatalogListFilter,
9    LINEAGE_DEFAULT_DEPTH,
10};
11use crate::serve::state::ServerState;
12use axum::Json;
13use axum::extract::{Path, Query, State};
14use serde::{Deserialize, Serialize};
15
16const DEFAULT_LIMIT: usize = 100;
17const MAX_LIMIT: usize = 1000;
18const MAX_DEPTH: u32 = 32;
19
20/// `GET /v1/catalog/datasets` query string.
21#[derive(Debug, Deserialize)]
22pub struct DatasetsQuery {
23    /// Exact connector-kind filter (`csv`, `postgres`, …).
24    pub kind: Option<String>,
25    /// Case-insensitive substring match on the dataset URI.
26    pub q: Option<String>,
27    pub limit: Option<usize>,
28    pub cursor: Option<String>,
29}
30
31/// `GET /v1/catalog/datasets` → 200.
32pub async fn list_datasets(
33    State(state): State<ServerState>,
34    Query(query): Query<DatasetsQuery>,
35) -> Result<Json<CatalogDatasetPage>, ServeError> {
36    let filter = CatalogListFilter {
37        kind: query.kind,
38        q: query.q,
39        limit: query.limit.unwrap_or(DEFAULT_LIMIT).clamp(1, MAX_LIMIT),
40        cursor: query.cursor,
41    };
42    let page = state
43        .history()
44        .catalog_list_datasets(&filter)
45        .await
46        .map_err(|e| ServeError::Internal(e.to_string()))?;
47    Ok(Json(page))
48}
49
50/// `GET /v1/catalog/datasets/{id}` → 200 / 404.
51pub async fn get_dataset(
52    State(state): State<ServerState>,
53    Path(id): Path<String>,
54) -> Result<Json<CatalogDatasetDetail>, ServeError> {
55    let detail = state
56        .history()
57        .catalog_get_dataset(&id)
58        .await
59        .map_err(|e| ServeError::Internal(e.to_string()))?
60        .ok_or(ServeError::NotFound)?;
61    Ok(Json(detail))
62}
63
64/// `GET /v1/catalog/lineage` query string.
65#[derive(Debug, Deserialize)]
66pub struct LineageQuery {
67    /// Dataset id to root the graph at; omitted = the whole graph.
68    pub root: Option<String>,
69    /// BFS hop bound around `root` (ignored without one).
70    pub depth: Option<u32>,
71}
72
73/// `GET /v1/catalog/lineage` response body.
74#[derive(Debug, Serialize)]
75pub struct LineageResponse {
76    pub edges: Vec<CatalogLineageEdge>,
77}
78
79/// `GET /v1/catalog/lineage` → 200.
80pub async fn lineage(
81    State(state): State<ServerState>,
82    Query(query): Query<LineageQuery>,
83) -> Result<Json<LineageResponse>, ServeError> {
84    let depth = query
85        .depth
86        .unwrap_or(LINEAGE_DEFAULT_DEPTH)
87        .clamp(1, MAX_DEPTH);
88    let edges = state
89        .history()
90        .catalog_lineage(query.root.as_deref(), depth)
91        .await
92        .map_err(|e| ServeError::Internal(e.to_string()))?;
93    Ok(Json(LineageResponse { edges }))
94}
faucet_cli/serve/handlers/catalog.rs

faucet_cli/serve/handlers/
catalog.rs
