diff --git a/clients/typescript/src/schema.d.ts b/clients/typescript/src/schema.d.ts index 7c24125d11..8304565a96 100644 --- a/clients/typescript/src/schema.d.ts +++ b/clients/typescript/src/schema.d.ts @@ -354,6 +354,22 @@ export interface paths { patch?: never; trace?: never; }; + "/api/control/locations/{location_id}/targets/{target}/jobs": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + get: operations["listPersonJobs"]; + put?: never; + post?: never; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; "/api/control/locations/{location_id}/targets/{target}/maintenance": { parameters: { query?: never; @@ -1134,6 +1150,17 @@ export interface components { failed: components["schemas"]["Table"][]; failure?: string | null; }; + /** @description A job or group of the current published snapshot. Settings are left out of person reads. */ + JobRecord: { + concurrency_groups: string[]; + description: string; + job_key: string; + /** @description `job` or `group`; a group's selection is the union of its member jobs' selections. */ + kind: components["schemas"]["RunKind"]; + selection: string[]; + /** Format: uuid */ + snapshot_id: string; + }; /** @description A trusted operator names a published job; execution inputs are never supplied by the caller. */ JobRunRequest: { job_key: string; @@ -2885,6 +2912,43 @@ export interface operations { }; }; }; + listPersonJobs: { + parameters: { + query?: never; + header?: never; + path: { + location_id: string; + target: string; + }; + cookie?: never; + }; + requestBody?: never; + responses: { + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["JobRecord"][]; + }; + }; + /** @description Session rejected */ + 401: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + 403: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["ApiErrorBody"]; + }; + }; + }; + }; readPersonMaintenance: { 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 35f6359639..be80c63098 100644 --- a/crates/sqlbuild-orchestration/src/api/_helpers/authentication/password_sessions.rs +++ b/crates/sqlbuild-orchestration/src/api/_helpers/authentication/password_sessions.rs @@ -101,6 +101,10 @@ pub(crate) fn person_routes() -> Router { "/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}/jobs", + get(crate::api::_helpers::resources::person_reads::jobs), + ) .route( "/api/control/locations/{location_id}/targets/{target}/triggers", get(crate::api::_helpers::resources::person_reads::triggers), diff --git a/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_reads.rs b/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_reads.rs index cd4392dea9..3a7d8f6d34 100644 --- a/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_reads.rs +++ b/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_reads.rs @@ -1,4 +1,4 @@ -//! Person reads of run nodes, triggers and ticks; refusal precedes every lookup, so unknown objects look forbidden. +//! Person reads of run nodes, jobs, triggers and ticks; refusal precedes every lookup, so unknown objects look forbidden. use axum::extract::{Path, Query, State}; use axum::http::StatusCode; @@ -8,8 +8,9 @@ use uuid::Uuid; use crate::api::errors::ApiError; use crate::api::models::ReadQuery; use crate::credentials::models::{AuditContext, AuthenticatedPerson, DefaultRole}; +use crate::declarations::main::list_person_jobs::list_person_jobs; use crate::declarations::main::list_person_triggers::list_person_triggers; -use crate::declarations::models::TriggerRecord; +use crate::declarations::models::{JobRecord, TriggerRecord}; 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; @@ -39,6 +40,27 @@ 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}/jobs", operation_id = "listPersonJobs", + params(("location_id" = Uuid, Path), ("target" = String, Path)), + responses((status = 200, body = Vec), (status = 401, description = "Session rejected"), + (status = 403, body = crate::api::models::ApiErrorBody)), security(("session" = []), ("cliSession" = [])))] +pub(crate) async fn jobs( + (State(store), Extension(context)): (State, Extension), + Extension(default_role): Extension, + Path((location_id, target)): Path<(Uuid, String)>, + Extension(AuthenticatedPerson(principal, _)): Extension, +) -> Result>, ApiError> { + Ok(Json( + list_person_jobs( + &store, + (&context, principal), + (location_id, &target), + default_role, + ) + .await?, + )) +} + #[utoipa::path(get, path = "/api/control/locations/{location_id}/targets/{target}/triggers", operation_id = "listPersonTriggers", params(("location_id" = Uuid, Path), ("target" = String, Path)), responses((status = 200, body = Vec), (status = 401, description = "Session rejected"), diff --git a/crates/sqlbuild-orchestration/src/api/models.rs b/crates/sqlbuild-orchestration/src/api/models.rs index 17dc9477f4..d52b2bac13 100644 --- a/crates/sqlbuild-orchestration/src/api/models.rs +++ b/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::jobs, person_reads::triggers, person_reads::ticks, person_operations::catalog, diff --git a/crates/sqlbuild-orchestration/src/declarations/_helpers/reads.rs b/crates/sqlbuild-orchestration/src/declarations/_helpers/reads.rs index a69248fd01..00f3bf5150 100644 --- a/crates/sqlbuild-orchestration/src/declarations/_helpers/reads.rs +++ b/crates/sqlbuild-orchestration/src/declarations/_helpers/reads.rs @@ -1,8 +1,8 @@ -//! Trigger reads inside a caller's transaction, so authorization and results share one view. +//! Job and trigger reads inside a caller's transaction, so authorization and results share one view. use uuid::Uuid; -use crate::declarations::models::TriggerRecord; +use crate::declarations::models::{JobRecord, StoredJob, TriggerRecord}; use crate::store::errors::ControlError; use crate::store::main::query_records::query_records; use crate::store::models::{ControlTransaction, Parameter}; @@ -17,3 +17,28 @@ pub(crate) async fn triggers_in( WHERE t.location_id=$1 AND t.target=$2 AND t.removed_in_snapshot_id IS NULL ORDER BY t.name", vec![Parameter::Id(location_id), Parameter::Text(target.into())]).await } + +pub(crate) async fn jobs_in( + transaction: ControlTransaction, + (location_id, target): (Uuid, &str), +) -> Result<(ControlTransaction, Vec), ControlError> { + let (transaction, stored): (_, Vec) = query_records( + transaction, + "SELECT job_key, kind, snapshot_id, definition_json FROM sqb_jobs \ + WHERE location_id=$1 AND target=$2 ORDER BY job_key", + vec![Parameter::Id(location_id), Parameter::Text(target.into())], + ) + .await?; + let jobs: Vec = stored + .into_iter() + .map(|job| JobRecord { + job_key: job.job_key, + kind: job.kind, + description: job.definition_json.description, + selection: job.definition_json.selection, + concurrency_groups: job.definition_json.concurrency_groups, + snapshot_id: job.snapshot_id, + }) + .collect(); + Ok((transaction, jobs)) +} diff --git a/crates/sqlbuild-orchestration/src/declarations/main/list_person_jobs.rs b/crates/sqlbuild-orchestration/src/declarations/main/list_person_jobs.rs new file mode 100644 index 0000000000..8675f3e0d0 --- /dev/null +++ b/crates/sqlbuild-orchestration/src/declarations/main/list_person_jobs.rs @@ -0,0 +1,29 @@ +//! A person lists jobs only with view on every current node of the target, like triggers. + +use uuid::Uuid; + +use crate::authorization::main::authorize_listing::authorize_listing; +use crate::credentials::models::{AuditContext, DefaultRole}; +use crate::declarations::_helpers::reads::jobs_in; +use crate::declarations::models::JobRecord; +use crate::store::errors::ControlError; +use crate::store::main::{begin::begin, commit::commit}; +use crate::store::models::{ControlStore, TransactionMode}; + +pub async fn list_person_jobs( + store: &ControlStore, + (context, credentials): (&AuditContext, Uuid), + (location_id, target): (Uuid, &str), + default_role: DefaultRole, +) -> Result, ControlError> { + let (transaction, _) = authorize_listing( + begin(store, TransactionMode::Write).await?, + (context, credentials), + (location_id, target), + default_role, + ) + .await?; + let (transaction, jobs) = jobs_in(transaction, (location_id, target)).await?; + commit(transaction).await?; + Ok(jobs) +} diff --git a/crates/sqlbuild-orchestration/src/declarations/main/mod.rs b/crates/sqlbuild-orchestration/src/declarations/main/mod.rs index 1a14985c37..0729a456c1 100644 --- a/crates/sqlbuild-orchestration/src/declarations/main/mod.rs +++ b/crates/sqlbuild-orchestration/src/declarations/main/mod.rs @@ -2,6 +2,7 @@ pub mod change_person_trigger; pub mod change_trigger; pub(crate) mod frozen_definitions; pub mod group_dependencies; +pub mod list_person_jobs; pub mod list_person_triggers; pub mod list_trigger_events; pub mod list_triggers; diff --git a/crates/sqlbuild-orchestration/src/declarations/models.rs b/crates/sqlbuild-orchestration/src/declarations/models.rs index afdcab28d0..726975e856 100644 --- a/crates/sqlbuild-orchestration/src/declarations/models.rs +++ b/crates/sqlbuild-orchestration/src/declarations/models.rs @@ -87,6 +87,27 @@ pub enum TriggerDeclaration { Sensor(SensorDefinition), } +/// A job or group of the current published snapshot. Settings are left out of person reads. +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, utoipa::ToSchema)] +pub struct JobRecord { + pub job_key: String, + /// `job` or `group`; a group's selection is the union of its member jobs' selections. + pub kind: crate::runs::models::RunKind, + pub description: String, + pub selection: BTreeSet, + pub concurrency_groups: BTreeSet, + pub snapshot_id: uuid::Uuid, +} + +/// A `sqb_jobs` row as stored. +#[derive(Clone, Debug, Deserialize)] +pub(crate) struct StoredJob { + pub job_key: String, + pub kind: crate::runs::models::RunKind, + pub snapshot_id: uuid::Uuid, + pub definition_json: JobDefinition, +} + #[derive(Clone, Debug, PartialEq, Serialize, Deserialize, utoipa::ToSchema)] pub struct TriggerRecord { pub trigger_id: uuid::Uuid, diff --git a/crates/sqlbuild-orchestration/tests/api/test_denials.rs b/crates/sqlbuild-orchestration/tests/api/test_denials.rs index 7f173d7916..439f159c7a 100644 --- a/crates/sqlbuild-orchestration/tests/api/test_denials.rs +++ b/crates/sqlbuild-orchestration/tests/api/test_denials.rs @@ -20,8 +20,9 @@ use crate::store::tests::store_helpers::{close_test_store, open_test_store, run} use crate::store::tests::store_test_types::Backend; /// Location-scoped reads and the action each needs. -const READS: [(&str, &str); 10] = [ +const READS: [(&str, &str); 11] = [ ("runs", "view"), + ("jobs", "view"), ("triggers", "view"), ("changes", "view"), ("catalog?schema=orders", "view"), diff --git a/crates/sqlbuild-orchestration/tests/api/test_person_sessions.rs b/crates/sqlbuild-orchestration/tests/api/test_person_sessions.rs index 6984b0a561..9753574753 100644 --- a/crates/sqlbuild-orchestration/tests/api/test_person_sessions.rs +++ b/crates/sqlbuild-orchestration/tests/api/test_person_sessions.rs @@ -93,6 +93,12 @@ fn given_browser_session_roles_when_work_is_requested_then_scope_revocation_and_ let mut input = snapshot(location); input.documents.get_mut("triggers").expect("triggers")["health_orders"]["when"] = json!({"kind":"custom", "function":"orders_ready", "code_hash":"orders_v1"}); + input.documents.get_mut("jobs").expect("jobs")["orders"]["concurrency_groups"] = + json!(["warehouse_heavy"]); + input.documents.insert( + "groups".into(), + json!({"nightly": {"description": "Nightly orders", "selection": ["orders"], "jobs": {"orders": []}}}), + ); let sql_input: Value = serde_json::from_str( &std::fs::read_to_string( std::path::Path::new(env!("CARGO_MANIFEST_DIR")) @@ -279,6 +285,7 @@ fn given_browser_session_roles_when_work_is_requested_then_scope_revocation_and_ assert!(!page.to_string().contains("select 1 as id")); let known_reads = [ format!("{metadata}/{}/nodes", seed_run.run_id), + format!("/api/control/locations/{location}/targets/development/jobs"), format!("/api/control/locations/{location}/targets/development/triggers"), format!( "/api/control/locations/{location}/targets/development/triggers/{}/ticks", @@ -312,6 +319,32 @@ fn given_browser_session_roles_when_work_is_requested_then_scope_revocation_and_ "{read_path}" ); } + let jobs: Value = request( + &provider.client, + session, + ( + Method::GET, + &format!("/api/control/locations/{location}/targets/development/jobs"), + ), + (&Value::Null, &session.csrf), + ) + .await + .json() + .await + .expect("jobs response"); + let listed_jobs: Vec = + jobs.as_array().into_iter().flatten().cloned().collect(); + let expected_jobs = [ + json!({"job_key": "nightly", "kind": "group", "description": "Nightly orders", + "selection": ["orders"], "concurrency_groups": [], "snapshot_id": published.snapshot_id}), + json!({"job_key": "orders", "kind": "job", "description": "Refresh orders", + "selection": ["orders"], "concurrency_groups": ["warehouse_heavy"], "snapshot_id": published.snapshot_id}), + ]; + assert_eq!( + listed_jobs, + expected_jobs[..expected_jobs.len() * usize::from(jobs.is_array())], + "jobs list declared metadata and never settings" + ); for read_path in &unknown_reads { let missing = request( &provider.client, diff --git a/fensu.toml b/fensu.toml index fbabf3ab4c..c66c31072f 100644 --- a/fensu.toml +++ b/fensu.toml @@ -1381,7 +1381,7 @@ paths = ["crates/sqlbuild-orchestration/src/api/_helpers/resources/administrator 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/_helpers/reads.rs", "crates/sqlbuild-orchestration/src/declarations/main/list_person_triggers.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_person_run_nodes.rs", "crates/sqlbuild-orchestration/src/runs/main/list_run_nodes.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_person_ticks.rs", "crates/sqlbuild-orchestration/src/store/main/read_records.rs", "crates/sqlbuild-orchestration/src/platform/main/location_work.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_history/main/read_person_microbatch_coverage.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/_helpers/reads.rs", "crates/sqlbuild-orchestration/src/declarations/main/list_person_jobs.rs", "crates/sqlbuild-orchestration/src/declarations/main/list_person_triggers.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_person_run_nodes.rs", "crates/sqlbuild-orchestration/src/runs/main/list_run_nodes.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_person_ticks.rs", "crates/sqlbuild-orchestration/src/store/main/read_records.rs", "crates/sqlbuild-orchestration/src/platform/main/location_work.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_history/main/read_person_microbatch_coverage.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]] diff --git a/ui/src/lib/api/client.ts b/ui/src/lib/api/client.ts index fd900528bf..11d6581757 100644 --- a/ui/src/lib/api/client.ts +++ b/ui/src/lib/api/client.ts @@ -109,6 +109,8 @@ export interface ApiClient extends PlatformApi { listJobs(scope: Scope): Promise; getJob(scope: Scope, key: string): Promise; launchJob(scope: Scope, key: string, options?: JobLaunchOptions): Promise; + /** Launches with `--full-refresh`, rebuilding the job's tables from scratch. */ + launchJobFromScratch(scope: Scope, key: string, options?: JobLaunchOptions): Promise; listTriggers(scope: Scope): Promise; getTrigger(scope: Scope, id: string): Promise; diff --git a/ui/src/lib/api/http-client.ts b/ui/src/lib/api/http-client.ts index 52d835ca88..b72b3b57d3 100644 --- a/ui/src/lib/api/http-client.ts +++ b/ui/src/lib/api/http-client.ts @@ -1,15 +1,35 @@ import createFetch from 'openapi-fetch'; import type { components, paths } from '$control/schema.d.ts'; import { ApiError, type ApiClient } from './client'; -import { RUN_WINDOW, runDetailOf, runRowPage, runTabCountsOf, toRunNode, toRuns } from './http-runs'; +import { jobDetailOf, toJob, toTick, toTrigger, triggerDetailOf } from './http-jobs'; +import { RUN_WINDOW, runDetailOf, runRowPage, runsMatching, runTabCountsOf, toRunNode, toRuns } from './http-runs'; import type { ChangeSubscription, ChangedChannel, ChangedSignal, ConnectionState } from './signals'; import type { AuthMode, Bootstrap, Capability, Location, Scope } from './types'; -import type { RerunMode, Run, RunDetail, RunRowPage, RunRowQuery, RunTabCounts } from './types/operate'; +import type { Job, JobDetail, RerunMode, Run, RunDetail, RunListQuery, RunRowPage, RunRowQuery, RunTabCounts, Trigger, TriggerDetail, TriggerState } from './types/operate'; type Wire = components['schemas']; /** The client methods the server has endpoints for; the shell hides pages that need others. */ -const SUPPORTED: ReadonlySet = new Set(['supports', 'bootstrap', 'changes', 'listRunRows', 'runTabCounts', 'getRun', 'getRunDetail', 'cancelRun', 'rerunRun']); +const SUPPORTED: ReadonlySet = new Set([ + 'supports', + 'bootstrap', + 'changes', + 'listRunRows', + 'runTabCounts', + 'listRuns', + 'getRun', + 'getRunDetail', + 'cancelRun', + 'rerunRun', + 'listJobs', + 'getJob', + 'launchJob', + 'listTriggers', + 'getTrigger', + 'setTriggerStates', + 'resetTriggerToDefault', + 'resetSensorCursor' +]); /** A UI location is one server location and target; its id carries both. */ export function locationKey(locationId: string, target: string): string { @@ -190,9 +210,105 @@ export function createHttpClient({ baseUrl = '', onUnauthorized }: HttpClientOpt if (response.status === 401) onUnauthorized(); if (!response.ok) throw new ApiError(errorText(error, response), response.status); return requireRun(scope, body.run_id); + }, + listRuns: async (scope: Scope, query: RunListQuery = {}): Promise => runsMatching(toRuns(await listedRuns(scope), scope.locationId), query), + async listJobs(scope: Scope): Promise { + const [jobs, triggers, runs] = await Promise.all([listedJobs(scope), triggersOf(scope), recentRuns(scope)]); + return jobs.map((job) => toJob(job, triggers, runs)); + }, + async getJob(scope: Scope, key: string): Promise { + const [jobs, triggers, runs] = await Promise.all([listedJobs(scope), triggersOf(scope), recentRuns(scope)]); + const wire = jobs.find((job) => job.job_key === key); + return wire ? jobDetailOf(toJob(wire, triggers, runs), wire) : null; + }, + async launchJob(scope: Scope, key: string): Promise { + const { locationId, target } = parseLocationKey(scope.locationId); + const body = { run_id: crypto.randomUUID(), location_id: locationId, target, job_key: key, priority: 0, reason: 'Started from the web UI' }; + const { error, response } = await http.POST('/api/control/runs', { body, params: { header: { 'x-sqb-csrf': csrfToken(document.cookie) } } }); + if (response.status === 401) onUnauthorized(); + if (!response.ok) throw new ApiError(errorText(error, response), response.status); + return requireRun(scope, body.run_id); + }, + listTriggers: (scope: Scope): Promise => triggersOf(scope), + async getTrigger(scope: Scope, id: string): Promise { + const [wires, runs] = await Promise.all([listedTriggers(scope), recentRuns(scope)]); + const wire = wires.find((t) => t.trigger_id === id); + if (!wire) return null; + const ticks = (await ticksOf(scope, id, 100)).map(toTick); + return triggerDetailOf(toTrigger(wire, runs, ticks[0] ?? null), wire, ticks); + }, + async setTriggerStates(scope: Scope, ids: string[], state: TriggerState, reason?: string | null): Promise { + for (const [done, id] of ids.entries()) { + try { + await changeTrigger(id, { kind: state === 'running' ? 'enable' : 'pause' }, reason); + } catch (error) { + if (done === 0 || !(error instanceof ApiError)) throw error; + throw new ApiError(`${error.message} (${done} of ${ids.length} triggers were already changed)`, error.status); + } + } + return (await triggersOf(scope)).filter((t) => ids.includes(t.id)); + }, + async resetTriggerToDefault(scope: Scope, id: string): Promise { + await changeTrigger(id, { kind: 'reset_default' }); + return requireTrigger(scope, id); + }, + async resetSensorCursor(scope: Scope, id: string, reason?: string | null): Promise { + await changeTrigger(id, { kind: 'reset_cursor' }, reason); + const detail = await implemented.getTrigger(scope, id); + if (!detail) throw new ApiError('The trigger is no longer published', 404); + return detail; } }; + async function listedJobs(scope: Scope) { + const { locationId, target } = parseLocationKey(scope.locationId); + const { data, error, response } = await http.GET('/api/control/locations/{location_id}/targets/{target}/jobs', { params: { path: { location_id: locationId, target } } }); + if (response.status === 401) onUnauthorized(); + if (!data) throw new ApiError(errorText(error, response), response.status); + return data; + } + + async function listedTriggers(scope: Scope) { + const { locationId, target } = parseLocationKey(scope.locationId); + const { data, error, response } = await http.GET('/api/control/locations/{location_id}/targets/{target}/triggers', { params: { path: { location_id: locationId, target } } }); + if (response.status === 401) onUnauthorized(); + if (!data) throw new ApiError(errorText(error, response), response.status); + return data; + } + + async function ticksOf(scope: Scope, triggerId: string, limit: number) { + const { locationId, target } = parseLocationKey(scope.locationId); + const { data, error, response } = await http.GET('/api/control/locations/{location_id}/targets/{target}/triggers/{trigger_id}/ticks', { + params: { path: { location_id: locationId, target, trigger_id: triggerId }, query: { limit } } + }); + if (response.status === 401) onUnauthorized(); + if (!data) throw new ApiError(errorText(error, response), response.status); + return data; + } + + const recentRuns = async (scope: Scope) => toRuns(await listedRuns(scope), scope.locationId); + + /** Triggers with their latest tick and the newest listed run each started. */ + async function triggersOf(scope: Scope): Promise { + const [wires, runs] = await Promise.all([listedTriggers(scope), recentRuns(scope)]); + const latest = await Promise.all(wires.map(async (t) => (await ticksOf(scope, t.trigger_id, 1)).map(toTick)[0] ?? null)); + return wires.map((t, i) => toTrigger(t, runs, latest[i])); + } + + async function requireTrigger(scope: Scope, id: string): Promise { + const trigger = (await triggersOf(scope)).find((t) => t.id === id); + if (!trigger) throw new ApiError('The trigger is no longer published', 404); + return trigger; + } + + /** Records an operator change; the server takes the actor from the session and needs a reason. */ + async function changeTrigger(triggerId: string, change: Wire['TriggerChange'], reason?: string | null) { + const body = { trigger_id: triggerId, event_id: crypto.randomUUID(), actor: '', reason: reason?.trim() || 'Changed from the web UI', change }; + const { error, response } = await http.POST('/api/control/triggers', { body, params: { header: { 'x-sqb-csrf': csrfToken(document.cookie) } } }); + if (response.status === 401) onUnauthorized(); + if (!response.ok) throw new ApiError(errorText(error, response), response.status); + } + /** The newest runs of the location, up to the server's cap. */ async function listedRuns(scope: Scope) { const { locationId, target } = parseLocationKey(scope.locationId); diff --git a/ui/src/lib/api/http-jobs.test.ts b/ui/src/lib/api/http-jobs.test.ts new file mode 100644 index 0000000000..de992dfd7d --- /dev/null +++ b/ui/src/lib/api/http-jobs.test.ts @@ -0,0 +1,71 @@ +import { describe, expect, it } from 'vitest'; +import type { components } from '$control/schema.d.ts'; +import { cadenceOf, jobDetailOf, toJob, toTick, toTrigger, triggerDetailOf, triggerKindOf } from './http-jobs'; +import type { Run } from './types/operate'; + +type W = components['schemas']; + +const SNAPSHOT = '0191f0dd-0000-7000-8000-000000000000'; + +function trigger(id: string, name: string, jobKey: string, definition: W['TriggerDeclaration'], status: W['TriggerStatus'] = 'running'): W['TriggerRecord'] { + return { trigger_id: id, name, job_key: jobKey, definition_json: definition, status, last_tick_for: null, next_due_at: '2026-11-07T06:00:00Z' }; +} + +const daily = trigger('t-daily', 'orders_daily', 'orders', { description: 'Daily orders', job_key: 'orders', schedule: { kind: 'cron', expression: '0 6 * * *', timezone: 'UTC' }, default_status: 'paused' }); +const onInventory = trigger('t-inv', 'inventory_landed', 'inventory', { + description: 'New stock feed', + job_key: 'inventory', + minimum_interval_seconds: 60, + when: { kind: 'source_new_data', selection: ['source:stock_feed'] } +}); + +function run(id: string, jobKey: string, triggerId: string | null): Run { + return { id, jobKey, triggerId, status: 'success', startedAt: '2026-11-07T01:00:00Z', endedAt: '2026-11-07T01:01:00Z' } as Run; +} + +describe('trigger mapping', () => { + it('classifies schedules and sensors and describes their cadence', () => { + expect(triggerKindOf(daily.definition_json)).toBe('cron_schedule'); + expect(triggerKindOf(onInventory.definition_json)).toBe('node_sensor'); + expect(triggerKindOf({ ...onInventory.definition_json, when: { kind: 'run_event', job_key: 'orders', kinds: ['succeeded' as never] } } as W['SensorDefinition'])).toBe('job_sensor'); + expect(cadenceOf(daily.definition_json)).toBe('0 6 * * * (UTC)'); + expect(cadenceOf(onInventory.definition_json)).toBe('when source:stock_feed has new data'); + }); + + it('keeps the code default apart from the effective state and links the newest run it started', () => { + const runs = [run('r-new', 'orders', 't-daily'), run('r-manual', 'orders', null), run('r-old', 'orders', 't-daily')]; + const t = toTrigger({ ...daily, status: 'running' }, runs, null); + expect([t.defaultState, t.state, t.type, t.cron, t.timezone]).toEqual(['paused', 'running', 'schedule', '0 6 * * *', 'UTC']); + expect(t.lastRun?.id).toBe('r-new'); + expect(toTrigger(onInventory, runs, null).lastRun).toBeNull(); + }); + + it('links a group trigger to its group run, not a child listed before it', () => { + const child = { ...run('r-child', 'orders_member', 't-daily'), parentRunId: 'r-group' } as Run; + const group = { ...run('r-group', 'orders', 't-daily'), status: 'failed' } as Run; + expect(toTrigger(daily, [child, group], null).lastRun?.id).toBe('r-group'); + }); + + it('reports lateness from evaluation time and hides sensor parts the server lacks', () => { + const tick = toTick({ tick_id: 'k1', trigger_id: 't-daily', epoch: 1, scheduled_for: '2026-11-07T06:00:00Z', evaluated_at: '2026-11-07T06:02:30Z', outcome: 'requested', run_id: 'r-new' }); + expect(tick.lateBySeconds).toBe(150); + const detail = triggerDetailOf(toTrigger(onInventory, [], tick), onInventory, [tick]); + expect(detail.sensor).toMatchObject({ minimumIntervalSeconds: 60, cursorKnown: false }); + expect([detail.declaration, detail.upcoming, detail.evaluations, detail.stateHistory]).toEqual([null, null, null, null]); + expect(triggerDetailOf(toTrigger(daily, [], null), daily, []).sensor).toBeNull(); + }); +}); + +describe('job mapping', () => { + it('links a job to its triggers and newest run, and leaves unreported policy null', () => { + const wire: W['JobRecord'] = { job_key: 'orders', kind: 'job', description: 'Build orders', selection: ['model:customers', 'model:orders'], concurrency_groups: ['warehouse_heavy'], snapshot_id: SNAPSHOT }; + const triggers = [toTrigger(daily, [], null), toTrigger(onInventory, [], null)]; + const job = toJob(wire, triggers, [run('r-inv', 'inventory', null), run('r-ord', 'orders', null)]); + expect(job).toMatchObject({ key: 'orders', kind: 'job', nodeCount: 2, triggerIds: ['t-daily'], concurrencyGroups: ['warehouse_heavy'], select: null, overlap: null, priority: null }); + expect(job.lastRun?.id).toBe('r-ord'); + const detail = jobDetailOf(job, wire); + expect(detail.resolvedNodes).toEqual(['model:customers', 'model:orders']); + expect([detail.settings, detail.concurrency, detail.memberOf, detail.cursorType]).toEqual([null, null, null, null]); + expect(toJob({ ...wire, job_key: 'nightly', kind: 'group' }, triggers, []).kind).toBe('group'); + }); +}); diff --git a/ui/src/lib/api/http-jobs.ts b/ui/src/lib/api/http-jobs.ts new file mode 100644 index 0000000000..af8de57e62 --- /dev/null +++ b/ui/src/lib/api/http-jobs.ts @@ -0,0 +1,145 @@ +import type { components } from '$control/schema.d.ts'; +import type { Job, JobDetail, Run, RunRef, Tick, Trigger, TriggerDetail, TriggerKind } from './types/operate'; + +type Wire = components['schemas']; + +const refOf = (run: Run): RunRef => ({ id: run.id, status: run.status, startedAt: run.startedAt, endedAt: run.endedAt }); + +/** The newest listed run matching, as listed newest first. */ +const newest = (runs: Run[], match: (run: Run) => boolean): RunRef | null => { + const run = runs.find(match); + return run ? refOf(run) : null; +}; + +function isSchedule(declaration: Wire['TriggerDeclaration']): declaration is Wire['TriggerDefinition'] { + return 'schedule' in declaration; +} + +export function triggerKindOf(declaration: Wire['TriggerDeclaration']): TriggerKind { + if (isSchedule(declaration)) return declaration.schedule.kind === 'cron' ? 'cron_schedule' : 'calendar_schedule'; + const kind = declaration.when.kind; + if (kind === 'run_event') return 'job_sensor'; + if (kind === 'platform_event') return 'platform_sensor'; + if (kind === 'custom') return 'custom_sensor'; + return 'node_sensor'; +} + +/** The declared cadence or condition in words. */ +export function cadenceOf(declaration: Wire['TriggerDeclaration']): string { + if (isSchedule(declaration)) { + const s = declaration.schedule; + return s.kind === 'cron' ? `${s.expression} (${s.timezone})` : `${s.times.length} calendar time${s.times.length === 1 ? '' : 's'}`; + } + const w = declaration.when; + if (w.kind === 'run_event') return `when ${w.job_key} ${w.kinds.join(' or ')}`; + if (w.kind === 'node_event') return `when ${w.selection.join(', ')} ${w.kinds.join(' or ')}`; + if (w.kind === 'source_new_data') return `when ${w.selection.join(', ')} has new data`; + if (w.kind === 'source_stale') return `when ${w.selection.join(', ')} is stale for ${w.after_seconds}s`; + if (w.kind === 'platform_event') return `on platform events: ${w.kinds.join(', ')}`; + return `custom check ${w.function}`; +} + +export function toTick(wire: Wire['TickRecord']): Tick { + return { + id: wire.tick_id, + scheduledFor: wire.scheduled_for, + evaluatedAt: wire.evaluated_at, + outcome: wire.outcome, + runId: wire.run_id ?? null, + coveredByRunId: wire.covered_by_run_id ?? null, + reason: null, + lateBySeconds: Math.max(0, Math.round((Date.parse(wire.evaluated_at) - Date.parse(wire.scheduled_for)) / 1000)), + catchUp: false + }; +} + +export function toTrigger(wire: Wire['TriggerRecord'], runs: Run[], lastTick: Tick | null): Trigger { + const d = wire.definition_json; + const schedule = isSchedule(d) ? d.schedule : null; + return { + id: wire.trigger_id, + name: wire.name, + type: schedule ? 'schedule' : 'sensor', + kind: triggerKindOf(d), + description: d.description, + jobKey: wire.job_key, + cadence: cadenceOf(d), + cron: schedule?.kind === 'cron' ? schedule.expression : null, + timezone: schedule?.kind === 'cron' ? schedule.timezone : null, + defaultState: d.default_status ?? 'running', + state: wire.status, + nextDueAt: wire.next_due_at ?? null, + lastTick, + // Group children copy the group's trigger, so only top-level runs are the trigger's own. + lastRun: newest(runs, (r) => r.triggerId === wire.trigger_id && !r.parentRunId), + sourceFile: null, + definitionChanged: false, + warning: null + }; +} + +/** A trigger's detail; parts with no server source are null and their widgets stay hidden. */ +export function triggerDetailOf(trigger: Trigger, wire: Wire['TriggerRecord'], ticks: Tick[]): TriggerDetail { + const d = wire.definition_json; + return { + ...trigger, + declaration: null, + targets: null, + catchUp: null, + upcoming: null, + calendar: null, + sensor: isSchedule(d) + ? null + : { + condition: trigger.cadence, + nodeConditions: [], + minimumIntervalSeconds: d.minimum_interval_seconds, + timeoutSeconds: null, + cursor: null, + cursorUpdatedAt: null, + cursorKnown: false, + currentFingerprint: null, + lastEvaluatedFingerprint: null + }, + ticks, + evaluations: null, + runRequests: null, + stateHistory: null + }; +} + +export function toJob(wire: Wire['JobRecord'], triggers: Trigger[], runs: Run[]): Job { + return { + key: wire.job_key, + kind: wire.kind === 'group' ? 'group' : 'job', + description: wire.description, + // The server stores the resolved node list, not the selector text. + select: null, + members: null, + nodeCount: wire.selection.length, + overlap: null, + onFailure: null, + concurrencyGroups: wire.concurrency_groups, + priority: null, + owner: null, + sourceFile: null, + triggerIds: triggers.filter((t) => t.jobKey === wire.job_key).map((t) => t.id), + lastRun: newest(runs, (r) => r.jobKey === wire.job_key) + }; +} + +export function jobDetailOf(job: Job, wire: Wire['JobRecord']): JobDetail { + return { + ...job, + declaration: null, + exclude: [], + settings: null, + resolvedNodes: wire.selection, + concurrency: null, + targets: null, + groupMembers: null, + criticalNodes: [], + memberOf: null, + cursorType: null + }; +} diff --git a/ui/src/lib/api/http-runs.ts b/ui/src/lib/api/http-runs.ts index 305ffa1848..86fb68311f 100644 --- a/ui/src/lib/api/http-runs.ts +++ b/ui/src/lib/api/http-runs.ts @@ -1,7 +1,7 @@ import type { components } from '$control/schema.d.ts'; import { isActiveRun } from '$lib/status'; import type { Principal } from './types'; -import type { NodeCounts, Run, RunDetail, RunNode, RunNodeStatus, RunRow, RunRowPage, RunRowQuery, RunTabCounts } from './types/operate'; +import type { NodeCounts, Run, RunDetail, RunListQuery, RunNode, RunNodeStatus, RunRow, RunRowPage, RunRowQuery, RunTabCounts } from './types/operate'; type Wire = components['schemas']; @@ -75,6 +75,23 @@ export function toRuns(wires: Wire['RunRecord'][], locationId: string): Run[] { return wires.map((w) => toRun(w, locationId, children.get(w.run_id) ?? 0)); } +/** Listed runs under a run-list query, newest first. */ +export function runsMatching(runs: Run[], query: RunListQuery): Run[] { + const matching = runs.filter( + (r) => + (!query.jobKey || r.jobKey === query.jobKey) && + (!query.triggerId || r.triggerId === query.triggerId) && + (!query.parentRunId || r.parentRunId === query.parentRunId) && + (!query.topLevelOnly || !r.parentRunId) && + (!query.statuses?.length || query.statuses.includes(r.status)) && + (!query.kinds?.length || query.kinds.includes(r.kind)) && + (!query.origins?.length || query.origins.includes(r.origin)) && + (!query.since || r.queuedAt >= query.since) && + (!query.until || r.queuedAt < query.until) + ); + return query.limit === undefined ? matching : matching.slice(0, query.limit); +} + function rowOf(run: Run, wire: Wire['RunRecord']): RunRow { const c = run.nodeCounts; const total = c.success + c.warn + c.failed + c.skipped + c.running + c.pending; diff --git a/ui/src/lib/api/mock/mock-client.ts b/ui/src/lib/api/mock/mock-client.ts index 0e632aba05..49e78f99f0 100644 --- a/ui/src/lib/api/mock/mock-client.ts +++ b/ui/src/lib/api/mock/mock-client.ts @@ -274,6 +274,11 @@ export class MockApiClient implements ApiClient { return this.#operate.launchJob(scope, key, options); } + async launchJobFromScratch(scope: Scope, key: string, options?: JobLaunchOptions): Promise { + this.#location(scope); + return this.#operate.launchJob(scope, key, { ...options, fullRefresh: true }); + } + async listTriggers(scope: Scope): Promise { this.#location(scope); return this.#operate.listTriggers(scope); diff --git a/ui/src/lib/api/mock/operate-store.ts b/ui/src/lib/api/mock/operate-store.ts index 9765fa7a3a..813a2bc267 100644 --- a/ui/src/lib/api/mock/operate-store.ts +++ b/ui/src/lib/api/mock/operate-store.ts @@ -464,7 +464,7 @@ export class MockOperate { return this.#ctx.reply(this.#view(rec, scope)); } - launchJob(scope: Scope, key: string, options: JobLaunchOptions = {}): Promise { + launchJob(scope: Scope, key: string, options: JobLaunchOptions & { fullRefresh?: boolean } = {}): Promise { this.#need('job.run'); const job = jobDef(key); if (!job) throw new ApiError(`Unknown job ${key}`, 404); @@ -1163,6 +1163,7 @@ export class MockOperate { ...def.sensor, cursor: rt.cursor, cursorUpdatedAt: rt.cursorUpdatedAt ? new Date(rt.cursorUpdatedAt).toISOString() : null, + cursorKnown: true, currentFingerprint: currentFp, lastEvaluatedFingerprint: sensorFp } diff --git a/ui/src/lib/api/mock/operate.ts b/ui/src/lib/api/mock/operate.ts index 35237ec9b7..c500962542 100644 --- a/ui/src/lib/api/mock/operate.ts +++ b/ui/src/lib/api/mock/operate.ts @@ -31,7 +31,13 @@ export const SNAPSHOT_GENERATION = 42; // ---- jobs and groups ---- -type JobDef = Omit & { +/** The mock knows every field the server may leave out. */ +type Known = Omit & { [P in K]-?: NonNullable }; + +type JobDef = Known< + Omit, + 'members' | 'overlap' | 'onFailure' | 'priority' | 'sourceFile' | 'declaration' | 'settings' +> & { after: Record; }; diff --git a/ui/src/lib/api/types/operate.ts b/ui/src/lib/api/types/operate.ts index 40bbd8a81a..5d0a831c31 100644 --- a/ui/src/lib/api/types/operate.ts +++ b/ui/src/lib/api/types/operate.ts @@ -483,7 +483,6 @@ export type CursorWindow = { export type JobLaunchOptions = { window?: CursorWindow; - fullRefresh?: boolean; }; /** `selected` reruns exactly the nodes passed with the request. */ @@ -503,15 +502,16 @@ export type Job = { description: string; /** Selector for jobs; null for groups. */ select: string | null; - /** Member jobs for groups. */ - members: string[]; + /** Member jobs for groups; null when the server doesn't say. */ + members: string[] | null; nodeCount: number; - overlap: OverlapPolicy; - onFailure: OnFailure; + /** Null where the server doesn't report it (also `onFailure`, `priority`, `sourceFile`). */ + overlap: OverlapPolicy | null; + onFailure: OnFailure | null; concurrencyGroups: string[]; - priority: number; + priority: number | null; owner: string | null; - sourceFile: string; + sourceFile: string | null; triggerIds: string[]; lastRun: RunRef | null; }; @@ -532,18 +532,19 @@ export type JobGroupMember = { }; export type JobDetail = Job & { - declaration: { sourceFile: string; line: number; code: string }; + /** Null where the server doesn't report it, as for `settings`, `concurrency`, `groupMembers` and `memberOf`. */ + declaration: { sourceFile: string; line: number; code: string } | null; exclude: string[]; /** Settings are CLI flags validated against the command table. */ - settings: Record; + settings: Record | null; resolvedNodes: string[]; - concurrency: ConcurrencyGroup[]; + concurrency: ConcurrencyGroup[] | null; /** Null when the job applies to every target. */ targets: string[] | null; - groupMembers: JobGroupMember[]; + groupMembers: JobGroupMember[] | null; criticalNodes: string[]; /** Groups this job belongs to. */ - memberOf: string[]; + memberOf: string[] | null; /** Nodes with a cursor, so a cursor window applies. */ cursorType: 'timestamp' | 'integer' | null; }; @@ -601,7 +602,7 @@ export type Trigger = { nextDueAt: Iso | null; lastTick: Tick | null; lastRun: RunRef | null; - sourceFile: string; + sourceFile: string | null; /** Code changed since the last evaluation (sensors). */ definitionChanged: boolean; /** The watchdog's warning, e.g. an overdue tick or a low calendar horizon. */ @@ -647,11 +648,12 @@ export type SensorRunRequest = { export type CatchUpPolicy = { mode: 'latest' | 'all'; max: number | null }; export type TriggerDetail = Trigger & { - declaration: { sourceFile: string; line: number; code: string }; + /** Null where the server doesn't report it, as for `upcoming`, `evaluations`, `runRequests` and `stateHistory`. */ + declaration: { sourceFile: string; line: number; code: string } | null; targets: string[] | null; catchUp: CatchUpPolicy | null; /** Upcoming due times (schedules). */ - upcoming: Iso[]; + upcoming: Iso[] | null; /** Calendar schedules: times are computed ahead and stored up to `storedUntil`. */ calendar: { horizon: string; @@ -668,11 +670,13 @@ export type TriggerDetail = Trigger & { timeoutSeconds: number | null; cursor: string | null; cursorUpdatedAt: Iso | null; - currentFingerprint: string; + /** False when the server doesn't report the cursor; resetting it still works. */ + cursorKnown: boolean; + currentFingerprint: string | null; lastEvaluatedFingerprint: string | null; } | null; ticks: Tick[]; - evaluations: SensorEvaluation[]; - runRequests: SensorRunRequest[]; - stateHistory: TriggerStateEvent[]; + evaluations: SensorEvaluation[] | null; + runRequests: SensorRunRequest[] | null; + stateHistory: TriggerStateEvent[] | null; }; diff --git a/ui/src/routes/jobs/+page.svelte b/ui/src/routes/jobs/+page.svelte index 3046ca2745..5f1243506b 100644 --- a/ui/src/routes/jobs/+page.svelte +++ b/ui/src/routes/jobs/+page.svelte @@ -17,8 +17,8 @@ import { isActiveRun, isProblemRun, runStatusLabel, runStatusTone } from '$lib/status'; import { overlapHelp, overlapPlain } from '$lib/operate/labels'; - const jobs = createResource((client, scope) => client.listJobs(scope), ['jobs', 'runs']); - const triggers = createResource((client, scope) => client.listTriggers(scope), ['triggers']); + const jobs = createResource((client, scope) => client.listJobs(scope), ['jobs', 'runs', 'snapshot']); + const triggers = createResource((client, scope) => client.listTriggers(scope), ['triggers', 'ticks']); const triggerById = $derived(new Map((triggers.data ?? []).map((t) => [t.id, t]))); const canToggle = $derived(app.can('trigger.toggle')); @@ -46,10 +46,14 @@ if (lastRun === 'active' && !(j.lastRun && isActiveRun(j.lastRun.status))) return false; if (lastRun === 'never' && j.lastRun) return false; const q = query.trim().toLowerCase(); - return !q || `${j.key} ${j.description} ${j.select ?? ''} ${j.members.join(' ')}`.toLowerCase().includes(q); + return !q || `${j.key} ${j.description} ${j.select ?? ''} ${(j.members ?? []).join(' ')}`.toLowerCase().includes(q); }) ); const scheduledCount = $derived(all.filter((j) => j.triggerIds.some((id) => triggerById.get(id)?.state === 'running')).length); + // Columns the server doesn't report stay hidden rather than showing dashes. + const showPolicy = $derived(all.some((j) => j.overlap !== null)); + const showOwner = $derived(all.some((j) => j.owner !== null)); + const showSelection = $derived(all.some((j) => j.select !== null || j.members !== null)); const failingCount = $derived(all.filter((j) => j.lastRun && isProblemRun(j.lastRun.status)).length); @@ -96,11 +100,11 @@ Job - Selection + {#if showSelection}Selection{/if} Nodes - If already running + {#if showPolicy}If already running{/if} Started by - Owner + {#if showOwner}Owner{/if} Last run @@ -117,15 +121,17 @@ - - {#if j.kind === 'group'} - Group · {j.members.join(', ')} - {:else} - {j.select} - {/if} - + {#if showSelection} + + {#if j.kind === 'group'} + Group{#if j.members} · {j.members.join(', ')}{/if} + {:else} + {j.select} + {/if} + + {/if} {j.nodeCount} - {overlapPlain[j.overlap]}{#if j.priority} · priority {j.priority}{/if} + {#if showPolicy}{j.overlap ? overlapPlain[j.overlap] : '—'}{#if j.priority} · priority {j.priority}{/if}{/if} {#if jobTriggers.length === 0} Only by hand @@ -151,7 +157,7 @@ {/if} - {j.owner ?? '—'} + {#if showOwner}{j.owner ?? '—'}{/if} {#if j.lastRun} e.stopPropagation()}> diff --git a/ui/src/routes/jobs/[name]/group-order.svelte b/ui/src/routes/jobs/[name]/group-order.svelte index 3ad0c95d05..fe89275eba 100644 --- a/ui/src/routes/jobs/[name]/group-order.svelte +++ b/ui/src/routes/jobs/[name]/group-order.svelte @@ -6,8 +6,8 @@ let { job }: { job: JobDetail } = $props(); const stages = $derived.by(() => { - const max = Math.max(0, ...job.groupMembers.map((m) => m.stage)); - return Array.from({ length: max + 1 }, (_, i) => job.groupMembers.filter((m) => m.stage === i)); + const max = Math.max(0, ...(job.groupMembers ?? []).map((m) => m.stage)); + return Array.from({ length: max + 1 }, (_, i) => (job.groupMembers ?? []).filter((m) => m.stage === i)); }); diff --git a/ui/src/routes/jobs/[name]/job-view.svelte b/ui/src/routes/jobs/[name]/job-view.svelte index 9503871597..e61bec3718 100644 --- a/ui/src/routes/jobs/[name]/job-view.svelte +++ b/ui/src/routes/jobs/[name]/job-view.svelte @@ -14,15 +14,16 @@ let { name }: { name: string } = $props(); - const job = createResource((client, scope) => client.getJob(scope, name), ['jobs', 'runs']); + const job = createResource((client, scope) => client.getJob(scope, name), ['jobs', 'runs', 'snapshot']); const jobRuns = createResource((client, scope) => client.listRuns(scope, { jobKey: name, limit: 12 }), ['runs']); - const triggers = createResource((client, scope) => client.listTriggers(scope), ['triggers']); + const triggers = createResource((client, scope) => client.listTriggers(scope), ['triggers', 'ticks']); const nodes = createResource((client, scope) => client.listSnapshotNodes(scope), ['snapshot']); const def = $derived(job.data ?? null); const kindOf = $derived(new Map((nodes.data ?? []).map((n) => [n.key, n.kind]))); const jobTriggers = $derived((triggers.data ?? []).filter((t) => def?.triggerIds.includes(t.id))); const runs = $derived(jobRuns.data ?? []); + const showCost = $derived(runs.some((r) => r.cost !== null)); const canToggle = $derived(app.can('trigger.toggle')); let launchOpen = $state(false); let actionError = $state(null); @@ -68,7 +69,7 @@ ← Jobs {name} {#if def} - {def.kind === 'group' ? `Group of ${def.members.length} jobs` : 'Job'} · {def.nodeCount} nodes + {def.kind === 'group' ? (def.members ? `Group of ${def.members.length} jobs` : 'Group') : 'Job'} · {def.nodeCount} nodes {def.description} {/if} @@ -82,12 +83,14 @@ {#if def}
-
-

Code read-only; change it in the project and deploy

- -
+ {#if def.declaration} +
+

Code read-only; change it in the project and deploy

+ +
+ {/if} - {#if def.kind === 'group'} + {#if def.kind === 'group' && def.groupMembers}

Jobs in this group, in order from the data each job reads and writes, plus any after in the code

@@ -107,8 +110,8 @@ {formatNodeCounts(r.nodeCounts)} {r.startedAt ? formatDuration(r.startedAt, r.endedAt, app.now) : '—'} {formatAgo(r.startedAt ?? r.queuedAt, app.now)} - {r.requestedBy.name} - {formatCost(r.cost)} + {r.requestedBy.displayName} + {#if showCost}{formatCost(r.cost)}{/if} {/each} @@ -119,7 +122,7 @@
-

What it builds · {def.resolvedNodes.length} nodes {def.select ? `--select ${def.select}` : 'everything its jobs build'}{def.exclude.length ? ` --exclude ${def.exclude.join(' ')}` : ''}

+

What it builds · {def.resolvedNodes.length} nodes {def.select ? `--select ${def.select}` : def.kind === 'group' ? 'everything its jobs build' : ''}{def.exclude.length ? ` --exclude ${def.exclude.join(' ')}` : ''}

{#each def.resolvedNodes as n (n)} @@ -138,29 +141,37 @@
Kind
{def.kind === 'group' ? 'Group of jobs' : 'Job'}
{#if def.select}
Selection
{def.select}
{/if} -
If already running
{overlapPlain[def.overlap]}
-
If something fails
{onFailurePlain[def.onFailure]}
-
Priority
{def.priority}
-
Targets
{def.targets?.join(', ') ?? 'all targets'}
-
Owner
{def.owner ?? '—'}
-
Defined in
{def.sourceFile}
- {#if def.memberOf.length} + {#if def.overlap}
If already running
{overlapPlain[def.overlap]}
{/if} + {#if def.onFailure}
If something fails
{onFailurePlain[def.onFailure]}
{/if} + {#if def.priority !== null}
Priority
{def.priority}
{/if} + {#if def.declaration}
Targets
{def.targets?.join(', ') ?? 'all targets'}
{/if} + {#if def.owner}
Owner
{def.owner}
{/if} + {#if def.sourceFile}
Defined in
{def.sourceFile}
{/if} + {#if def.memberOf?.length} {/if}
-

{overlapHelp[def.overlap]}. {onFailureHelp[def.onFailure]}.

+ {#if def.overlap && def.onFailure}

{overlapHelp[def.overlap]}. {onFailureHelp[def.onFailure]}.

{/if} -

Extra options

- {#if Object.keys(def.settings).length} - {#each Object.entries(def.settings) as [flag, value] (flag)} -

{flag}{value === true ? '' : ` ${value}`}

- {/each} - {:else} -

None; it runs with the defaults.

+ {#if def.settings} +

Extra options

+ {#if Object.keys(def.settings).length} + {#each Object.entries(def.settings) as [flag, value] (flag)} +

{flag}{value === true ? '' : ` ${value}`}

+ {/each} + {:else} +

None; it runs with the defaults.

+ {/if} {/if}

Shared slots

- {#if def.concurrency.length} + {#if def.concurrency === null} + {#if def.concurrencyGroups.length} + {#each def.concurrencyGroups as g (g)}

Uses {slotLabel(g)} slots

{/each} + {:else} +

None. It only waits for runs that update the same data.

+ {/if} + {:else if def.concurrency.length} {#each def.concurrency as g (g.key)}

{g.holders.length} of {g.limit} {slotLabel(g.key)} slot{g.limit === 1 ? '' : 's'} in use @@ -173,7 +184,7 @@

Schedules and sensors · {jobTriggers.length}

{#if jobTriggers.length === 0} -

Nothing starts it automatically. Run it by hand{def.memberOf.length ? ', or through its groups' : ''}.

+

Nothing starts it automatically. Run it by hand{def.memberOf?.length ? ', or through its groups' : ''}.

{:else} {#each jobTriggers as t (t.id)}
diff --git a/ui/src/routes/jobs/[name]/launch-form.svelte b/ui/src/routes/jobs/[name]/launch-form.svelte index aa2edd2976..8da2536351 100644 --- a/ui/src/routes/jobs/[name]/launch-form.svelte +++ b/ui/src/routes/jobs/[name]/launch-form.svelte @@ -14,6 +14,7 @@ let error = $state(null); const canRewrite = $derived(app.can('run.rewrite_history')); + const canFullRefresh = $derived(app.client.supports('launchJobFromScratch')); const rewrites = $derived(fullRefresh || useWindow); const placeholder = $derived(job.cursorType === 'integer' ? 'e.g. 1200' : 'e.g. 2026-10-01T00:00:00Z'); const windowInvalid = $derived( @@ -31,10 +32,8 @@ busy = true; error = null; try { - const run = await app.client.launchJob(scope, job.key, { - fullRefresh: fullRefresh || undefined, - window: useWindow && job.cursorType ? { type: job.cursorType, start: start || null, end: end || null } : undefined - }); + const options = { window: useWindow && job.cursorType ? { type: job.cursorType, start: start || null, end: end || null } : undefined }; + const run = fullRefresh ? await app.client.launchJobFromScratch(scope, job.key, options) : await app.client.launchJob(scope, job.key, options); onclose(); await goto(`/runs/${run.id}`); } catch (e) { @@ -48,13 +47,16 @@

Run {job.kind === 'group' ? 'group' : 'job'} {job.key} now

- {job.kind === 'group' ? `Starts a group run, with one run per job in it (${job.members.length}).` : `Builds ${job.nodeCount} nodes with the job's options.`} If another run is updating the same data, it waits for it to finish. + {job.kind === 'group' ? `Starts a group run, with one run per job in it${job.members ? ` (${job.members.length})` : ''}.` : `Builds ${job.nodeCount} nodes with the job's options.`} If another run is updating the same data, it waits for it to finish.

+ {#if canFullRefresh || job.cursorType}
+ {#if canFullRefresh} + {/if} {#if job.cursorType}
+ {/if} {#if error}

{error}

{/if}
diff --git a/ui/src/routes/triggers/[id]/sensor-panel.svelte b/ui/src/routes/triggers/[id]/sensor-panel.svelte index b85ec51d57..8064bca98e 100644 --- a/ui/src/routes/triggers/[id]/sensor-panel.svelte +++ b/ui/src/routes/triggers/[id]/sensor-panel.svelte @@ -54,9 +54,13 @@

Where it's up to

- {sensor.cursor ?? '(the start: nothing handled yet)'} - {#if sensor.cursorUpdatedAt}updated {formatAgo(sensor.cursorUpdatedAt, app.now)}{/if} - (confirmReset = !confirmReset)}>Start over + {#if sensor.cursorKnown} + {sensor.cursor ?? '(the start: nothing handled yet)'} + {#if sensor.cursorUpdatedAt}updated {formatAgo(sensor.cursorUpdatedAt, app.now)}{/if} + {:else} + Starting over makes the next check look at everything again. + {/if} + (confirmReset = !confirmReset)}>Start over
{#if confirmReset}
@@ -71,6 +75,7 @@ {#if error}

{error}

{/if}
+{#if trigger.evaluations}

Checks · {trigger.evaluations.length}

@@ -95,7 +100,9 @@
{#if trigger.evaluations.length > shown}{/if}
+{/if} +{#if trigger.runRequests}

Runs it started · {trigger.runRequests.length}

{#if trigger.runRequests.length} @@ -117,6 +124,7 @@

{trigger.jobKey ? 'It hasn’t started any runs yet.' : 'This sensor only runs its own code; it doesn’t start runs.'}

{/if}
+{/if}