diff --git a/benchmarks/duckshop/edge_cases/orchestration/maintenance_history.json b/benchmarks/duckshop/edge_cases/orchestration/maintenance_history.json new file mode 100644 index 0000000000..860b31bec6 --- /dev/null +++ b/benchmarks/duckshop/edge_cases/orchestration/maintenance_history.json @@ -0,0 +1,81 @@ +{ + "migrations": [ + { + "event_id": "move-orders-1", + "destination_model": "orders", + "origin_model": "orders_v1", + "origin_name": "orders_v1", + "decision": "migrate", + "created_at": "2026-02-01T04:05:06Z" + }, + { + "event_id": "move-orders-2", + "destination_model": "orders", + "origin_model": null, + "origin_name": "orders_🧾", + "decision": "renamed", + "discovery": "automatic", + "created_at": "2026-02-03T04:05:06Z" + }, + { + "event_id": "move-customers-1", + "destination_model": "customers", + "origin_model": "clients", + "origin_name": "clients", + "decision": "superseded_replace", + "created_at": "2026-02-02T04:05:06Z" + } + ], + "old_name_views": [ + { + "event_id": "view-orders-1", + "migration_event_id": "move-orders-2", + "destination_model": "orders", + "event_type": "required", + "created_at": "2026-02-03T04:05:06Z" + }, + { + "event_id": "view-orders-2", + "migration_event_id": "move-orders-2", + "destination_model": "orders", + "event_type": "view_created", + "created_at": "2026-02-03T04:05:07Z", + "column_aliases": "{\"order_id\":\"legacy_order_id\"}" + }, + { + "event_id": "view-customers-1", + "migration_event_id": "move-customers-1", + "destination_model": "customers", + "event_type": "required", + "created_at": "2026-02-02T04:05:06Z" + }, + { + "event_id": "view-orders-old-1", + "migration_event_id": "move-orders-1", + "destination_model": "orders", + "event_type": "required", + "created_at": "2026-02-01T04:05:06Z" + } + ], + "janitor_events": [ + { + "event_id": "cleanup-1", + "event_type": "archive", + "occurred_at": "2026-02-01T05:05:06Z" + }, + { + "event_id": "cleanup-2", + "event_type": "delete", + "occurred_at": "2026-02-04T05:05:06Z" + } + ], + "preview_old_name_views": [ + { + "event_id": "view-preview-1", + "migration_event_id": "move-orders-2", + "destination_model": "orders", + "event_type": "view_dropped", + "created_at": "2026-02-05T04:05:06Z" + } + ] +} diff --git a/clients/typescript/src/schema.d.ts b/clients/typescript/src/schema.d.ts index c2f2ab1fdb..ae6e30de48 100644 --- a/clients/typescript/src/schema.d.ts +++ b/clients/typescript/src/schema.d.ts @@ -322,6 +322,22 @@ export interface paths { patch?: never; trace?: never; }; + "/api/control/locations/{location_id}/targets/{target}/maintenance": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + get: operations["readPersonMaintenance"]; + put?: never; + post?: never; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; "/api/control/locations/{location_id}/targets/{target}/node-status": { parameters: { query?: never; @@ -1121,6 +1137,14 @@ export interface components { /** Format: uuid */ session_id: string; }; + /** @description Recorded moves, their old-name view steps and janitor cleanups, newest moves first. */ + MaintenanceHistory: { + /** @description Empty when the history is for one model; cleanups are not tied to models. */ + janitor_events: components["schemas"]["WarehouseJanitorRecord"][]; + migrations: components["schemas"]["WarehouseMigrationRecord"][]; + /** @description Every step recorded for the listed moves, oldest first; same-time steps are not in step order. */ + old_name_views: components["schemas"]["WarehouseOldNameRecord"][]; + }; /** @description One bounded page of one state table from the registered reader of a published generation. */ MetadataBatch: { /** Format: int64 */ @@ -1731,12 +1755,81 @@ export interface components { * @enum {string} */ WarehouseHistoryKind: "fingerprints" | "source_freshness" | "audit_results" | "janitor_events" | "migrations" | "old_name_views" | "microbatches" | "catalog" | "statistics"; + /** @description Warehouse cleanup facts do not authorize control-plane deletion or worker recovery. */ + WarehouseJanitorRecord: { + archive_name: string; + archive_qualified_name: string; + /** Format: date-time */ + archived_at: string; + event_id: string; + event_type: string; + /** Format: date-time */ + occurred_at: string; + original_name?: string | null; + original_qualified_name?: string | null; + relation_database?: string | null; + relation_schema: string; + relation_type?: string | null; + run_id: string; + /** Format: int64 */ + schema_version: number; + state_database?: string | null; + state_schema: string; + }; + /** @description Model-move history is evidence, not authority to migrate a warehouse relation. */ + WarehouseMigrationRecord: { + /** Format: date-time */ + created_at: string; + decision: string; + destination_database?: string | null; + destination_model: string; + destination_name: string; + destination_schema: string; + discovery: string; + event_id: string; + origin_database?: string | null; + origin_model?: string | null; + origin_name: string; + origin_schema?: string | null; + origin_version_hash: string; + run_id: string; + state_database?: string | null; + state_schema: string; + target_name?: string | null; + }; WarehouseMirrorCursor: { /** Format: uuid */ event_id: string; /** Format: date-time */ observed_at: string; }; + /** @description Preserve nullable compatibility-view facts without projecting live warehouse existence. */ + WarehouseOldNameRecord: { + archive_name?: string | null; + column_aliases?: string | null; + /** Format: date-time */ + created_at: string; + destination_model: string; + drop_reason?: string | null; + event_id: string; + event_type: string; + /** Format: date-time */ + expires_at?: string | null; + grants_copied?: string | null; + migration_event_id: string; + new_database?: string | null; + new_name: string; + new_schema: string; + old_database?: string | null; + old_name: string; + old_schema?: string | null; + run_id: string; + state_database?: string | null; + state_schema: string; + target_name?: string | null; + view_retention?: string | null; + view_sql?: string | null; + }; /** @description Physical namespace of SQLBuild's source-freshness state table, never a source query. */ WarehouseStateLocation: { database?: string | null; @@ -2547,6 +2640,48 @@ export interface operations { }; }; }; + readPersonMaintenance: { + parameters: { + query?: { + /** @description Only moves into this model, and no janitor events */ + node?: string; + /** @description Maximum moves and janitor events each; defaults to 100, capped at 1000 */ + limit?: number; + }; + header?: never; + path: { + location_id: string; + target: string; + }; + cookie?: never; + }; + requestBody?: never; + responses: { + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["MaintenanceHistory"]; + }; + }; + /** @description Session rejected */ + 401: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + 403: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["ApiErrorBody"]; + }; + }; + }; + }; readPersonNodeStatus: { parameters: { query: { diff --git a/crates/sqlbuild-orchestration/src/api/_helpers/authentication/password_sessions.rs b/crates/sqlbuild-orchestration/src/api/_helpers/authentication/password_sessions.rs index 171a2401d1..0a955d6235 100644 --- a/crates/sqlbuild-orchestration/src/api/_helpers/authentication/password_sessions.rs +++ b/crates/sqlbuild-orchestration/src/api/_helpers/authentication/password_sessions.rs @@ -125,6 +125,10 @@ pub(crate) fn person_routes() -> Router { "/api/control/locations/{location_id}/targets/{target}/checks/{binding_key}/results/{event_id}/evidence", get(crate::api::_helpers::resources::person_checks::evidence), ) + .route( + "/api/control/locations/{location_id}/targets/{target}/maintenance", + get(crate::api::_helpers::resources::person_maintenance::read), + ) .layer(DefaultBodyLimit::max(PASSWORD_BODY_BYTES)) } diff --git a/crates/sqlbuild-orchestration/src/api/_helpers/resources/mod.rs b/crates/sqlbuild-orchestration/src/api/_helpers/resources/mod.rs index 3b6efd68ac..be262332c1 100644 --- a/crates/sqlbuild-orchestration/src/api/_helpers/resources/mod.rs +++ b/crates/sqlbuild-orchestration/src/api/_helpers/resources/mod.rs @@ -7,6 +7,7 @@ pub(crate) mod location_health; pub(crate) mod location_work; pub(crate) mod person_checks; pub(crate) mod person_location_work; +pub(crate) mod person_maintenance; pub(crate) mod person_node_sql; pub(crate) mod person_operations; pub(crate) mod router; diff --git a/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_checks.rs b/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_checks.rs index 8f210a579c..8fbb6f4fb2 100644 --- a/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_checks.rs +++ b/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_checks.rs @@ -8,11 +8,11 @@ use crate::api::errors::ApiError; use crate::api::models::{CheckResultPath, ReadQuery}; use crate::credentials::models::{AuditContext, AuthenticatedPerson, DefaultRole}; use crate::store::models::ControlStore; -use crate::warehouse_metadata::main::read_person_check::read_person_check; -use crate::warehouse_metadata::main::read_person_check_evidence::read_person_check_evidence; -use crate::warehouse_metadata::main::read_person_check_sql::read_person_check_sql; -use crate::warehouse_metadata::main::read_person_checks::read_person_checks; -use crate::warehouse_metadata::models::{CheckEvidence, CheckResult, CheckSql}; +use crate::warehouse_history::main::read_person_check::read_person_check; +use crate::warehouse_history::main::read_person_check_evidence::read_person_check_evidence; +use crate::warehouse_history::main::read_person_check_sql::read_person_check_sql; +use crate::warehouse_history::main::read_person_checks::read_person_checks; +use crate::warehouse_history::models::{CheckEvidence, CheckResult, CheckSql}; #[utoipa::path(get, path = "/api/control/locations/{location_id}/targets/{target}/checks", operation_id = "listPersonChecks", params(("location_id" = Uuid, Path), ("target" = String, Path)), diff --git a/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_maintenance.rs b/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_maintenance.rs new file mode 100644 index 0000000000..1c704c0b52 --- /dev/null +++ b/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_maintenance.rs @@ -0,0 +1,34 @@ +//! Maintenance history reads use live person grants over the whole target. + +use axum::extract::{Path, Query, State}; +use axum::{Extension, Json}; +use uuid::Uuid; + +use crate::api::errors::ApiError; +use crate::credentials::models::{AuditContext, AuthenticatedPerson, DefaultRole}; +use crate::store::models::ControlStore; +use crate::warehouse_history::main::read_person_maintenance::read_person_maintenance; +use crate::warehouse_history::models::{MaintenanceHistory, MaintenanceQuery}; + +#[utoipa::path(get, path = "/api/control/locations/{location_id}/targets/{target}/maintenance", operation_id = "readPersonMaintenance", + params(("location_id" = Uuid, Path), ("target" = String, Path), + ("node" = Option, Query, description = "Only moves into this model, and no janitor events"), + ("limit" = Option, Query, description = "Maximum moves and janitor events each; defaults to 100, capped at 1000")), + responses((status = 200, body = MaintenanceHistory), (status = 401, description = "Session rejected"), + (status = 403, body = crate::api::models::ApiErrorBody)), security(("session" = []), ("cliSession" = [])))] +pub(crate) async fn read( + (State(store), Extension(context)): (State, Extension), + Extension(default_role): Extension, + (Path((location_id, target)), Query(options)): (Path<(Uuid, String)>, Query), + Extension(AuthenticatedPerson(principal, _)): Extension, +) -> Result, ApiError> { + Ok(Json( + read_person_maintenance( + &store, + (&context, principal), + (location_id, &target, &options), + default_role, + ) + .await?, + )) +} diff --git a/crates/sqlbuild-orchestration/src/api/models.rs b/crates/sqlbuild-orchestration/src/api/models.rs index 16288de89a..0174bb33c6 100644 --- a/crates/sqlbuild-orchestration/src/api/models.rs +++ b/crates/sqlbuild-orchestration/src/api/models.rs @@ -6,8 +6,8 @@ use crate::api::_helpers::authentication::password_sessions; use crate::api::_helpers::authentication::{cli_login, oidc}; use crate::api::_helpers::resources::{ administrators, audit, changes, location_health, location_work, person_checks, - person_location_work, person_node_sql, person_operations, run_reads, runs, snapshots, - trigger_reads, warehouse_metadata, + person_location_work, person_maintenance, person_node_sql, person_operations, run_reads, runs, + snapshots, trigger_reads, warehouse_metadata, }; #[derive(Debug, utoipa::OpenApi)] @@ -28,6 +28,7 @@ use crate::api::_helpers::resources::{ person_checks::history, person_checks::sql, person_checks::evidence, + person_maintenance::read, person_location_work::request, person_location_work::result, person_node_sql::read, diff --git a/crates/sqlbuild-orchestration/src/lib.rs b/crates/sqlbuild-orchestration/src/lib.rs index 87aebeb15e..213b03ccff 100644 --- a/crates/sqlbuild-orchestration/src/lib.rs +++ b/crates/sqlbuild-orchestration/src/lib.rs @@ -12,6 +12,7 @@ pub mod scheduler; pub mod sensors; pub mod snapshots; pub mod store; +pub mod warehouse_history; pub mod warehouse_metadata; pub mod webhooks; pub mod worker; diff --git a/crates/sqlbuild-orchestration/src/store/schema.sql b/crates/sqlbuild-orchestration/src/store/schema.sql index 39a45c68a7..537f4ccdd4 100644 --- a/crates/sqlbuild-orchestration/src/store/schema.sql +++ b/crates/sqlbuild-orchestration/src/store/schema.sql @@ -429,6 +429,7 @@ CREATE TABLE sqb_warehouse_migrations ( scan_id {uuid} ); CREATE INDEX sqb_migrations_namespace ON sqb_warehouse_migrations (location_id,target,state_database,state_schema,observed_at,event_id); +CREATE INDEX sqb_migrations_destination ON sqb_warehouse_migrations (location_id,target,destination_model,observed_at,event_id); CREATE TABLE sqb_warehouse_old_name_views ( event_id {uuid} PRIMARY KEY, location_id {uuid} NOT NULL REFERENCES sqb_locations(location_id), @@ -444,6 +445,7 @@ CREATE TABLE sqb_warehouse_old_name_views ( scan_id {uuid} ); CREATE INDEX sqb_old_name_views_namespace ON sqb_warehouse_old_name_views (location_id,target,state_database,state_schema,observed_at,event_id); +CREATE INDEX sqb_old_name_views_migration ON sqb_warehouse_old_name_views (location_id,target,migration_event_id,observed_at,event_id); CREATE TABLE sqb_warehouse_microbatches ( event_id {uuid} PRIMARY KEY, location_id {uuid} NOT NULL REFERENCES sqb_locations(location_id), diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/_helpers/checks.rs b/crates/sqlbuild-orchestration/src/warehouse_history/_helpers/checks.rs similarity index 98% rename from crates/sqlbuild-orchestration/src/warehouse_metadata/_helpers/checks.rs rename to crates/sqlbuild-orchestration/src/warehouse_history/_helpers/checks.rs index 34340e5fd7..d4a78fbad3 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_metadata/_helpers/checks.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_history/_helpers/checks.rs @@ -12,7 +12,8 @@ use crate::credentials::models::{Action, AuditContext, DefaultRole, PermissionSc use crate::store::errors::ControlError; use crate::store::main::{commit::commit, query::query, query_records::query_records}; use crate::store::models::{ControlTransaction, Parameter}; -use crate::warehouse_metadata::models::{CheckEvidence, WarehouseAuditRecord}; +use crate::warehouse_history::models::CheckEvidence; +use crate::warehouse_metadata::models::WarehouseAuditRecord; #[derive(Deserialize)] pub(crate) struct AuditRow { diff --git a/crates/sqlbuild-orchestration/src/warehouse_history/_helpers/maintenance.rs b/crates/sqlbuild-orchestration/src/warehouse_history/_helpers/maintenance.rs new file mode 100644 index 0000000000..279f7a16dc --- /dev/null +++ b/crates/sqlbuild-orchestration/src/warehouse_history/_helpers/maintenance.rs @@ -0,0 +1,86 @@ +//! Maintenance reads over the migration, old-name view and janitor mirrors. + +use serde::Deserialize; +use serde::de::DeserializeOwned; +use uuid::Uuid; + +use crate::store::errors::ControlError; +use crate::store::main::query_records::query_records; +use crate::store::models::{ControlTransaction, Parameter}; +use crate::warehouse_history::models::MaintenanceHistory; +use crate::warehouse_metadata::models::{ + WarehouseJanitorRecord, WarehouseMigrationRecord, WarehouseOldNameRecord, +}; + +#[derive(Deserialize)] +struct Recorded { + record: T, +} + +async fn records( + transaction: ControlTransaction, + sql: &str, + parameters: Vec, +) -> Result<(ControlTransaction, Vec), ControlError> { + let (transaction, rows): (ControlTransaction, Vec>) = + query_records(transaction, sql, parameters).await?; + Ok(( + transaction, + rows.into_iter().map(|row| row.record).collect(), + )) +} + +/// The newest moves, optionally into one model, with every step of their old-name views. +pub(crate) async fn maintenance_history( + transaction: ControlTransaction, + (location_id, target): (Uuid, &str), + (node, limit): (Option<&str>, u32), +) -> Result<(ControlTransaction, MaintenanceHistory), ControlError> { + let limit = Parameter::Integer(i64::from(limit.clamp(1, 1000))); + let scope = vec![Parameter::Id(location_id), Parameter::Text(target.into())]; + let (transaction, migrations): (ControlTransaction, Vec) = + match node { + Some(node) => { + records(transaction, + "SELECT record_json AS record FROM sqb_warehouse_migrations WHERE location_id=$1 AND target=$2 \ + AND destination_model=$3 ORDER BY observed_at DESC,event_id DESC LIMIT $4", + [scope.clone(), vec![Parameter::Text(node.into()), limit.clone()]].concat()).await? + } + None => { + records(transaction, + "SELECT record_json AS record FROM sqb_warehouse_migrations WHERE location_id=$1 AND target=$2 \ + ORDER BY observed_at DESC,event_id DESC LIMIT $3", + [scope.clone(), vec![limit.clone()]].concat()).await? + } + }; + let mut transaction = transaction; + let mut old_name_views: Vec = Vec::new(); + if !migrations.is_empty() { + let mut parameters = scope.clone(); + parameters.extend( + migrations + .iter() + .map(|migration| Parameter::Text(migration.event_id.clone())), + ); + let placeholders = Parameter::placeholders(3..=parameters.len()); + (transaction, old_name_views) = records(transaction, + &format!("SELECT record_json AS record FROM sqb_warehouse_old_name_views WHERE location_id=$1 AND target=$2 \ + AND migration_event_id IN ({placeholders}) ORDER BY observed_at,event_id"), + parameters).await?; + } + let mut janitor_events: Vec = Vec::new(); + if node.is_none() { + (transaction, janitor_events) = records(transaction, + "SELECT record_json AS record FROM sqb_warehouse_janitor_events WHERE location_id=$1 AND target=$2 \ + ORDER BY observed_at DESC,event_id DESC LIMIT $3", + [scope, vec![limit]].concat()).await?; + } + Ok(( + transaction, + MaintenanceHistory { + migrations, + old_name_views, + janitor_events, + }, + )) +} diff --git a/crates/sqlbuild-orchestration/src/warehouse_history/_helpers/mod.rs b/crates/sqlbuild-orchestration/src/warehouse_history/_helpers/mod.rs new file mode 100644 index 0000000000..e0260321e5 --- /dev/null +++ b/crates/sqlbuild-orchestration/src/warehouse_history/_helpers/mod.rs @@ -0,0 +1,2 @@ +pub(crate) mod checks; +pub(crate) mod maintenance; diff --git a/crates/sqlbuild-orchestration/src/warehouse_history/main/mod.rs b/crates/sqlbuild-orchestration/src/warehouse_history/main/mod.rs new file mode 100644 index 0000000000..5a314e91ca --- /dev/null +++ b/crates/sqlbuild-orchestration/src/warehouse_history/main/mod.rs @@ -0,0 +1,5 @@ +pub mod read_person_check; +pub mod read_person_check_evidence; +pub mod read_person_check_sql; +pub mod read_person_checks; +pub mod read_person_maintenance; diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_check.rs b/crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_check.rs similarity index 90% rename from crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_check.rs rename to crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_check.rs index c3f9a121c1..e48000c366 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_check.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_check.rs @@ -7,8 +7,8 @@ use crate::credentials::models::{AuditContext, DefaultRole}; use crate::store::errors::ControlError; use crate::store::main::{begin::begin, commit::commit}; use crate::store::models::{ControlStore, TransactionMode}; -use crate::warehouse_metadata::_helpers::checks::check_results; -use crate::warehouse_metadata::models::CheckResult; +use crate::warehouse_history::_helpers::checks::check_results; +use crate::warehouse_history::models::CheckResult; pub async fn read_person_check( store: &ControlStore, diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_check_evidence.rs b/crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_check_evidence.rs similarity index 84% rename from crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_check_evidence.rs rename to crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_check_evidence.rs index 8132c3e6f5..2efdbabf0a 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_check_evidence.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_check_evidence.rs @@ -6,8 +6,8 @@ use crate::credentials::models::{Action, AuditContext}; use crate::store::errors::ControlError; use crate::store::main::begin::begin; use crate::store::models::{ControlStore, TransactionMode}; -use crate::warehouse_metadata::_helpers::checks::{check_result, evidence, sensitive_check_read}; -use crate::warehouse_metadata::models::CheckEvidence; +use crate::warehouse_history::_helpers::checks::{check_result, evidence, sensitive_check_read}; +use crate::warehouse_history::models::CheckEvidence; pub async fn read_person_check_evidence( store: &ControlStore, diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_check_sql.rs b/crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_check_sql.rs similarity index 90% rename from crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_check_sql.rs rename to crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_check_sql.rs index 6e615cd52b..32f4f0c1e4 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_check_sql.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_check_sql.rs @@ -6,8 +6,8 @@ use crate::credentials::models::{Action, AuditContext}; use crate::store::errors::ControlError; use crate::store::main::begin::begin; use crate::store::models::{ControlStore, TransactionMode}; -use crate::warehouse_metadata::_helpers::checks::{check_results, sensitive_check_read}; -use crate::warehouse_metadata::models::CheckSql; +use crate::warehouse_history::_helpers::checks::{check_results, sensitive_check_read}; +use crate::warehouse_history::models::CheckSql; pub async fn read_person_check_sql( store: &ControlStore, diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_checks.rs b/crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_checks.rs similarity index 89% rename from crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_checks.rs rename to crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_checks.rs index 13632263e9..4a31d398bd 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_checks.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_checks.rs @@ -7,8 +7,8 @@ use crate::credentials::models::{AuditContext, DefaultRole}; use crate::store::errors::ControlError; use crate::store::main::{begin::begin, commit::commit}; use crate::store::models::{ControlStore, TransactionMode}; -use crate::warehouse_metadata::_helpers::checks::latest_results; -use crate::warehouse_metadata::models::CheckResult; +use crate::warehouse_history::_helpers::checks::latest_results; +use crate::warehouse_history::models::CheckResult; pub async fn read_person_checks( store: &ControlStore, diff --git a/crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_maintenance.rs b/crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_maintenance.rs new file mode 100644 index 0000000000..a74a92e6e8 --- /dev/null +++ b/crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_maintenance.rs @@ -0,0 +1,34 @@ +//! Live view permission over the whole target and its recorded maintenance history. + +use uuid::Uuid; + +use crate::authorization::main::authorize_listing::authorize_listing; +use crate::credentials::models::{AuditContext, DefaultRole}; +use crate::store::errors::ControlError; +use crate::store::main::{begin::begin, commit::commit}; +use crate::store::models::{ControlStore, TransactionMode}; +use crate::warehouse_history::_helpers::maintenance::maintenance_history; +use crate::warehouse_history::models::{MaintenanceHistory, MaintenanceQuery}; + +pub async fn read_person_maintenance( + store: &ControlStore, + credentials: (&AuditContext, Uuid), + (location_id, target, options): (Uuid, &str, &MaintenanceQuery), + default_role: DefaultRole, +) -> Result { + let (transaction, _) = authorize_listing( + begin(store, TransactionMode::Write).await?, + credentials, + (location_id, target), + default_role, + ) + .await?; + let (transaction, history) = maintenance_history( + transaction, + (location_id, target), + (options.node.as_deref(), options.limit.unwrap_or(100)), + ) + .await?; + commit(transaction).await?; + Ok(history) +} diff --git a/crates/sqlbuild-orchestration/src/warehouse_history/mod.rs b/crates/sqlbuild-orchestration/src/warehouse_history/mod.rs new file mode 100644 index 0000000000..7488923fa9 --- /dev/null +++ b/crates/sqlbuild-orchestration/src/warehouse_history/mod.rs @@ -0,0 +1,5 @@ +//! Person reads of warehouse-owned history mirrored into the control store: checks and maintenance. + +pub(crate) mod _helpers; +pub mod main; +pub mod models; diff --git a/crates/sqlbuild-orchestration/src/warehouse_history/models.rs b/crates/sqlbuild-orchestration/src/warehouse_history/models.rs new file mode 100644 index 0000000000..cca287650b --- /dev/null +++ b/crates/sqlbuild-orchestration/src/warehouse_history/models.rs @@ -0,0 +1,114 @@ +//! Check results, their SQL and evidence, and maintenance history as people read them. + +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; +use uuid::Uuid; + +use crate::warehouse_metadata::models::{ + WarehouseAuditRecord, WarehouseJanitorRecord, WarehouseMigrationRecord, WarehouseOldNameRecord, +}; + +/// One stored check result. Its SQL, error texts and evidence rows need their own permissions. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)] +pub struct CheckResult { + pub event_id: Uuid, + /// Identity of the check; a renamed model or changed severity or thresholds starts a new one. + pub binding_key: String, + pub audit_name: String, + pub audit_definition_name: String, + pub audit_description: Option, + pub attachment_kind: String, + pub attached_target_kind: Option, + pub attached_target_name: Option, + pub attached_column_name: Option, + pub severity: String, + pub evaluation_mode: String, + pub run_scope_phase: String, + pub thresholds_json: Option, + pub occurred_at: DateTime, + pub run_id: String, + pub invocation_id: String, + pub outcome: String, + pub violation_count: Option, + pub measured_value: Option, + pub sample_count: Option, + pub sample_unit: Option, + pub minimum_samples: Option, + pub evidence_count: Option, + pub evidence_truncated: Option, + pub reused: String, +} + +impl CheckResult { + pub(crate) fn new(event_id: Uuid, record: WarehouseAuditRecord) -> Self { + Self { + event_id, + binding_key: record.binding_key, + audit_name: record.audit_name, + audit_definition_name: record.audit_definition_name, + audit_description: record.audit_description, + attachment_kind: record.attachment_kind, + attached_target_kind: record.attached_target_kind, + attached_target_name: record.attached_target_name, + attached_column_name: record.attached_column_name, + severity: record.severity, + evaluation_mode: record.evaluation_mode, + run_scope_phase: record.run_scope_phase, + thresholds_json: record.thresholds_json, + occurred_at: record.occurred_at, + run_id: record.run_id, + invocation_id: record.invocation_id, + outcome: record.outcome, + violation_count: record.violation_count, + measured_value: record.measured_value, + sample_count: record.sample_count, + sample_unit: record.sample_unit, + minimum_samples: record.minimum_samples, + evidence_count: record.evidence_count, + evidence_truncated: record.evidence_truncated, + reused: record.reused, + } + } +} + +/// A check's latest SQL and error texts, which quote SQL; `view.sql` reads them, audited. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)] +pub struct CheckSql { + pub event_id: Uuid, + pub binding_key: String, + pub executed_sql: Option, + pub measurement_sql: Option, + pub evidence_sql: Option, + pub execution_error: Option, + pub evidence_error: Option, +} + +/// One result's stored evidence rows; `view.evidence` reads them, audited. +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, utoipa::ToSchema)] +pub struct CheckEvidence { + pub event_id: Uuid, + pub binding_key: String, + /// Every row key, in sorted order; the warehouse stores rows with sorted keys. + pub columns: Vec, + /// Scalar cells aligned to `columns`; a row without a key holds null there. + #[schema(value_type = Vec>)] + pub rows: Vec>, + pub truncated: Option, +} + +/// Recorded moves, their old-name view steps and janitor cleanups, newest moves first. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)] +pub struct MaintenanceHistory { + pub migrations: Vec, + /// Every step recorded for the listed moves, oldest first; same-time steps are not in step order. + pub old_name_views: Vec, + /// Empty when the history is for one model; cleanups are not tied to models. + pub janitor_events: Vec, +} + +#[derive(Clone, Debug, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct MaintenanceQuery { + pub node: Option, + pub limit: Option, +} diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/_helpers/mod.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/_helpers/mod.rs index 6df3ded4b0..35e0fdc180 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_metadata/_helpers/mod.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_metadata/_helpers/mod.rs @@ -1,5 +1,4 @@ pub(crate) mod catalog; -pub(crate) mod checks; pub(crate) mod events; pub(crate) mod fingerprint_reads; pub(crate) mod fingerprints; diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/mod.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/mod.rs index 722d6939c3..4ebb52f84f 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/mod.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/mod.rs @@ -9,10 +9,6 @@ pub mod ingest_source_freshness; pub mod ingest_statistics; pub(crate) mod prune_statistics; pub mod read_person_catalog; -pub mod read_person_check; -pub mod read_person_check_evidence; -pub mod read_person_check_sql; -pub mod read_person_checks; pub mod read_person_fingerprint; pub mod read_person_node_status; pub mod read_person_statistics; diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/models.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/models.rs index 31c94dcb87..61e87d6a6c 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_metadata/models.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_metadata/models.rs @@ -74,94 +74,6 @@ pub struct WarehouseAuditRecord { pub audit_meta_json: Option, } -/// One stored check result. Its SQL, error texts and evidence rows need their own permissions. -#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)] -pub struct CheckResult { - pub event_id: Uuid, - /// Identity of the check; a renamed model or changed severity or thresholds starts a new one. - pub binding_key: String, - pub audit_name: String, - pub audit_definition_name: String, - pub audit_description: Option, - pub attachment_kind: String, - pub attached_target_kind: Option, - pub attached_target_name: Option, - pub attached_column_name: Option, - pub severity: String, - pub evaluation_mode: String, - pub run_scope_phase: String, - pub thresholds_json: Option, - pub occurred_at: DateTime, - pub run_id: String, - pub invocation_id: String, - pub outcome: String, - pub violation_count: Option, - pub measured_value: Option, - pub sample_count: Option, - pub sample_unit: Option, - pub minimum_samples: Option, - pub evidence_count: Option, - pub evidence_truncated: Option, - pub reused: String, -} - -impl CheckResult { - pub(crate) fn new(event_id: Uuid, record: WarehouseAuditRecord) -> Self { - Self { - event_id, - binding_key: record.binding_key, - audit_name: record.audit_name, - audit_definition_name: record.audit_definition_name, - audit_description: record.audit_description, - attachment_kind: record.attachment_kind, - attached_target_kind: record.attached_target_kind, - attached_target_name: record.attached_target_name, - attached_column_name: record.attached_column_name, - severity: record.severity, - evaluation_mode: record.evaluation_mode, - run_scope_phase: record.run_scope_phase, - thresholds_json: record.thresholds_json, - occurred_at: record.occurred_at, - run_id: record.run_id, - invocation_id: record.invocation_id, - outcome: record.outcome, - violation_count: record.violation_count, - measured_value: record.measured_value, - sample_count: record.sample_count, - sample_unit: record.sample_unit, - minimum_samples: record.minimum_samples, - evidence_count: record.evidence_count, - evidence_truncated: record.evidence_truncated, - reused: record.reused, - } - } -} - -/// A check's latest SQL and error texts, which quote SQL; `view.sql` reads them, audited. -#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)] -pub struct CheckSql { - pub event_id: Uuid, - pub binding_key: String, - pub executed_sql: Option, - pub measurement_sql: Option, - pub evidence_sql: Option, - pub execution_error: Option, - pub evidence_error: Option, -} - -/// One result's stored evidence rows; `view.evidence` reads them, audited. -#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, utoipa::ToSchema)] -pub struct CheckEvidence { - pub event_id: Uuid, - pub binding_key: String, - /// Every row key, in sorted order; the warehouse stores rows with sorted keys. - pub columns: Vec, - /// Scalar cells aligned to `columns`; a row without a key holds null there. - #[schema(value_type = Vec>)] - pub rows: Vec>, - pub truncated: Option, -} - /// Warehouse cleanup facts do not authorize control-plane deletion or worker recovery. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)] #[serde(deny_unknown_fields)] diff --git a/crates/sqlbuild-orchestration/tests/api/checks/helpers.rs b/crates/sqlbuild-orchestration/tests/api/checks/helpers.rs deleted file mode 100644 index ca8d39c2d2..0000000000 --- a/crates/sqlbuild-orchestration/tests/api/checks/helpers.rs +++ /dev/null @@ -1,68 +0,0 @@ -//! Password sessions, fixture loading and audit reads for the check API tests. - -use reqwest::{Client, StatusCode, header}; -use serde_json::{Value, json}; - -use crate::store::main::{begin::begin, commit::commit, query::query}; -use crate::store::models::{ControlStore, TransactionMode}; - -pub(crate) const ORIGIN: &str = "https://control.example.com"; -pub(crate) const PASSWORD: &str = "orders-password-2026"; - -pub(crate) fn fixture(name: &str) -> Value { - let path = std::path::Path::new(env!("CARGO_MANIFEST_DIR")) - .join("../../benchmarks/duckshop/edge_cases/orchestration") - .join(name); - serde_json::from_str(&std::fs::read_to_string(path).expect("fixture")).expect("fixture JSON") -} - -pub(crate) async fn get( - client: &Client, - (url, cookie): (&str, &str), - path: &str, -) -> (StatusCode, Value) { - let response = client - .get(format!("{url}{path}")) - .header(header::HOST, "control.example.com") - .header(header::ORIGIN, ORIGIN) - .header("cookie", cookie) - .send() - .await - .expect("check read"); - let status = response.status(); - (status, response.json().await.unwrap_or(Value::Null)) -} - -pub(crate) async fn login(client: &Client, url: &str, username: &str) -> String { - let response = client - .post(format!("{url}/api/auth/password/login")) - .header(header::HOST, "control.example.com") - .header(header::ORIGIN, ORIGIN) - .json(&json!({"username": username, "password": PASSWORD})) - .send() - .await - .expect("password login"); - assert_eq!(response.status(), StatusCode::OK); - response - .headers() - .get_all(header::SET_COOKIE) - .iter() - .map(|value| value.to_str().expect("cookie")) - .find(|value| value.starts_with("__Host-sqb-session=")) - .expect("browser session") - .split(';') - .next() - .expect("cookie pair") - .to_owned() -} - -pub(crate) async fn latest_audit(store: &ControlStore) -> Value { - let transaction = begin(store, TransactionMode::Read) - .await - .expect("audit cut"); - let (transaction, rows) = query(transaction, - "SELECT action,granting_role,scope_json,after_json FROM sqb_security_audit ORDER BY storage_id DESC LIMIT 1", - vec![]).await.expect("audit"); - commit(transaction).await.expect("close audit cut"); - json!(rows[0]) -} diff --git a/crates/sqlbuild-orchestration/tests/api/test_checks.rs b/crates/sqlbuild-orchestration/tests/api/test_checks.rs index 24e1fc9acc..46e3c3bf7a 100644 --- a/crates/sqlbuild-orchestration/tests/api/test_checks.rs +++ b/crates/sqlbuild-orchestration/tests/api/test_checks.rs @@ -1,35 +1,19 @@ //! Check reads through the public API: keyed by binding key, view-gated, with audited SQL. -use reqwest::{Client, StatusCode}; +use reqwest::StatusCode; use serde_json::{Value, json}; use uuid::Uuid; -use crate::api::models::{ApiAuthentication, HumanLogin}; -use crate::api::tests::checks_helpers::{ORIGIN, PASSWORD, fixture, get, latest_audit, login}; -use crate::api::tests::helpers::{start_api_with_authentication, stop_api}; +use crate::api::tests::helpers::stop_api; use crate::api::tests::test_types::WarehouseMetadataTestCase; -use crate::authorization::main::{ - assign_builtin_role::assign_builtin_role, bootstrap_user::bootstrap_user, +use crate::api::tests::warehouse_history_helpers::{ + batch, fixture, get, latest_audit, login, seed_target, start_password_api, }; -use crate::credentials::main::{ - identities::create_principal::create_principal, sessions::login_password::login_password, - sessions::set_password::set_password, -}; -use crate::credentials::models::{ - AuditContext, BuiltinRole, PasswordAuthority, PasswordLogin, RoleAssignmentRequest, -}; -use crate::credentials::tests::location_helpers::create_location; -use crate::platform::main::report_location_health::report_location_health; -use crate::platform::models::{LocationHealth, LocationHealthReport}; -use crate::runs::tests::helpers::snapshot; -use crate::snapshots::main::publish_snapshot::publish_snapshot; -use crate::store::main::migrate_store::migrate_store; use crate::store::tests::fault_helpers::{install_fault, remove_fault}; use crate::store::tests::fault_types::TransactionFault; use crate::store::tests::store_helpers::{close_test_store, open_test_store, run}; use crate::store::tests::store_test_types::Backend; use crate::warehouse_metadata::main::ingest_audit_results::ingest_audit_results; -use crate::warehouse_metadata::models::MetadataBatch; #[test] fn given_checks_sharing_an_audit_name_when_people_read_them_then_binding_keys_and_sql_permissions_hold() @@ -50,79 +34,7 @@ fn given_checks_sharing_an_audit_name_when_people_read_them_then_binding_keys_an run(async { let fixture_store = open_test_store(test_case.backend).await.expect("store"); let store = &fixture_store.store; - migrate_store(store).await.expect("migrate"); - let context = AuditContext { - actor: "control-store-owner".into(), - granting_role: "control-store-owner".into(), - request_id: Uuid::now_v7(), - ip_address: Some("127.0.0.1".parse().expect("local peer")), - }; - let admin = bootstrap_user(store, &context, "orders-admin") - .await - .expect("bootstrap"); - let PasswordLogin::Authenticated(admin_session) = ({ - set_password( - store, - &context, - PasswordAuthority::Operator(admin), - PASSWORD, - ) - .await - .expect("admin password"); - login_password(store, &context, ("orders-admin", PASSWORD)) - .await - .expect("admin login") - }) else { - panic!("authenticated administrator"); - }; - let location = create_location(store, "orders", "test-admin") - .await - .expect("location"); - for (name, role) in [ - ("orders-viewer", BuiltinRole::Viewer), - ("orders-developer", BuiltinRole::Developer), - ] { - let person = create_principal(store, &context, name) - .await - .expect("person"); - set_password( - store, - &context, - PasswordAuthority::Operator(person), - PASSWORD, - ) - .await - .expect("password"); - assign_builtin_role( - store, - (&context, admin_session.identity.principal_id), - &RoleAssignmentRequest { - principal_id: person, - role: role.into(), - location_id: Some(location), - target: Some("development".into()), - revoke: false, - }, - ) - .await - .expect("scoped role"); - } - publish_snapshot(store, &snapshot(location)) - .await - .expect("snapshot"); - let session_id = Uuid::now_v7(); - report_location_health( - store, - location, - &LocationHealthReport { - event_id: Uuid::now_v7(), - session_id, - target: "development".into(), - health: LocationHealth::Heartbeat {}, - }, - ) - .await - .expect("reader"); + let (location, session_id) = seed_target(store).await; let data = fixture("check_identity.json"); let base = fixture("audit_result_history.json")["record"].clone(); let records: Vec = data["results"] @@ -146,24 +58,18 @@ fn given_checks_sharing_an_audit_name_when_people_read_them_then_binding_keys_an }) .collect(); assert_eq!(records.len(), test_case.expected_observations); - let batch: MetadataBatch = serde_json::from_value(json!({"kind":"audit_results","session_id":session_id,"generation":1,"records":records, - "scan":{"scan_id":Uuid::now_v7(),"location":{"database":null,"schema":"main"},"complete":false}})).expect("batch"); assert!( - ingest_audit_results(store, (location, "development"), &batch) - .await - .expect("real audit ingestion") - .applied + ingest_audit_results( + store, + (location, "development"), + &batch(session_id, "audit_results", records) + ) + .await + .expect("real audit ingestion") + .applied ); - let api = start_api_with_authentication( - store, - ApiAuthentication { - login: HumanLogin::Password(ORIGIN.into()), - ..Default::default() - }, - ) - .await; - let client = Client::new(); + let (api, client) = start_password_api(store).await; let viewer = login(&client, &api.url, "orders-viewer").await; let developer = login(&client, &api.url, "orders-developer").await; let administrator = login(&client, &api.url, "orders-admin").await; diff --git a/crates/sqlbuild-orchestration/tests/api/test_contracts.rs b/crates/sqlbuild-orchestration/tests/api/test_contracts.rs index e9878e27e2..25527b6ad0 100644 --- a/crates/sqlbuild-orchestration/tests/api/test_contracts.rs +++ b/crates/sqlbuild-orchestration/tests/api/test_contracts.rs @@ -2,8 +2,6 @@ #[path = "changes/helpers.rs"] mod changes_helpers; -#[path = "checks/helpers.rs"] -pub(crate) mod checks_helpers; #[path = "fingerprint_reads/helpers.rs"] mod fingerprint_reads_helpers; #[path = "helpers.rs"] @@ -50,6 +48,8 @@ mod test_external_sessions; mod test_location_health; #[path = "test_location_work.rs"] mod test_location_work; +#[path = "test_maintenance.rs"] +mod test_maintenance; #[path = "test_oidc.rs"] mod test_oidc; #[path = "test_oidc_groups.rs"] @@ -76,6 +76,8 @@ mod test_worker; mod warehouse_catalog_helpers; #[path = "warehouse_metadata/events/helpers.rs"] mod warehouse_events_helpers; +#[path = "warehouse_history/helpers.rs"] +pub(crate) mod warehouse_history_helpers; #[path = "warehouse_metadata/helpers.rs"] mod warehouse_metadata_helpers; #[path = "warehouse_metadata/reconciliation/helpers.rs"] diff --git a/crates/sqlbuild-orchestration/tests/api/test_maintenance.rs b/crates/sqlbuild-orchestration/tests/api/test_maintenance.rs new file mode 100644 index 0000000000..c19db6b760 --- /dev/null +++ b/crates/sqlbuild-orchestration/tests/api/test_maintenance.rs @@ -0,0 +1,162 @@ +//! Maintenance history through the public API: moves, their old-name view steps and cleanups. + +use reqwest::StatusCode; +use serde_json::{Value, json}; +use uuid::Uuid; + +use crate::api::tests::helpers::stop_api; +use crate::api::tests::test_types::WarehouseMetadataTestCase; +use crate::api::tests::warehouse_history_helpers::{ + batch, fixture, get, ids, login, records, seed_target, start_password_api, +}; +use crate::platform::main::report_location_health::report_location_health; +use crate::platform::models::{LocationHealth, LocationHealthReport}; +use crate::runs::tests::helpers::snapshot; +use crate::snapshots::main::publish_snapshot::publish_snapshot; +use crate::store::tests::store_helpers::{close_test_store, open_test_store, run}; +use crate::store::tests::store_test_types::Backend; +use crate::warehouse_metadata::main::{ + ingest_janitor_events::ingest_janitor_events, ingest_migrations::ingest_migrations, + ingest_old_name_views::ingest_old_name_views, +}; + +#[test] +fn given_moves_views_and_cleanups_when_people_read_maintenance_then_each_model_sees_its_own_history() + { + let test_cases = [ + WarehouseMetadataTestCase { + description: "SQLite maintenance reads", + backend: Backend::Sqlite, + expected_observations: 9, + }, + WarehouseMetadataTestCase { + description: "PostgreSQL maintenance reads", + backend: Backend::Postgres, + expected_observations: 9, + }, + ]; + for test_case in test_cases { + run(async { + let fixture_store = open_test_store(test_case.backend).await.expect("store"); + let store = &fixture_store.store; + let (location, session_id) = seed_target(store).await; + let data = fixture("maintenance_history.json"); + let scope = (location, "development"); + let moves = records("migration_history.json", &data["migrations"]); + let views = records("old_name_history.json", &data["old_name_views"]); + let cleanups = records("janitor_event_history.json", &data["janitor_events"]); + assert_eq!( + moves.len() + views.len() + cleanups.len(), + test_case.expected_observations + ); + ingest_migrations(store, scope, &batch(session_id, "migrations", moves)) + .await + .expect("real migration ingestion"); + ingest_old_name_views(store, scope, &batch(session_id, "old_name_views", views)) + .await + .expect("real old-name ingestion"); + ingest_janitor_events(store, scope, &batch(session_id, "janitor_events", cleanups)) + .await + .expect("real janitor ingestion"); + + let mut preview = snapshot(location); + preview.target = "preview".into(); + publish_snapshot(store, &preview) + .await + .expect("preview snapshot"); + let preview_session = Uuid::now_v7(); + report_location_health( + store, + location, + &LocationHealthReport { + event_id: Uuid::now_v7(), + session_id: preview_session, + target: "preview".into(), + health: LocationHealth::Heartbeat {}, + }, + ) + .await + .expect("preview reader"); + ingest_old_name_views( + store, + (location, "preview"), + &batch( + preview_session, + "old_name_views", + records("old_name_history.json", &data["preview_old_name_views"]), + ), + ) + .await + .expect("another target's view"); + + let (api, client) = start_password_api(store).await; + let viewer = login(&client, &api.url, "orders-viewer").await; + let path = format!("/api/control/locations/{location}/targets/development/maintenance"); + + let (status, all) = get(&client, (&api.url, &viewer), &path).await; + assert_eq!(status, StatusCode::OK, "{}", test_case.description); + assert_eq!( + ids(&all, "migrations"), + ["move-orders-2", "move-customers-1", "move-orders-1"] + ); + assert_eq!( + ids(&all, "old_name_views"), + [ + "view-orders-old-1", + "view-customers-1", + "view-orders-1", + "view-orders-2" + ], + "another target's steps never appear" + ); + assert_eq!(ids(&all, "janitor_events"), ["cleanup-2", "cleanup-1"]); + assert_eq!(all["migrations"][0]["origin_model"], Value::Null); + assert_eq!(all["migrations"][0]["discovery"], "automatic"); + + let (_, orders) = + get(&client, (&api.url, &viewer), &format!("{path}?node=orders")).await; + assert_eq!( + ids(&orders, "migrations"), + ["move-orders-2", "move-orders-1"] + ); + assert_eq!( + ids(&orders, "old_name_views"), + ["view-orders-old-1", "view-orders-1", "view-orders-2"], + "another model's views never appear" + ); + assert_eq!(orders["janitor_events"], json!([])); + + let (_, newest) = get(&client, (&api.url, &viewer), &format!("{path}?limit=1")).await; + assert_eq!(ids(&newest, "migrations"), ["move-orders-2"]); + assert_eq!( + ids(&newest, "old_name_views"), + ["view-orders-1", "view-orders-2"], + "only the listed moves' steps, not every step of their models" + ); + assert_eq!(ids(&newest, "janitor_events"), ["cleanup-2"]); + + let (_, unknown) = get( + &client, + (&api.url, &viewer), + &format!("{path}?node=unknown"), + ) + .await; + assert_eq!( + unknown, + json!({"migrations": [], "old_name_views": [], "janitor_events": []}) + ); + assert_eq!( + get( + &client, + (&api.url, &viewer), + &path.replace("development", "preview") + ) + .await + .0, + StatusCode::FORBIDDEN + ); + stop_api(api).await; + close_test_store(fixture_store).await; + }); + } +} diff --git a/crates/sqlbuild-orchestration/tests/api/test_worker.rs b/crates/sqlbuild-orchestration/tests/api/test_worker.rs index 0d21031a7c..ffc06aa6ce 100644 --- a/crates/sqlbuild-orchestration/tests/api/test_worker.rs +++ b/crates/sqlbuild-orchestration/tests/api/test_worker.rs @@ -8,7 +8,7 @@ use crate::runs::main::list_runs::list_runs; use crate::runs::models::{ NodeOutcome, RunAuthority, RunChange, RunChangeRequest, RunOutcome, RunStatus, }; -use crate::scheduler::tests::contracts::helpers::acquire_lease; +use crate::scheduler::tests::contracts::helpers::{acquire_lease, hold_lease}; use crate::snapshots::main::publish_snapshot::publish_snapshot; use crate::store::errors::ControlError; use crate::store::main::migrate_store::migrate_store; @@ -68,6 +68,7 @@ fn given_a_claimed_run_when_a_worker_connects_then_verification_precedes_start_a .await .expect("acquire") .expect("leader"); + let holder = hold_lease(store, &lease); let claim = claim_next_run(store, &lease, &BTreeMap::new(), None) .await .expect("claim") @@ -329,6 +330,7 @@ fn given_a_claimed_run_when_a_worker_connects_then_verification_precedes_start_a "{}", test_case.description ); + holder.abort(); drop(session); stop_api(api).await; close_test_store(fixture).await; diff --git a/crates/sqlbuild-orchestration/tests/api/warehouse_history/helpers.rs b/crates/sqlbuild-orchestration/tests/api/warehouse_history/helpers.rs new file mode 100644 index 0000000000..b7679b1b7d --- /dev/null +++ b/crates/sqlbuild-orchestration/tests/api/warehouse_history/helpers.rs @@ -0,0 +1,214 @@ +//! A seeded target with scoped people, password sessions, fixtures and audit reads for warehouse history API tests. + +use reqwest::{Client, StatusCode, header}; +use serde_json::{Value, json}; +use uuid::Uuid; + +use crate::api::models::{ApiAuthentication, HumanLogin}; +use crate::api::tests::helpers::start_api_with_authentication; +use crate::api::tests::test_types::TestApi; +use crate::authorization::main::{ + assign_builtin_role::assign_builtin_role, bootstrap_user::bootstrap_user, +}; +use crate::credentials::main::{ + identities::create_principal::create_principal, sessions::login_password::login_password, + sessions::set_password::set_password, +}; +use crate::credentials::models::{ + AuditContext, BuiltinRole, PasswordAuthority, PasswordLogin, RoleAssignmentRequest, +}; +use crate::credentials::tests::location_helpers::create_location; +use crate::platform::main::report_location_health::report_location_health; +use crate::platform::models::{LocationHealth, LocationHealthReport}; +use crate::runs::tests::helpers::snapshot; +use crate::snapshots::main::publish_snapshot::publish_snapshot; +use crate::store::main::{ + begin::begin, commit::commit, migrate_store::migrate_store, query::query, +}; +use crate::store::models::{ControlStore, TransactionMode}; +use crate::warehouse_metadata::models::MetadataBatch; + +pub(crate) const ORIGIN: &str = "https://control.example.com"; +pub(crate) const PASSWORD: &str = "orders-password-2026"; + +pub(crate) fn fixture(name: &str) -> Value { + let path = std::path::Path::new(env!("CARGO_MANIFEST_DIR")) + .join("../../benchmarks/duckshop/edge_cases/orchestration") + .join(name); + serde_json::from_str(&std::fs::read_to_string(path).expect("fixture")).expect("fixture JSON") +} + +pub(crate) async fn get( + client: &Client, + (url, cookie): (&str, &str), + path: &str, +) -> (StatusCode, Value) { + let response = client + .get(format!("{url}{path}")) + .header(header::HOST, "control.example.com") + .header(header::ORIGIN, ORIGIN) + .header("cookie", cookie) + .send() + .await + .expect("check read"); + let status = response.status(); + (status, response.json().await.unwrap_or(Value::Null)) +} + +pub(crate) async fn login(client: &Client, url: &str, username: &str) -> String { + let response = client + .post(format!("{url}/api/auth/password/login")) + .header(header::HOST, "control.example.com") + .header(header::ORIGIN, ORIGIN) + .json(&json!({"username": username, "password": PASSWORD})) + .send() + .await + .expect("password login"); + assert_eq!(response.status(), StatusCode::OK); + response + .headers() + .get_all(header::SET_COOKIE) + .iter() + .map(|value| value.to_str().expect("cookie")) + .find(|value| value.starts_with("__Host-sqb-session=")) + .expect("browser session") + .split(';') + .next() + .expect("cookie pair") + .to_owned() +} + +pub(crate) async fn latest_audit(store: &ControlStore) -> Value { + let transaction = begin(store, TransactionMode::Read) + .await + .expect("audit cut"); + let (transaction, rows) = query(transaction, + "SELECT action,granting_role,scope_json,after_json FROM sqb_security_audit ORDER BY storage_id DESC LIMIT 1", + vec![]).await.expect("audit"); + commit(transaction).await.expect("close audit cut"); + json!(rows[0]) +} + +/// A migrated store with scoped people, the orders snapshot and a live metadata reader session. +pub(crate) async fn seed_target(store: &ControlStore) -> (Uuid, Uuid) { + migrate_store(store).await.expect("migrate"); + let context = AuditContext { + actor: "control-store-owner".into(), + granting_role: "control-store-owner".into(), + request_id: Uuid::now_v7(), + ip_address: Some("127.0.0.1".parse().expect("local peer")), + }; + let admin = bootstrap_user(store, &context, "orders-admin") + .await + .expect("bootstrap"); + let PasswordLogin::Authenticated(admin_session) = ({ + set_password( + store, + &context, + PasswordAuthority::Operator(admin), + PASSWORD, + ) + .await + .expect("admin password"); + login_password(store, &context, ("orders-admin", PASSWORD)) + .await + .expect("admin login") + }) else { + panic!("authenticated administrator"); + }; + let location = create_location(store, "orders", "test-admin") + .await + .expect("location"); + for (name, role) in [ + ("orders-viewer", BuiltinRole::Viewer), + ("orders-developer", BuiltinRole::Developer), + ] { + let person = create_principal(store, &context, name) + .await + .expect("person"); + set_password( + store, + &context, + PasswordAuthority::Operator(person), + PASSWORD, + ) + .await + .expect("password"); + assign_builtin_role( + store, + (&context, admin_session.identity.principal_id), + &RoleAssignmentRequest { + principal_id: person, + role: role.into(), + location_id: Some(location), + target: Some("development".into()), + revoke: false, + }, + ) + .await + .expect("scoped role"); + } + publish_snapshot(store, &snapshot(location)) + .await + .expect("snapshot"); + let session_id = Uuid::now_v7(); + report_location_health( + store, + location, + &LocationHealthReport { + event_id: Uuid::now_v7(), + session_id, + target: "development".into(), + health: LocationHealth::Heartbeat {}, + }, + ) + .await + .expect("reader"); + (location, session_id) +} + +/// One metadata batch of `kind` from the reader session, scanning the `main` state schema. +pub(crate) fn batch(session_id: Uuid, kind: &str, records: Vec) -> MetadataBatch { + serde_json::from_value(json!({"kind":kind,"session_id":session_id,"generation":1,"records":records, + "scan":{"scan_id":Uuid::now_v7(),"location":{"database":null,"schema":"main"},"complete":false}})).expect("batch") +} + +/// An API with password sign-in and a client for it. +pub(crate) async fn start_password_api(store: &ControlStore) -> (TestApi, Client) { + let api = start_api_with_authentication( + store, + ApiAuthentication { + login: HumanLogin::Password(ORIGIN.into()), + ..Default::default() + }, + ) + .await; + (api, Client::new()) +} + +/// Fixture records of `base` with each override applied. +pub(crate) fn records(base: &str, overrides: &Value) -> Vec { + let base = fixture(base)["record"].clone(); + overrides + .as_array() + .expect("overrides") + .iter() + .map(|values| { + let mut record = base.clone(); + for (key, value) in values.as_object().expect("override") { + record[key] = value.clone(); + } + record + }) + .collect() +} + +/// The event IDs of one kind of record in a history response. +pub(crate) fn ids(history: &Value, kind: &str) -> Vec { + history[kind] + .as_array() + .expect("records") + .iter() + .map(|record| record["event_id"].as_str().expect("event").to_owned()) + .collect() +} diff --git a/fensu.toml b/fensu.toml index 146aa9b37e..b36a40292e 100644 --- a/fensu.toml +++ b/fensu.toml @@ -1367,11 +1367,11 @@ path = "src/sqlbuild/adapters/sqlserver/classes/sqlserver_adapter.py" reason = "The SQL Server adapter intentionally keeps overrideable backend behavior on one class." [[dupes.allowlist]] -paths = ["crates/sqlbuild-orchestration/src/api/_helpers/resources/administrators.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/audit.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/location_health.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/location_work.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/person_checks.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/person_location_work.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/person_node_sql.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/person_operations.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/run_reads.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/runs.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/snapshots.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/trigger_reads.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/warehouse_metadata.rs"] +paths = ["crates/sqlbuild-orchestration/src/api/_helpers/resources/administrators.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/audit.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/location_health.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/location_work.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/person_checks.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/person_location_work.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/person_maintenance.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/person_node_sql.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/person_operations.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/run_reads.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/runs.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/snapshots.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/trigger_reads.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/warehouse_metadata.rs"] reason = "Each function is one utoipa-annotated axum route: OpenAPI needs one annotated handler per operation, and each body only checks the caller's scope and calls one store operation. Scope checks and store work live in their shared owners." [[dupes.allowlist]] -paths = ["crates/sqlbuild-orchestration/src/authorization/main/read_security_audit.rs", "crates/sqlbuild-orchestration/src/credentials/_helpers/active_session.rs", "crates/sqlbuild-orchestration/src/declarations/main/list_triggers.rs", "crates/sqlbuild-orchestration/src/platform/main/read_location_work.rs", "crates/sqlbuild-orchestration/src/runs/main/list_person_runs.rs", "crates/sqlbuild-orchestration/src/runs/main/list_runs.rs", "crates/sqlbuild-orchestration/src/runs/_helpers/lifecycle/reads.rs", "crates/sqlbuild-orchestration/src/warehouse_metadata/_helpers/checks.rs", "crates/sqlbuild-orchestration/src/scheduler/main/list_ticks.rs", "crates/sqlbuild-orchestration/src/sensors/_helpers/attempts.rs", "crates/sqlbuild-orchestration/src/sensors/_helpers/facts.rs", "crates/sqlbuild-orchestration/src/snapshots/main/read_current_record.rs", "crates/sqlbuild-orchestration/src/snapshots/main/read_snapshot_documents.rs", "crates/sqlbuild-orchestration/src/snapshots/main/read_snapshot_nodes.rs", "crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_catalog.rs", "crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_check.rs", "crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_checks.rs", "crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_statistics.rs"] +paths = ["crates/sqlbuild-orchestration/src/authorization/main/read_security_audit.rs", "crates/sqlbuild-orchestration/src/credentials/_helpers/active_session.rs", "crates/sqlbuild-orchestration/src/declarations/main/list_triggers.rs", "crates/sqlbuild-orchestration/src/platform/main/read_location_work.rs", "crates/sqlbuild-orchestration/src/runs/main/list_person_runs.rs", "crates/sqlbuild-orchestration/src/runs/main/list_runs.rs", "crates/sqlbuild-orchestration/src/runs/_helpers/lifecycle/reads.rs", "crates/sqlbuild-orchestration/src/warehouse_history/_helpers/checks.rs", "crates/sqlbuild-orchestration/src/scheduler/main/list_ticks.rs", "crates/sqlbuild-orchestration/src/sensors/_helpers/attempts.rs", "crates/sqlbuild-orchestration/src/sensors/_helpers/facts.rs", "crates/sqlbuild-orchestration/src/snapshots/main/read_current_record.rs", "crates/sqlbuild-orchestration/src/snapshots/main/read_snapshot_documents.rs", "crates/sqlbuild-orchestration/src/snapshots/main/read_snapshot_nodes.rs", "crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_catalog.rs", "crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_check.rs", "crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_checks.rs", "crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_maintenance.rs", "crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_statistics.rs"] reason = "Each is one control-store read with its own SQL and record type. Transactions, person authorization and decoding go through the shared begin_authorized, query_records, read_records and query_bodies helpers; what remains alike is the call shape." [[dupes.allowlist]] @@ -1632,8 +1632,6 @@ allowed = [ "sqb_warehouse_microbatches.execution_run_id=Reader comes with the UI epic (D132), which filters on it: the run that executed the batch.", "sqb_warehouse_microbatches.record_type=Reader comes with the UI epic (D132), which filters on it: the batch state.", "sqb_warehouse_migrations.decision=Reader comes with the UI epic (D132), which filters on it.", - "sqb_warehouse_migrations.destination_model=Reader comes with the UI epic (D132), which filters on it.", - "sqb_warehouse_old_name_views.migration_event_id=Reader comes with the UI epic (D132), which filters on it.", ] [targets.rust.rule_options.XSB077]