feat(ui): list the live catalog from the control API

clients/typescript/src/schema.d.ts

@@ -418,6 +418,22 @@ export interface paths {
         patch?: never;
         trace?: never;
     };
+    "/api/control/locations/{location_id}/targets/{target}/nodes": {
+        parameters: {
+            query?: never;
+            header?: never;
+            path?: never;
+            cookie?: never;
+        };
+        get: operations["listPersonCatalogNodes"];
+        put?: never;
+        post?: never;
+        delete?: never;
+        options?: never;
+        head?: never;
+        patch?: never;
+        trace?: never;
+    };
     "/api/control/locations/{location_id}/targets/{target}/nodes/{node_key}/sql": {
         parameters: {
             query?: never;
@@ -956,6 +972,31 @@ export interface components {
             /** Format: uuid */
             run_id: string;
         };
+        /** @description A fingerprint row's build identity; the `sqb` run that wrote it, not a scheduler run. */
+        CatalogNodeBuild: {
+            /** Format: date-time */
+            observed_at: string;
+            run_id: string;
+            version_hash: string;
+        };
+        /** @description A current snapshot node's declared metadata and latest build, without SQL, source or config. */
+        CatalogNodeRecord: {
+            column_count?: number | null;
+            /** @description Keys of the nodes this node reads. */
+            dependencies: string[];
+            description?: string | null;
+            /** @description The snapshot node kind, such as `model`, `seed`, `source` or `audit`. */
+            kind: string;
+            last_build?: null | components["schemas"]["CatalogNodeBuild"];
+            materialization?: string | null;
+            name: string;
+            node_key: string;
+            path?: string | null;
+            relation?: null | components["schemas"]["CatalogRelation"];
+            /** Format: uuid */
+            snapshot_id: string;
+            tags: string[];
+        };
         CatalogObservation: {
             /** Format: uuid */
             event_id: string;
@@ -967,6 +1008,11 @@ export interface components {
             next_after?: null | components["schemas"]["WarehouseMirrorCursor"];
             records: components["schemas"]["CatalogObservation"][];
         };
+        CatalogRelation: {
+            database?: string | null;
+            name: string;
+            schema: string;
+        };
         /** @description SSE invalidation topics; `all` means refetch the authorized REST resources. */
         ChangeNotice: {
             topic: string;
@@ -3080,6 +3126,44 @@ export interface operations {
             };
         };
     };
+    listPersonCatalogNodes: {
+        parameters: {
+            query?: never;
+            header?: never;
+            path: {
+                location_id: string;
+                target: string;
+            };
+            cookie?: never;
+        };
+        requestBody?: never;
+        responses: {
+            /** @description Every node of the current snapshot; empty before the first publication */
+            200: {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content: {
+                    "application/json": components["schemas"]["CatalogNodeRecord"][];
+                };
+            };
+            /** @description Session rejected */
+            401: {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content?: never;
+            };
+            403: {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content: {
+                    "application/json": components["schemas"]["ApiErrorBody"];
+                };
+            };
+        };
+    };
     readPersonNodeSql: {
         parameters: {
             query?: never;

crates/sqlbuild-orchestration/src/api/_helpers/authentication/password_sessions.rs

@@ -101,6 +101,10 @@ pub(crate) fn person_routes() -> Router<ControlStore> {
             "/api/control/locations/{location_id}/targets/{target}/runs/{run_id}/nodes",
             get(crate::api::_helpers::resources::person_reads::run_nodes),
         )
+        .route(
+            "/api/control/locations/{location_id}/targets/{target}/nodes",
+            get(crate::api::_helpers::resources::person_reads::nodes),
+        )
         .route(
             "/api/control/locations/{location_id}/targets/{target}/jobs",
             get(crate::api::_helpers::resources::person_reads::jobs),

crates/sqlbuild-orchestration/src/api/_helpers/resources/person_reads.rs

@@ -1,4 +1,4 @@
-//! Person reads of run nodes, jobs, triggers and ticks; refusal precedes every lookup, so unknown objects look forbidden.
+//! Person reads of run nodes, catalog nodes, jobs, triggers and ticks; refusal precedes every lookup, so unknown objects look forbidden.
 
 use axum::extract::{Path, Query, State};
 use axum::http::StatusCode;
@@ -16,6 +16,8 @@ use crate::runs::models::RunNodeRecord;
 use crate::scheduler::main::list_person_ticks::list_person_ticks;
 use crate::scheduler::models::TickRecord;
 use crate::store::models::ControlStore;
+use crate::warehouse_metadata::main::list_person_catalog_nodes::list_person_catalog_nodes;
+use crate::warehouse_metadata::models::CatalogNodeRecord;
 
 #[utoipa::path(get, path = "/api/control/locations/{location_id}/targets/{target}/runs/{run_id}/nodes", operation_id = "listPersonRunNodes",
     params(("location_id" = Uuid, Path), ("target" = String, Path), ("run_id" = Uuid, Path)),
@@ -40,6 +42,28 @@ pub(crate) async fn run_nodes(
     .ok_or_else(|| not_found("run not found"))
 }
 
+#[utoipa::path(get, path = "/api/control/locations/{location_id}/targets/{target}/nodes", operation_id = "listPersonCatalogNodes",
+    params(("location_id" = Uuid, Path), ("target" = String, Path)),
+    responses((status = 200, body = Vec<CatalogNodeRecord>, description = "Every node of the current snapshot; empty before the first publication"),
+        (status = 401, description = "Session rejected"),
+        (status = 403, body = crate::api::models::ApiErrorBody)), security(("session" = []), ("cliSession" = [])))]
+pub(crate) async fn nodes(
+    (State(store), Extension(context)): (State<ControlStore>, Extension<AuditContext>),
+    Extension(default_role): Extension<DefaultRole>,
+    Path((location_id, target)): Path<(Uuid, String)>,
+    Extension(AuthenticatedPerson(principal, _)): Extension<AuthenticatedPerson>,
+) -> Result<Json<Vec<CatalogNodeRecord>>, ApiError> {
+    Ok(Json(
+        list_person_catalog_nodes(
+            &store,
+            (&context, principal),
+            (location_id, &target),
+            default_role,
+        )
+        .await?,
+    ))
+}
+
 #[utoipa::path(get, path = "/api/control/locations/{location_id}/targets/{target}/jobs", operation_id = "listPersonJobs",
     params(("location_id" = Uuid, Path), ("target" = String, Path)),
     responses((status = 200, body = Vec<JobRecord>), (status = 401, description = "Session rejected"),

crates/sqlbuild-orchestration/src/api/models.rs

@@ -21,6 +21,7 @@ use crate::api::_helpers::resources::{
     person_operations::change_trigger,
     person_operations::recent_runs,
     person_reads::run_nodes,
+    person_reads::nodes,
     person_reads::jobs,
     person_reads::triggers,
     person_reads::ticks,

crates/sqlbuild-orchestration/src/warehouse_metadata/_helpers/catalog_nodes.rs

@@ -0,0 +1,80 @@
+//! Catalog listing rows from snapshot node definitions and each node's latest build.
+
+use serde_json::Value;
+use uuid::Uuid;
+
+use crate::sensors::constants::SOURCE_NODE_KIND;
+use crate::snapshots::models::SnapshotNode;
+use crate::warehouse_metadata::models::{CatalogNodeBuild, CatalogNodeRecord, CatalogRelation};
+
+/// A declared value from the definition itself, its schema entry or its config, in that order.
+fn declared<'a>(definition: &'a Value, key: &str) -> Option<&'a Value> {
+    [
+        definition.get(key),
+        definition.get("schema").and_then(|entry| entry.get(key)),
+        definition.get("config").and_then(|config| config.get(key)),
+    ]
+    .into_iter()
+    .flatten()
+    .find(|value| !value.is_null())
+}
+
+fn texts(value: Option<&Value>) -> Vec<String> {
+    match value {
+        Some(Value::String(text)) => vec![text.clone()],
+        Some(Value::Array(items)) => items
+            .iter()
+            .filter_map(|item| item.as_str().map(str::to_owned))
+            .collect(),
+        _ => Vec::new(),
+    }
+}
+
+fn text(value: Option<&Value>) -> Option<String> {
+    value.and_then(Value::as_str).map(str::to_owned)
+}
+
+fn relation(node: &SnapshotNode) -> Option<CatalogRelation> {
+    let definition = &node.definition;
+    let (relation, name_key) = match definition.get("destination") {
+        Some(destination) if !destination.is_null() => (destination, "name"),
+        _ if node.kind == SOURCE_NODE_KIND => (definition, "table"),
+        _ => return None,
+    };
+    Some(CatalogRelation {
+        database: text(relation.get("database")),
+        schema: text(relation.get("schema"))?,
+        name: text(relation.get(name_key))?,
+    })
+}
+
+pub(crate) fn catalog_node(
+    (node_key, node): (&str, &SnapshotNode),
+    snapshot_id: Uuid,
+    last_build: Option<CatalogNodeBuild>,
+) -> CatalogNodeRecord {
+    let definition = &node.definition;
+    CatalogNodeRecord {
+        node_key: node_key.into(),
+        kind: node.kind.clone(),
+        name: text(definition.get("name")).unwrap_or_else(|| {
+            node_key
+                .split_once(':')
+                .map_or(node_key, |(_, name)| name)
+                .into()
+        }),
+        path: text(definition.get("path")),
+        materialization: text(definition.get("materialization")),
+        description: text(declared(definition, "description")),
+        tags: texts(declared(definition, "tags")),
+        dependencies: texts(definition.get("deps")),
+        relation: relation(node),
+        column_count: match definition.get("columns") {
+            Some(Value::Array(columns)) => Some(columns.len()),
+            Some(Value::Object(columns)) => Some(columns.len()),
+            _ => None,
+        },
+        last_build,
+        snapshot_id,
+    }
+}