feat(ui): wire the lineage and changes pages to the live control API

clients/typescript/src/schema.d.ts

@@ -482,6 +482,22 @@ export interface paths {
         patch?: never;
         trace?: never;
     };
+    "/api/control/locations/{location_id}/targets/{target}/snapshots": {
+        parameters: {
+            query?: never;
+            header?: never;
+            path?: never;
+            cookie?: never;
+        };
+        get: operations["listPersonSnapshots"];
+        put?: never;
+        post?: never;
+        delete?: never;
+        options?: never;
+        head?: never;
+        patch?: never;
+        trace?: never;
+    };
     "/api/control/locations/{location_id}/targets/{target}/statistics": {
         parameters: {
             query?: never;
@@ -1488,6 +1504,28 @@ export interface components {
         };
         /** @enum {string} */
         PlatformEventKind: "heartbeat" | "scheduler_no_leader" | "tick_overdue" | "calendar_horizon_low" | "location_broken" | "launch_failed";
+        /** @description A changed subject, such as `model:orders`, or `role:analyst` (`role:` for those who can't assign roles). */
+        PublishedChange: {
+            change_kind: string;
+            subject_key: string;
+        };
+        /** @description One publication as people browse it: its commit, environment and changed definitions. */
+        PublishedSnapshot: {
+            /** @description Added, changed and removed nodes and roles, in publication order. */
+            changes: components["schemas"]["PublishedChange"][];
+            /** @description Digest of the lock file, Python and `sqb` versions; a change means dependencies changed. */
+            environment_hash: string;
+            /** Format: int64 */
+            generation: number;
+            git_author?: string | null;
+            git_dirty: boolean;
+            git_revision?: string | null;
+            /** Format: date-time */
+            published_at: string;
+            /** Format: uuid */
+            snapshot_id: string;
+            sqb_version: string;
+        };
         /** @description One invocation's records; the server owns its claim namespace, start time and version. */
         RecallBatch: {
             invocation_id: string;
@@ -3286,6 +3324,47 @@ export interface operations {
             };
         };
     };
+    listPersonSnapshots: {
+        parameters: {
+            query?: {
+                /** @description Maximum recent publications; defaults to 20, capped at 100 */
+                limit?: number;
+            };
+            header?: never;
+            path: {
+                location_id: string;
+                target: string;
+            };
+            cookie?: never;
+        };
+        requestBody?: never;
+        responses: {
+            /** @description Publications newest first, each with its changed nodes and roles */
+            200: {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content: {
+                    "application/json": components["schemas"]["PublishedSnapshot"][];
+                };
+            };
+            /** @description Session rejected */
+            401: {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content?: never;
+            };
+            403: {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content: {
+                    "application/json": components["schemas"]["ApiErrorBody"];
+                };
+            };
+        };
+    };
     readPersonStatistics: {
         parameters: {
             query: {

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

@@ -117,6 +117,10 @@ pub(crate) fn person_routes() -> Router<ControlStore> {
             "/api/control/locations/{location_id}/targets/{target}/triggers/{trigger_id}/ticks",
             get(crate::api::_helpers::resources::person_reads::ticks),
         )
+        .route(
+            "/api/control/locations/{location_id}/targets/{target}/snapshots",
+            get(crate::api::_helpers::resources::person_reads::snapshots),
+        )
         .route(
             "/api/control/locations/{location_id}/targets/{target}/catalog",
             get(crate::api::_helpers::resources::person_operations::catalog),

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

@@ -1,4 +1,4 @@
-//! Person reads of run nodes, catalog nodes, jobs, triggers and ticks; refusal precedes every lookup, so unknown objects look forbidden.
+//! Person reads of run nodes, catalog nodes, jobs, triggers, ticks and publications; refusal precedes every lookup, so unknown objects look forbidden.
 
 use axum::extract::{Path, Query, State};
 use axum::http::StatusCode;
@@ -15,6 +15,8 @@ use crate::runs::main::list_person_run_nodes::list_person_run_nodes;
 use crate::runs::models::RunNodeRecord;
 use crate::scheduler::main::list_person_ticks::list_person_ticks;
 use crate::scheduler::models::TickRecord;
+use crate::snapshots::main::list_person_snapshots::list_person_snapshots;
+use crate::snapshots::models::PublishedSnapshot;
 use crate::store::models::ControlStore;
 use crate::warehouse_metadata::main::list_person_catalog_nodes::list_person_catalog_nodes;
 use crate::warehouse_metadata::models::CatalogNodeRecord;
@@ -136,6 +138,29 @@ pub(crate) async fn ticks(
     .ok_or_else(|| not_found("trigger not found"))
 }
 
+#[utoipa::path(get, path = "/api/control/locations/{location_id}/targets/{target}/snapshots", operation_id = "listPersonSnapshots",
+    params(("location_id" = Uuid, Path), ("target" = String, Path),
+        ("limit" = Option<u32>, Query, description = "Maximum recent publications; defaults to 20, capped at 100")),
+    responses((status = 200, body = Vec<PublishedSnapshot>, description = "Publications newest first, each with its changed nodes and roles"),
+        (status = 401, description = "Session rejected"),
+        (status = 403, body = crate::api::models::ApiErrorBody)), security(("session" = []), ("cliSession" = [])))]
+pub(crate) async fn snapshots(
+    (State(store), Extension(context)): (State<ControlStore>, Extension<AuditContext>),
+    Extension(default_role): Extension<DefaultRole>,
+    (Path((location_id, target)), Query(options)): (Path<(Uuid, String)>, Query<ReadQuery>),
+    Extension(AuthenticatedPerson(principal, _)): Extension<AuthenticatedPerson>,
+) -> Result<Json<Vec<PublishedSnapshot>>, ApiError> {
+    Ok(Json(
+        list_person_snapshots(
+            &store,
+            (&context, principal),
+            ((location_id, &target), options.limit.unwrap_or(20)),
+            default_role,
+        )
+        .await?,
+    ))
+}
+
 fn not_found(message: &str) -> ApiError {
     ApiError {
         status: StatusCode::NOT_FOUND,

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

@@ -25,6 +25,7 @@ use crate::api::_helpers::resources::{
     person_reads::jobs,
     person_reads::triggers,
     person_reads::ticks,
+    person_reads::snapshots,
     person_operations::catalog,
     person_operations::statistics,
     person_operations::fingerprint,

crates/sqlbuild-orchestration/src/authorization/main/mod.rs

@@ -7,6 +7,7 @@ pub mod change_administrator;
 pub mod compile_roles;
 pub(crate) mod deny_authorization;
 pub mod grant_admin;
+pub(crate) mod person_assigns_roles;
 pub(crate) mod person_lists_target;
 pub(crate) mod publish_custom_roles;
 pub(crate) mod publish_group_roles;

crates/sqlbuild-orchestration/src/authorization/main/person_assigns_roles.rs

@@ -0,0 +1,26 @@
+//! Whether a person may assign roles server-wide; unaudited, for deciding what a read may name.
+
+use std::collections::BTreeSet;
+
+use uuid::Uuid;
+
+use crate::authorization::_helpers::person_grants::{allowed, person_role_rows};
+use crate::credentials::models::{Action, DefaultRole};
+use crate::store::errors::ControlError;
+use crate::store::models::ControlTransaction;
+
+/// Inside the caller's transaction, so the answer matches the rest of its read.
+pub(crate) async fn person_assigns_roles(
+    transaction: ControlTransaction,
+    principal_id: Uuid,
+) -> Result<(ControlTransaction, bool), ControlError> {
+    let (transaction, rows) = person_role_rows(transaction, principal_id, None).await?;
+    let assigns = allowed(
+        &rows.unwrap_or_default(),
+        (None, None),
+        &BTreeSet::new(),
+        DefaultRole::None,
+    )
+    .contains(&Action::RoleAssign);
+    Ok((transaction, assigns))
+}

crates/sqlbuild-orchestration/src/snapshots/main/list_person_snapshots.rs

@@ -0,0 +1,87 @@
+//! A person browses a target's publications only with view on every current node, like its jobs.
+
+use std::collections::BTreeMap;
+
+use chrono::{DateTime, Utc};
+use uuid::Uuid;
+
+use crate::authorization::main::authorize_listing::authorize_listing;
+use crate::authorization::main::person_assigns_roles::person_assigns_roles;
+use crate::credentials::models::{AuditContext, DefaultRole};
+use crate::snapshots::models::{PublishedChange, PublishedSnapshot};
+use crate::store::errors::ControlError;
+use crate::store::main::{begin::begin, commit::commit, decode_row::decode_row, query::query};
+use crate::store::models::{ControlStore, Parameter, TransactionMode};
+use crate::store::types::Row;
+
+/// Newest first, at most 100 publications; changed roles are named only for people who assign roles.
+pub async fn list_person_snapshots(
+    store: &ControlStore,
+    (context, credentials): (&AuditContext, Uuid),
+    ((location_id, target), limit): ((Uuid, &str), u32),
+    default_role: DefaultRole,
+) -> Result<Vec<PublishedSnapshot>, ControlError> {
+    let (transaction, _) = authorize_listing(
+        begin(store, TransactionMode::Write).await?,
+        (context, credentials),
+        (location_id, target),
+        default_role,
+    )
+    .await?;
+    let scope = vec![
+        Parameter::Id(location_id),
+        Parameter::Text(target.into()),
+        Parameter::Integer(i64::from(limit.min(100))),
+    ];
+    let (transaction, published) = query(transaction,
+        "SELECT snapshot_id, generation, git_revision, CASE WHEN git_dirty THEN 1 ELSE 0 END AS git_dirty, git_author, environment_hash, sqb_version, published_at \
+         FROM sqb_snapshots WHERE location_id=$1 AND target=$2 ORDER BY generation DESC LIMIT $3",
+        scope.clone()).await?;
+    let (transaction, changed) = query(transaction,
+        "SELECT c.snapshot_id, c.change_kind, c.subject_key FROM sqb_snapshot_changes c \
+         WHERE c.snapshot_id IN (SELECT s.snapshot_id FROM sqb_snapshots s WHERE s.location_id=$1 AND s.target=$2 \
+           ORDER BY s.generation DESC LIMIT $3) ORDER BY c.snapshot_id, c.change_seq",
+        scope).await?;
+    let (transaction, names_roles) = person_assigns_roles(transaction, credentials).await?;
+    commit(transaction).await?;
+    let mut changes: BTreeMap<Uuid, Vec<PublishedChange>> = BTreeMap::new();
+    for row in changed {
+        let snapshot_id = id_of(&row)?;
+        let mut change: PublishedChange = decode_row(row)?;
+        if !names_roles && change.subject_key.starts_with("role:") {
+            change.subject_key = "role:".into();
+        }
+        changes.entry(snapshot_id).or_default().push(change);
+    }
+    published
+        .into_iter()
+        .map(|row| {
+            let mut snapshot = snapshot_of(row)?;
+            snapshot.changes = changes.remove(&snapshot.snapshot_id).unwrap_or_default();
+            Ok(snapshot)
+        })
+        .collect()
+}
+
+fn snapshot_of(row: Row) -> Result<PublishedSnapshot, ControlError> {
+    let snapshot_id = id_of(&row)?;
+    let text = |key: &str| row[key].as_str().map(String::from);
+    let published_at: DateTime<Utc> = serde_json::from_value(row["published_at"].clone())
+        .map_err(|error| ControlError::Invalid(format!("stored publication time: {error}")))?;
+    Ok(PublishedSnapshot {
+        snapshot_id,
+        generation: row["generation"].as_i64().unwrap_or_default(),
+        git_revision: text("git_revision"),
+        git_dirty: row["git_dirty"].as_i64() == Some(1),
+        git_author: text("git_author"),
+        environment_hash: text("environment_hash").unwrap_or_default(),
+        sqb_version: text("sqb_version").unwrap_or_default(),
+        published_at,
+        changes: Vec::new(),
+    })
+}
+
+fn id_of(row: &Row) -> Result<Uuid, ControlError> {
+    serde_json::from_value(row["snapshot_id"].clone())
+        .map_err(|error| ControlError::Invalid(format!("stored snapshot id: {error}")))
+}

crates/sqlbuild-orchestration/src/snapshots/main/mod.rs

@@ -2,6 +2,7 @@ pub(super) mod _publish_snapshot;
 pub mod current_snapshot;
 pub mod current_snapshot_record;
 pub(crate) mod job_write_set;
+pub mod list_person_snapshots;
 pub mod next_remote_generation;
 pub(crate) mod prune_snapshots;
 pub(crate) mod publish_development_snapshot;

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

@@ -122,3 +122,26 @@ impl Default for SnapshotRetention {
         }
     }
 }
+
+/// One publication as people browse it: its commit, environment and changed definitions.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, utoipa::ToSchema)]
+pub struct PublishedSnapshot {
+    pub snapshot_id: Uuid,
+    pub generation: i64,
+    pub git_revision: Option<String>,
+    pub git_dirty: bool,
+    pub git_author: Option<String>,
+    /// Digest of the lock file, Python and `sqb` versions; a change means dependencies changed.
+    pub environment_hash: String,
+    pub sqb_version: String,
+    pub published_at: DateTime<Utc>,
+    /// Added, changed and removed nodes and roles, in publication order.
+    pub changes: Vec<PublishedChange>,
+}
+
+/// A changed subject, such as `model:orders`, or `role:analyst` (`role:` for those who can't assign roles).
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)]
+pub struct PublishedChange {
+    pub change_kind: String,
+    pub subject_key: String,
+}

crates/sqlbuild-orchestration/tests/api/test_person_sessions.rs

@@ -288,6 +288,7 @@ fn given_browser_session_roles_when_work_is_requested_then_scope_revocation_and_
                         format!("/api/control/locations/{location}/targets/development/nodes"),
                         format!("/api/control/locations/{location}/targets/development/jobs"),
                         format!("/api/control/locations/{location}/targets/development/triggers"),
+                        format!("/api/control/locations/{location}/targets/development/snapshots"),
                         format!(
                             "/api/control/locations/{location}/targets/development/triggers/{}/ticks",
                             trigger.trigger_id