diff --git a/CHANGELOG.md b/CHANGELOG.md index 37552ab..5d89723 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,17 @@ # Changelog +## Unreleased + +### Added + +- **Draft sources.** A source that is not complete yet saves as a draft, which cannot be enabled, + instead of being refused. Its row carries `draft` and `problems`, each with the pipeline stage it + belongs to. An enabled source is still saved only complete, and enabling re-checks it. +- **Every problem with a source at once.** A refused save or enable is a 422 that lists all of the + source's `problems` (`{kind, stage, field?, message}`), not just the first, and an empty + transport setting (host, port, URL, bind address, topic, method, path) is now reported by name. + The SQLite schema moves to version 20. + ## 1.0.0, 2026-10-06 The first open-source release, under the MIT License. 1.0 carries everything up to 0.4.7, plus the diff --git a/crates/ot-server/src/api.rs b/crates/ot-server/src/api.rs index b337be4..49c4642 100644 --- a/crates/ot-server/src/api.rs +++ b/crates/ot-server/src/api.rs @@ -13,7 +13,7 @@ use axum::{Extension, Json, Router}; use ot_source::frame::Frame; use ot_source::pipeline::Pipeline; use ot_source::schema::{ExtensionField, ExtensionSchema}; -use ot_source::source::SourceSpec; +use ot_source::source::{Problem, ProblemKind, SourceSpec, Stage}; use ot_source::trace::Trace; use ot_store::{Entity, SourceRow, SourceWrite}; use serde::Deserialize; @@ -73,25 +73,72 @@ pub(crate) fn actor(headers: &HeaderMap) -> String { .map_or_else(|| "op:api".to_owned(), str::to_owned) } -/// Parse and validate a source spec, including against the published -/// extension schema version its mapping targets. Returns the spec, its -/// normalised JSON (defaults filled in) and that schema. -async fn parse_spec( - s: &AppState, - body: Value, -) -> Result<(SourceSpec, Value, ExtensionSchema), ApiError> { - let spec: SourceSpec = serde_json::from_value(body) - .map_err(|e| ApiError::unprocessable(format!("invalid source spec: {e}")))?; - crate::probe::file_allowed(&s.common, &spec.transport).map_err(ApiError::unprocessable)?; +/// A source spec as sent, checked: the spec, its normalised JSON (defaults +/// filled in), the published extension schema its mapping targets (if it +/// is published) and everything that stops it running. A spec that does +/// not parse, or is not allowed at all, is refused here. +struct Checked { + spec: SourceSpec, + normalised: Value, + schema: Option, + problems: Vec, +} + +impl Checked { + /// Refuse it if it cannot be saved (an invalid id or name). + fn savable(self) -> Result { + let invalid: Vec = self + .problems + .iter() + .filter(|p| p.kind == ProblemKind::Invalid) + .cloned() + .collect(); + if invalid.is_empty() { + Ok(self) + } else { + Err(ApiError::source_problems(&invalid)) + } + } + + /// Refuse it unless it is complete (it may run). + fn complete(self) -> Result<(SourceSpec, Value, ExtensionSchema), ApiError> { + match self.schema { + Some(schema) if self.problems.is_empty() => Ok((self.spec, self.normalised, schema)), + _ => Err(ApiError::source_problems(&self.problems)), + } + } +} + +/// One problem with a spec as a whole, for a refusal. +fn invalid(message: String) -> ApiError { + ApiError::source_problems(&[Problem { + kind: ProblemKind::Invalid, + stage: None, + field: None, + message, + }]) +} + +async fn check_spec(s: &AppState, body: Value) -> Result { + let spec: SourceSpec = + serde_json::from_value(body).map_err(|e| invalid(format!("invalid source spec: {e}")))?; + crate::probe::file_allowed(&s.common, &spec.transport).map_err(|e| { + ApiError::source_problems(&[Problem { + kind: ProblemKind::Invalid, + stage: Some(Stage::Source), + field: Some("path".into()), + message: e, + }]) + })?; let schema = published_schema(s, spec.pipeline.mapping.schema_version).await?; - spec.validate_against(schema.as_ref()) - .map_err(|e| ApiError::unprocessable(e.to_string()))?; + let problems = spec.problems(schema.as_ref()); let normalised = serde_json::to_value(&spec).map_err(|e| ApiError::internal(e.to_string()))?; - Ok(( + Ok(Checked { spec, normalised, - schema.expect("validated against a published schema"), - )) + schema, + problems, + }) } /// A published extension schema version, typed. @@ -175,17 +222,20 @@ async fn get_source( ))) } -async fn save( - s: &AppState, - spec: SourceSpec, - normalised: Value, - actor: String, -) -> Result { +async fn save(s: &AppState, c: Checked, actor: String) -> Result { + let Checked { + spec, + normalised, + problems, + .. + } = c; if spec.unauthenticated.is_some() { // The decision records the acceptance; the log says it plainly. tracing::warn!(source = %spec.id, transport = spec.transport.kind(), %actor, "listening source saved with unauthenticated senders accepted"); } + let problems = + serde_json::to_value(&problems).map_err(|e| ApiError::internal(e.to_string()))?; s.with_db(move |db| { db.put_source( &SourceWrite { @@ -195,6 +245,7 @@ async fn save( codec: spec.pipeline.codec.name(), priority: spec.priority, spec: &normalised, + problems: &problems, }, &actor, ) @@ -202,37 +253,47 @@ async fn save( .await } +/// A new source. One that is not complete yet is saved as a draft. async fn create_source( State(s): State, headers: HeaderMap, Json(body): Json, ) -> Result<(StatusCode, Json), ApiError> { - let (spec, normalised, _) = parse_spec(&s, body).await?; - let id = spec.id.clone(); + let c = check_spec(&s, body).await?.savable()?; + let id = c.spec.id.clone(); if s.with_db(move |db| db.get_source(&id)).await?.is_some() { return Err(ApiError::conflict(format!( "source {} already exists; use PUT to update", - spec.id + c.spec.id ))); } - let row = save(&s, spec, normalised, actor(&headers)).await?; + let row = save(&s, c, actor(&headers)).await?; Ok((StatusCode::CREATED, Json(with_status(&s, row).await))) } +/// Save a source. A disabled one may be saved as a draft; an enabled one +/// only complete, so that an edit never stops a running source. async fn put_source( State(s): State, Path(id): Path, headers: HeaderMap, Json(body): Json, ) -> Result, ApiError> { - let (spec, normalised, _) = parse_spec(&s, body).await?; - if spec.id != id { + let c = check_spec(&s, body).await?.savable()?; + if c.spec.id != id { return Err(ApiError::unprocessable(format!( "body id {:?} does not match path id {id:?} (source ids never change)", - spec.id + c.spec.id ))); } - let row = save(&s, spec, normalised, actor(&headers)).await?; + let enabled = s + .with_db(move |db| db.get_source(&id)) + .await? + .is_some_and(|r| r.enabled); + if enabled && !c.problems.is_empty() { + return Err(ApiError::source_problems(&c.problems)); + } + let row = save(&s, c, actor(&headers)).await?; Ok(Json(with_status(&s, row).await)) } @@ -250,21 +311,38 @@ async fn toggle( s: AppState, id: String, enabled: bool, + problems: Option, headers: HeaderMap, ) -> Result, ApiError> { let actor = actor(&headers); let row = s - .with_db(move |db| db.set_source_enabled(&id, enabled, &actor)) + .with_db(move |db| db.set_source_enabled(&id, enabled, problems.as_ref(), &actor)) .await?; Ok(Json(with_status(&s, row).await)) } +/// Enable a source: only a complete one, checked again now (the schema +/// version its mapping targets, say, may have been published since it was +/// saved). What the check finds is stored, so the list shows it too. async fn enable_source( State(s): State, Path(id): Path, headers: HeaderMap, ) -> Result, ApiError> { - toggle(s, id, true, headers).await + let key = id.clone(); + let row = s + .with_db(move |db| db.get_source(&key)) + .await? + .ok_or_else(|| ApiError::not_found("source"))?; + let c = check_spec(&s, row.spec).await?; + let found = serde_json::to_value(&c.problems).map_err(|e| ApiError::internal(e.to_string()))?; + if c.problems.is_empty() { + return toggle(s, id, true, Some(found), headers).await; + } + // Keep what was found, for the list; the source stays as it was. + s.with_db(move |db| db.set_source_problems(&id, &found)) + .await?; + Err(ApiError::source_problems(&c.problems)) } async fn disable_source( @@ -272,7 +350,7 @@ async fn disable_source( Path(id): Path, headers: HeaderMap, ) -> Result, ApiError> { - toggle(s, id, false, headers).await + toggle(s, id, false, None, headers).await } #[derive(Deserialize)] @@ -401,7 +479,7 @@ async fn validate_source( State(s): State, Json(body): Json, ) -> Result, ApiError> { - let (spec, normalised, schema) = parse_spec(&s, body.spec).await?; + let (spec, normalised, schema) = check_spec(&s, body.spec).await?.complete()?; let mut frames: Vec = body .samples .iter() @@ -768,8 +846,7 @@ mod tests { for (name, text) in [("aisstream", AIS), ("adsb-lol", ADSB)] { let spec: ot_source::source::SourceSpec = serde_json::from_str(text).unwrap_or_else(|e| panic!("{name}: {e}")); - spec.validate_against(Some(&schema)) - .unwrap_or_else(|e| panic!("{name}: {e}")); + assert_eq!(spec.problems(Some(&schema)), [], "{name}"); } // Examples that use only the core schema. for (name, text) in [ @@ -917,17 +994,30 @@ mod tests { spec["id"] = json!("udp-auth"); let accepted = spec.clone(); spec.as_object_mut().unwrap().remove("unauthenticated"); + // Without the acceptance it is a draft, and cannot be enabled. let (st, body) = call(&app, "POST", "/api/v1/sources", Some(spec.clone())).await; - assert_eq!(st, StatusCode::UNPROCESSABLE_ENTITY, "{body}"); + assert_eq!(st, StatusCode::CREATED, "{body}"); + let problem = &body["problems"][0]; + assert_eq!( + (&problem["stage"], &problem["field"]), + (&json!("source"), &json!("unauthenticated")), + "{body}" + ); assert!( - body["error"] + problem["message"] .as_str() .unwrap() .contains("UDP cannot carry TLS"), "{body}" ); - let (st, body) = call(&app, "POST", "/api/v1/sources", Some(accepted)).await; - assert_eq!(st, StatusCode::CREATED, "{body}"); + let (st, body) = call(&app, "POST", "/api/v1/sources/udp-auth/enable", None).await; + assert_eq!(st, StatusCode::UNPROCESSABLE_ENTITY, "{body}"); + let (st, body) = call(&app, "PUT", "/api/v1/sources/udp-auth", Some(accepted)).await; + assert_eq!( + (st, &body["draft"]), + (StatusCode::OK, &json!(false)), + "{body}" + ); // Moved to mutual TLS, the acceptance goes: a decision whose before // and after (the two revisions) show the change. spec["transport"] = json!({"type": "tcp_server", "bind": "127.0.0.1:0", @@ -944,9 +1034,10 @@ mod tests { .cloned() .unwrap() }; - assert_eq!(rev(1)["spec"]["unauthenticated"], "accepted", "{revs}"); - assert!(rev(2)["spec"].get("unauthenticated").is_none(), "{revs}"); - assert!(rev(2)["decision_id"].is_i64(), "{revs}"); + assert!(rev(1)["spec"].get("unauthenticated").is_none(), "{revs}"); + assert_eq!(rev(2)["spec"]["unauthenticated"], "accepted", "{revs}"); + assert!(rev(3)["spec"].get("unauthenticated").is_none(), "{revs}"); + assert!(rev(3)["decision_id"].is_i64(), "{revs}"); let (_, d) = call(&app, "GET", "/api/v1/decisions?op=update_source", None).await; assert_eq!(d["decisions"][0]["evidence"]["source"], "udp-auth", "{d}"); } @@ -959,22 +1050,51 @@ mod tests { }; let spec: Value = serde_json::from_str(ADSB).unwrap(); - // Its extension schema is not published yet. + // Its extension schema is not published yet: it saves, as a draft + // that cannot be enabled. let (st, body) = call(&app, "POST", "/api/v1/sources", Some(spec.clone())).await; - assert_eq!(st, StatusCode::UNPROCESSABLE_ENTITY); + assert_eq!(st, StatusCode::CREATED, "{body}"); + assert_eq!( + (body["revision"].as_i64(), body["enabled"].as_bool()), + (Some(1), Some(false)) + ); + assert_eq!(body["transport"], "http_poll"); + assert_eq!(body["draft"], true, "{body}"); + assert_eq!(body["problems"][0]["kind"], "incomplete", "{body}"); + assert_eq!(body["problems"][0]["stage"], "map", "{body}"); + assert!( + body["problems"][0]["message"] + .as_str() + .unwrap() + .contains("schema version 2"), + "{body}" + ); + let (st, body) = call(&app, "POST", "/api/v1/sources/adsb-lol/enable", None).await; + assert_eq!(st, StatusCode::UNPROCESSABLE_ENTITY, "{body}"); assert!( body["error"].as_str().unwrap().contains("schema version 2"), "{body}" ); + assert_eq!(body["problems"][0]["stage"], "map", "{body}"); + + // Once the schema is published, it may run as it is. publish_example_schema(&app).await; + let (st, body) = call(&app, "POST", "/api/v1/sources/adsb-lol/enable", None).await; + assert_eq!(st, StatusCode::OK, "{body}"); + assert_eq!( + (&body["enabled"], &body["draft"]), + (&json!(true), &json!(false)) + ); + let (st, _) = call(&app, "POST", "/api/v1/sources/adsb-lol/disable", None).await; + assert_eq!(st, StatusCode::OK); - let (st, body) = call(&app, "POST", "/api/v1/sources", Some(spec.clone())).await; - assert_eq!(st, StatusCode::CREATED, "{body}"); + let (st, body) = call(&app, "PUT", "/api/v1/sources/adsb-lol", Some(spec.clone())).await; + assert_eq!(st, StatusCode::OK, "{body}"); assert_eq!( - (body["revision"].as_i64(), body["enabled"].as_bool()), - (Some(1), Some(false)) + (body["revision"].as_i64(), body["draft"].as_bool()), + (Some(2), Some(false)) ); - assert_eq!(body["transport"], "http_poll"); + assert_eq!(body["problems"], json!([])); let (st, _) = call(&app, "POST", "/api/v1/sources", Some(spec.clone())).await; assert_eq!(st, StatusCode::CONFLICT); @@ -982,38 +1102,53 @@ mod tests { let (st, body) = call(&app, "PUT", "/api/v1/sources/other-id", Some(spec.clone())).await; assert_eq!(st, StatusCode::UNPROCESSABLE_ENTITY, "{body}"); + // A source with no name is not saved at all. + let mut nameless = spec.clone(); + nameless["name"] = json!(" "); + let (st, body) = call(&app, "PUT", "/api/v1/sources/adsb-lol", Some(nameless)).await; + assert_eq!(st, StatusCode::UNPROCESSABLE_ENTITY, "{body}"); + assert_eq!(body["problems"][0]["kind"], "invalid", "{body}"); + + // Off, a mistake leaves a draft, with every problem on its stage. let mut typo = spec.clone(); typo["pipeline"]["mapping"]["rules"][1]["fields"]["kinematic.speed_mps"] = json!("gs"); - let (st, body) = call(&app, "PUT", "/api/v1/sources/adsb-lol", Some(typo)).await; - assert_eq!(st, StatusCode::UNPROCESSABLE_ENTITY); + typo["pipeline"]["mapping"]["rules"][1]["fields"]["ext.tail_art"] = json!("r"); + let (st, body) = call(&app, "PUT", "/api/v1/sources/adsb-lol", Some(typo.clone())).await; + assert_eq!(st, StatusCode::OK, "{body}"); + assert_eq!(body["draft"], true, "{body}"); + let messages: Vec<&str> = body["problems"] + .as_array() + .unwrap() + .iter() + .inspect(|p| assert_eq!(p["stage"], "map", "{body}")) + .map(|p| p["message"].as_str().unwrap()) + .collect(); assert!( - body["error"] - .as_str() - .unwrap() - .contains("kinematic.speed_mps"), + messages.iter().any(|m| m.contains("kinematic.speed_mps")), "{body}" ); - - let mut undeclared = spec.clone(); - undeclared["pipeline"]["mapping"]["rules"][1]["fields"]["ext.tail_art"] = json!("r"); - let (st, body) = call(&app, "PUT", "/api/v1/sources/adsb-lol", Some(undeclared)).await; - assert_eq!(st, StatusCode::UNPROCESSABLE_ENTITY); assert!( - body["error"].as_str().unwrap().contains("ext.tail_art"), + messages.iter().any(|m| m.contains("ext.tail_art")), "{body}" ); let mut edited = spec.clone(); edited["name"] = json!("adsb.lol (edited)"); let (st, body) = call(&app, "PUT", "/api/v1/sources/adsb-lol", Some(edited)).await; - assert_eq!((st, body["revision"].as_i64()), (StatusCode::OK, Some(2))); + assert_eq!((st, body["revision"].as_i64()), (StatusCode::OK, Some(4))); + assert_eq!(body["draft"], false, "{body}"); let (_, body) = call(&app, "POST", "/api/v1/sources/adsb-lol/enable", None).await; assert_eq!(body["enabled"], true); + // On, the same mistake is refused, and the running source is kept. + let (st, body) = call(&app, "PUT", "/api/v1/sources/adsb-lol", Some(typo)).await; + assert_eq!(st, StatusCode::UNPROCESSABLE_ENTITY, "{body}"); + assert_eq!(body["problems"].as_array().unwrap().len(), 2, "{body}"); let (_, body) = call(&app, "GET", "/api/v1/sources/adsb-lol/revisions", None).await; - assert_eq!(body["revisions"].as_array().unwrap().len(), 2); + assert_eq!(body["revisions"].as_array().unwrap().len(), 4); let (_, body) = call(&app, "GET", "/api/v1/sources", None).await; assert_eq!(body["sources"][0]["name"], "adsb.lol (edited)"); + assert_eq!(body["sources"][0]["draft"], false); // Raw output needs explicit consent and a subject outside `tracks.>`. for bad in [ diff --git a/crates/ot-server/src/auth/api.rs b/crates/ot-server/src/auth/api.rs index 6da89ea..b84bc31 100644 --- a/crates/ot-server/src/auth/api.rs +++ b/crates/ot-server/src/auth/api.rs @@ -1364,6 +1364,7 @@ mod tests { codec: "json", priority: 0, spec: &spec, + problems: &serde_json::json!([]), }, "root@x.org", ) diff --git a/crates/ot-server/src/config_backup.rs b/crates/ot-server/src/config_backup.rs index c4e1c72..0d7efd0 100644 --- a/crates/ot-server/src/config_backup.rs +++ b/crates/ot-server/src/config_backup.rs @@ -296,11 +296,21 @@ fn check(common: &Common, f: &mut ConfigFile) -> Vec { p.push(format!("{at}: {e}")); continue; } + // A draft comes back a draft; an enabled source must be complete. let schema = schemas.get(&spec.pipeline.mapping.schema_version); - if let Err(e) = spec.validate_against(schema) { - p.push(format!("{at}: {e}")); + let problems = spec.problems(schema); + let refused: Vec<_> = problems + .iter() + .filter(|q| s.enabled || q.kind == ot_source::source::ProblemKind::Invalid) + .collect(); + if !refused.is_empty() { + p.extend(refused.iter().map(|q| format!("{at}: {}", q.message))); continue; } + match serde_json::to_value(&problems) { + Ok(v) => s.problems = v, + Err(e) => p.push(format!("{at}: {e}")), + } if let Some(sub) = &s.raw_subject && let Err(e) = crate::api::check_raw_subject(sub, &common.nats.tracks_subject) { @@ -817,11 +827,13 @@ pub(crate) mod tests { codec: "json", priority: 100, spec: &spec, + problems: &serde_json::json!([]), }, "op:test", ) .unwrap(); - db.set_source_enabled("aisstream", true, "op:test").unwrap(); + db.set_source_enabled("aisstream", true, None, "op:test") + .unwrap(); db.put_app_settings(&json!({"site_name": "Garden Island"}), "op:test") .unwrap(); // Saved before the account policy was fixed: those settings are diff --git a/crates/ot-server/src/control.rs b/crates/ot-server/src/control.rs index 2c3e0ad..b65f7dc 100644 --- a/crates/ot-server/src/control.rs +++ b/crates/ot-server/src/control.rs @@ -355,50 +355,52 @@ pub(crate) fn parse_uid(raw: &str) -> Result { pub struct ApiError { status: StatusCode, pub(crate) message: String, + /// What is wrong with a source, each problem with its pipeline stage + /// (`ot_source::source::Problem`), sent as the body's `problems`. + problems: Option, } impl ApiError { - pub(crate) fn internal(m: impl Into) -> Self { + fn new(status: StatusCode, m: impl Into) -> Self { Self { - status: StatusCode::INTERNAL_SERVER_ERROR, + status, message: m.into(), + problems: None, } } + + pub(crate) fn internal(m: impl Into) -> Self { + Self::new(StatusCode::INTERNAL_SERVER_ERROR, m) + } pub(crate) fn not_found(m: impl Into) -> Self { - Self { - status: StatusCode::NOT_FOUND, - message: m.into(), - } + Self::new(StatusCode::NOT_FOUND, m) } pub(crate) fn unprocessable(m: impl Into) -> Self { - Self { - status: StatusCode::UNPROCESSABLE_ENTITY, - message: m.into(), - } + Self::new(StatusCode::UNPROCESSABLE_ENTITY, m) } pub(crate) fn conflict(m: impl Into) -> Self { - Self { - status: StatusCode::CONFLICT, - message: m.into(), - } + Self::new(StatusCode::CONFLICT, m) } pub(crate) fn unauthorized(m: impl Into) -> Self { - Self { - status: StatusCode::UNAUTHORIZED, - message: m.into(), - } + Self::new(StatusCode::UNAUTHORIZED, m) } pub(crate) fn forbidden(m: impl Into) -> Self { + Self::new(StatusCode::FORBIDDEN, m) + } + /// A source that cannot be saved or enabled as it is: 422, with its + /// problems listed (the first is the `error`). + pub(crate) fn source_problems(problems: &[ot_source::source::Problem]) -> Self { Self { - status: StatusCode::FORBIDDEN, - message: m.into(), + status: StatusCode::UNPROCESSABLE_ENTITY, + message: problems + .first() + .map_or_else(|| "invalid source".to_owned(), |p| p.message.clone()), + problems: Some(serde_json::to_value(problems).unwrap_or_default()), } } + pub(crate) fn bad_request(m: impl Into) -> Self { - Self { - status: StatusCode::BAD_REQUEST, - message: m.into(), - } + Self::new(StatusCode::BAD_REQUEST, m) } } @@ -406,10 +408,7 @@ impl From for ApiError { fn from(e: ot_store::StoreError) -> Self { match e { ot_store::StoreError::NotFound(m) => Self::not_found(m), - ot_store::StoreError::Conflict(m) => Self { - status: StatusCode::CONFLICT, - message: m, - }, + ot_store::StoreError::Conflict(m) => Self::conflict(m), other => Self::internal(other.to_string()), } } @@ -430,7 +429,14 @@ impl IntoResponse for ApiError { ) .into_response(); } - (self.status, Json(json!({ "error": self.message }))).into_response() + match self.problems { + Some(problems) => ( + self.status, + Json(json!({ "error": self.message, "problems": problems })), + ) + .into_response(), + None => (self.status, Json(json!({ "error": self.message }))).into_response(), + } } } diff --git a/crates/ot-server/src/engine/bench.rs b/crates/ot-server/src/engine/bench.rs index 217eed7..636e76c 100644 --- a/crates/ot-server/src/engine/bench.rs +++ b/crates/ot-server/src/engine/bench.rs @@ -237,10 +237,11 @@ pub async fn run(common: &Common, opts: BenchOptions) -> anyhow::Result<()> { codec: spec.pipeline.codec.name(), priority: spec.priority, spec: raw, + problems: &serde_json::json!([]), }, "bench", )?; - db.set_source_enabled(&spec.id, true, "bench")?; + db.set_source_enabled(&spec.id, true, None, "bench")?; } } let mut settings = EngineSettings::default(); diff --git a/crates/ot-server/src/sources.rs b/crates/ot-server/src/sources.rs index 1a6a31c..6f20055 100644 --- a/crates/ot-server/src/sources.rs +++ b/crates/ot-server/src/sources.rs @@ -624,6 +624,8 @@ mod tests { raw_subject: None, created_at_ms: 0, updated_at_ms: 0, + problems: serde_json::json!([]), + draft: false, }; let schema = ExtensionSchema { version: accepted["pipeline"]["mapping"]["schema_version"] diff --git a/crates/ot-source/src/mapping.rs b/crates/ot-source/src/mapping.rs index 7c61110..ea8c32a 100644 --- a/crates/ot-source/src/mapping.rs +++ b/crates/ot-source/src/mapping.rs @@ -209,17 +209,26 @@ pub enum MapOutcome { impl MappingSpec { pub fn validate(&self) -> Result<(), MappingError> { + match self.problems().into_iter().next() { + Some(e) => Err(e), + None => Ok(()), + } + } + + /// Every problem with the rules, in rule order. + pub fn problems(&self) -> Vec { if self.rules.is_empty() { - return Err(MappingError::NoRules); + return vec![MappingError::NoRules]; } + let mut out = Vec::new(); let mut names = std::collections::HashSet::new(); for rule in &self.rules { if !names.insert(rule.name.as_str()) { - return Err(MappingError::DuplicateRule(rule.name.clone())); + out.push(MappingError::DuplicateRule(rule.name.clone())); } for field in rule.fields.keys() { if target_type(field).is_none() { - return Err(MappingError::UnknownField { + out.push(MappingError::UnknownField { rule: rule.name.clone(), field: field.clone(), }); @@ -230,10 +239,10 @@ impl MappingSpec { && !(rule.fields.contains_key("position.latitude") && rule.fields.contains_key("position.longitude")) { - return Err(MappingError::NoPosition(rule.name.clone())); + out.push(MappingError::NoPosition(rule.name.clone())); } } - Ok(()) + out } pub fn apply(&self, record: &Value) -> MapOutcome { diff --git a/crates/ot-source/src/source.rs b/crates/ot-source/src/source.rs index 8d5e3b8..751770c 100644 --- a/crates/ot-source/src/source.rs +++ b/crates/ot-source/src/source.rs @@ -224,12 +224,60 @@ impl SourceSpec { } } + /// The first reason this source may not run (without the extension + /// schema checks): see [`Self::problems`] for all of them. pub fn validate(&self) -> Result<(), SourceError> { - if let Some(label) = &self.security { - label.validate().map_err(SourceError::Tracker)?; + match self.checks(None).into_iter().next() { + Some((_, _, e)) => Err(e), + None => Ok(()), } - if let Some(m) = &self.emitter_motion { - m.check().map_err(SourceError::Tracker)?; + } + + /// Everything that stops this source running, each with the pipeline + /// stage (and, where there is one, the setting) it belongs to. Empty: + /// the source is complete and may be enabled. A problem of kind + /// [`ProblemKind::Invalid`] also stops it being saved; the rest leave + /// it a draft. + pub fn problems(&self, schema: Option<&crate::schema::ExtensionSchema>) -> Vec { + self.checks(Some(schema)) + .into_iter() + .map(|(stage, field, e)| Problem { + kind: match e { + SourceError::BadId(_) | SourceError::NoName => ProblemKind::Invalid, + _ => ProblemKind::Incomplete, + }, + stage, + field: field.map(str::to_owned), + message: e.to_string(), + }) + .collect() + } + + /// Every check, in order, each failure with its stage and setting. + /// `schema`: `None` skips the extension schema checks; `Some(None)` + /// means the mapping's schema version is not published. + fn checks( + &self, + schema: Option>, + ) -> Vec<(Option, Option<&'static str>, SourceError)> { + let mut out = Vec::new(); + if let Some(label) = &self.security + && let Err(e) = label.validate() + { + out.push(( + Some(Stage::Output), + Some("security"), + SourceError::Tracker(e), + )); + } + if let Some(m) = &self.emitter_motion + && let Err(e) = m.check() + { + out.push(( + Some(Stage::Output), + Some("emitter_motion"), + SourceError::Tracker(e), + )); } let id_ok = !self.id.is_empty() && self.id.len() <= 64 @@ -238,86 +286,166 @@ impl SourceSpec { .bytes() .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-' || b == b'_'); if !id_ok { - return Err(SourceError::BadId(self.id.clone())); + out.push((None, Some("id"), SourceError::BadId(self.id.clone()))); } if self.name.trim().is_empty() { - return Err(SourceError::NoName); + out.push((None, Some("name"), SourceError::NoName)); } - self.pipeline.mapping.validate()?; - if let Some(t) = &self.pipeline.tracker { - t.validate().map_err(SourceError::Tracker)?; + for e in self.pipeline.mapping.problems() { + out.push((Some(Stage::Map), None, e.into())); } - if let crate::codec::CodecConfig::Plugin { .. } - | crate::codec::CodecConfig::Protobuf { .. } = &self.pipeline.codec + if let Some(t) = &self.pipeline.tracker + && let Err(e) = t.validate() { - crate::codec::Codec::new(self.pipeline.codec.clone()) - .map_err(|e| SourceError::Tracker(e.to_string()))?; + out.push((Some(Stage::Tracker), None, SourceError::Tracker(e))); } - self.transport.check().map_err(SourceError::Transport)?; - // gRPC carries protobuf: the method must fit the codec's message. - if let TransportConfig::GrpcClient { .. } | TransportConfig::GrpcServer { .. } = - &self.transport + let grpc = matches!( + self.transport, + TransportConfig::GrpcClient { .. } | TransportConfig::GrpcServer { .. } + ); + let protobuf = matches!( + self.pipeline.codec, + crate::codec::CodecConfig::Protobuf { .. } + ); + let plugin = matches!( + self.pipeline.codec, + crate::codec::CodecConfig::Plugin { .. } + ); + // A gRPC source's protobuf is compiled (once) by the check below. + if (plugin || (protobuf && !grpc)) + && let Err(e) = crate::codec::Codec::new(self.pipeline.codec.clone()) { - let proto = crate::transport::ProtoContext::of(&self.pipeline.codec) - .map_err(SourceError::Transport)? - .ok_or_else(|| { - SourceError::Transport("a gRPC source decodes with the protobuf codec".into()) - })?; - match &self.transport { - TransportConfig::GrpcClient { - method, request, .. - } => { - proto - .client_request(method, request) - .map_err(SourceError::Transport)?; - } - TransportConfig::GrpcServer { methods, .. } => { - proto - .server_methods(methods) - .map_err(SourceError::Transport)?; + out.push(( + Some(Stage::Decode), + None, + SourceError::Tracker(e.to_string()), + )); + } + let missing = self.transport.missing(); + for (field, what) in &missing { + out.push(( + Some(Stage::Source), + Some(*field), + SourceError::Transport(format!("{what} is not set")), + )); + } + if let Err(e) = self.transport.check() { + out.push((Some(Stage::Source), None, SourceError::Transport(e))); + } + // gRPC carries protobuf: the method must fit the codec's message. + if grpc { + match crate::transport::ProtoContext::of(&self.pipeline.codec) { + Err(e) => out.push((Some(Stage::Decode), None, SourceError::Transport(e))), + Ok(None) => out.push(( + Some(Stage::Decode), + None, + SourceError::Transport("a gRPC source decodes with the protobuf codec".into()), + )), + Ok(Some(_)) if !missing.is_empty() => {} + Ok(Some(proto)) => { + let fit = match &self.transport { + TransportConfig::GrpcClient { + method, request, .. + } => proto.client_request(method, request).map(drop), + TransportConfig::GrpcServer { methods, .. } => { + proto.server_methods(methods).map(drop) + } + _ => Ok(()), + }; + if let Err(e) = fit { + out.push((Some(Stage::Source), None, SourceError::Transport(e))); + } } - _ => {} } } if let TransportConfig::TcpClient { framing, .. } | TransportConfig::TcpServer { framing, .. } = &self.transport + && let Err(e) = crate::frame::Framer::new(framing.clone()) { - crate::frame::Framer::new(framing.clone())?; + out.push((Some(Stage::Source), Some("framing"), e.into())); } // Last, so that a broken setting is reported before its security. if let Some(e) = self.authentication_problem() { - return Err(SourceError::Unauthenticated(e)); + out.push(( + Some(Stage::Source), + Some("unauthenticated"), + SourceError::Unauthenticated(e), + )); } - Ok(()) - } - - /// Validate against the extension schema version the mapping targets - /// (`None` if that version is not published). - pub fn validate_against( - &self, - schema: Option<&crate::schema::ExtensionSchema>, - ) -> Result<(), SourceError> { - self.validate()?; - let wanted = self.pipeline.mapping.schema_version; - let Some(schema) = schema.filter(|s| s.version == wanted) else { - return Err(SourceError::SchemaVersion { - wanted, - found: "not published".into(), - }); - }; - for (key, rules) in crate::schema::mapped_ext_keys(&self.pipeline.mapping) { - if schema.field(&key).is_none() { - return Err(SourceError::UndeclaredExtension { - rule: rules[0].clone(), - key, - version: wanted, - }); + if let Some(schema) = schema { + let wanted = self.pipeline.mapping.schema_version; + match schema.filter(|s| s.version == wanted) { + None => out.push(( + Some(Stage::Map), + Some("schema_version"), + SourceError::SchemaVersion { + wanted, + found: "not published".into(), + }, + )), + Some(schema) => { + for (key, rules) in crate::schema::mapped_ext_keys(&self.pipeline.mapping) { + if schema.field(&key).is_none() { + out.push(( + Some(Stage::Map), + None, + SourceError::UndeclaredExtension { + rule: rules[0].clone(), + key, + version: wanted, + }, + )); + } + } + } } } - Ok(()) + out } } +/// One reason a source may not run (see [`SourceSpec::problems`]). +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct Problem { + pub kind: ProblemKind, + /// The pipeline stage it belongs to; `None` for the source as a whole + /// (its id and name). + pub stage: Option, + /// The setting, where there is one (for example `host`). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub field: Option, + pub message: String, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ProblemKind { + /// The source cannot be saved. + Invalid, + /// The source saves as a draft, which cannot be enabled. + Incomplete, +} + +/// A source's pipeline stages, in the order a frame meets them. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum Stage { + /// The transport and its framing. + Source, + /// The codec. + Decode, + Map, + StaticJoin, + Registry, + Affiliation, + Filter, + Tracker, + Throttle, + /// What the source reports and how: tracks or detections, security + /// label, priority. + Output, +} + #[cfg(test)] mod tests { use super::*; @@ -376,6 +504,71 @@ mod tests { assert!(chosen.publishes_alone()); } + #[test] + fn problems_name_their_stage_and_a_draft_is_only_incomplete() { + let ok: SourceSpec = serde_json::from_value(spec("cot")).unwrap(); + assert_eq!( + ok.problems(None), + vec![Problem { + kind: ProblemKind::Incomplete, + stage: Some(Stage::Map), + field: Some("schema_version".into()), + message: "mapping targets schema version 1, but it is not published".into(), + }] + ); + + // A new source: the settings left empty, no rules yet. + let draft: SourceSpec = serde_json::from_value(json!({ + "id": "radar", "name": "Radar", + "transport": { "type": "tcp_client", "host": " ", "port": 0 }, + "pipeline": { "codec": { "type": "json" }, "mapping": { "rules": [] } } + })) + .unwrap(); + let found: Vec<_> = draft + .problems(None) + .into_iter() + .map(|p| (p.kind, p.stage, p.field)) + .collect(); + let incomplete = ProblemKind::Incomplete; + assert_eq!( + found, + vec![ + (incomplete, Some(Stage::Map), None), + (incomplete, Some(Stage::Source), Some("host".into())), + (incomplete, Some(Stage::Source), Some("port".into())), + (incomplete, Some(Stage::Map), Some("schema_version".into())), + ] + ); + assert!(matches!(draft.validate(), Err(SourceError::Mapping(_)))); + + // Every rule's problems, not just the first. + let mut v = spec("cot"); + v["name"] = json!(""); + v["pipeline"]["mapping"]["rules"] = json!([ + { "name": "a", "key": "k", "fields": { "nope": "x" } }, + { "name": "a", "key": "k", "fields": {} } + ]); + let s: SourceSpec = serde_json::from_value(v).unwrap(); + let found: Vec<_> = s + .problems(None) + .into_iter() + .map(|p| (p.kind, p.stage, p.message)) + .collect(); + assert_eq!( + found[0], + ( + ProblemKind::Invalid, + None, + "source name must not be empty".into() + ) + ); + assert_eq!( + found.iter().filter(|p| p.1 == Some(Stage::Map)).count(), + 5, + "unknown field, no position (twice), duplicate rule, schema: {found:?}" + ); + } + #[test] fn ids_are_restricted() { let ok: SourceSpec = serde_json::from_value(spec("cot-udp_1")).unwrap(); diff --git a/crates/ot-source/src/transport.rs b/crates/ot-source/src/transport.rs index 20a76c7..b1d46a6 100644 --- a/crates/ot-source/src/transport.rs +++ b/crates/ot-source/src/transport.rs @@ -379,6 +379,57 @@ impl TransportConfig { } } + /// The settings a source needs that are empty, as (setting, what it + /// is): a draft may leave them so, an enabled source may not. + pub fn missing(&self) -> Vec<(&'static str, &'static str)> { + let blank = |s: &str| s.trim().is_empty(); + let mut out = Vec::new(); + match self { + TransportConfig::TcpClient { host, port, .. } => { + if blank(host) { + out.push(("host", "the host")); + } + if *port == 0 { + out.push(("port", "the port")); + } + } + TransportConfig::TcpServer { bind, .. } + | TransportConfig::Udp { bind, .. } + | TransportConfig::GrpcServer { bind, .. } => { + if blank(bind) { + out.push(("bind", "the address to listen on")); + } + } + TransportConfig::HttpPoll { url, .. } | TransportConfig::Websocket { url, .. } => { + if blank(url) { + out.push(("url", "the URL")); + } + } + TransportConfig::Mqtt { url, topics, .. } => { + if blank(url) { + out.push(("url", "the broker URL")); + } + if topics.iter().all(|t| blank(t)) { + out.push(("topics", "a topic")); + } + } + TransportConfig::GrpcClient { url, method, .. } => { + if blank(url) { + out.push(("url", "the URL")); + } + if blank(method) { + out.push(("method", "the method")); + } + } + TransportConfig::File { path, .. } => { + if blank(path) { + out.push(("path", "the path")); + } + } + } + out + } + /// Settings that cannot be caught by the type alone. pub fn check(&self) -> Result<(), String> { match self { diff --git a/crates/ot-store/migrations/0020_source_problems.sql b/crates/ot-store/migrations/0020_source_problems.sql new file mode 100644 index 0000000..73eb9d6 --- /dev/null +++ b/crates/ot-store/migrations/0020_source_problems.sql @@ -0,0 +1,5 @@ +-- Why a source may not run yet, as a JSON array of problems ({kind, stage, +-- field?, message}), worked out by the server each time the source is saved. +-- Empty: complete. Otherwise the source is a draft, which cannot be enabled. +-- Every source saved before drafts was complete. +ALTER TABLE sources ADD COLUMN problems TEXT NOT NULL DEFAULT '[]'; diff --git a/crates/ot-store/src/backup.rs b/crates/ot-store/src/backup.rs index 7956e9d..4cb6380 100644 --- a/crates/ot-store/src/backup.rs +++ b/crates/ot-store/src/backup.rs @@ -92,12 +92,24 @@ pub struct SourceEntry { pub created_at_ms: i64, #[serde(default)] pub updated_at_ms: i64, + /// Why it may not run yet (a draft); empty when it is complete. Files + /// from before drafts have none: every source was complete. + #[serde(default = "no_problems", skip_serializing_if = "is_empty_array")] + pub problems: Value, } fn one() -> i64 { 1 } +fn no_problems() -> Value { + Value::Array(Vec::new()) +} + +fn is_empty_array(v: &Value) -> bool { + v.as_array().is_some_and(Vec::is_empty) +} + /// An output schema version. #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] @@ -372,6 +384,7 @@ impl Db { raw_subject: r.raw_subject, created_at_ms: r.created_at_ms, updated_at_ms: r.updated_at_ms, + problems: r.problems, }) .collect(); let schema_versions = self @@ -652,8 +665,8 @@ impl Db { let revision = s.revision.max(1); tx.execute( "INSERT INTO sources (id, name, transport, codec, settings, enabled, priority, - raw_subject, raw_consent, revision, created_at_ms, updated_at_ms) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)", + raw_subject, raw_consent, revision, created_at_ms, updated_at_ms, problems) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)", params![ s.id, s.name, @@ -667,6 +680,11 @@ impl Db { revision, if s.created_at_ms > 0 { s.created_at_ms } else { now }, now, + if s.problems.is_array() { + s.problems.to_string() + } else { + "[]".to_owned() + }, ], )?; tx.execute( @@ -757,11 +775,12 @@ mod tests { codec: "json", priority: 50, spec: &spec, + problems: &serde_json::json!([]), }, "op:test", ) .unwrap(); - db.set_source_enabled("ais", true, "op:test").unwrap(); + db.set_source_enabled("ais", true, None, "op:test").unwrap(); db.set_raw_output("ais", Some("opentrack.raw.ais"), "op:test") .unwrap(); db.put_schema_draft(&[json!({"key": "hull", "type": "string"})], None, "op:test") @@ -888,6 +907,31 @@ mod tests { assert!(ops.contains(&"import_config".to_owned()), "{ops:?}"); } + #[test] + fn a_draft_source_comes_back_a_draft() { + let mut snap = configured().config_snapshot().unwrap(); + let problems = json!([{"kind": "incomplete", "stage": "source", "field": "url", + "message": "the URL is not set"}]); + snap.sources[0].enabled = false; + snap.sources[0].problems = problems.clone(); + // The file says so, and a file from before drafts reads as complete. + let file = serde_json::to_value(&snap).unwrap(); + assert_eq!(file["sources"][0]["problems"], problems); + let mut old = file.clone(); + old["sources"][0] + .as_object_mut() + .unwrap() + .remove("problems"); + let old: ConfigFile = serde_json::from_value(old).unwrap(); + assert_eq!(old.sources[0].problems, json!([])); + + let mut fresh = Db::open_in_memory().unwrap(); + fresh.restore_config(&snap, "cli", || Ok(())).unwrap(); + let row = fresh.get_source("ais").unwrap().unwrap(); + assert!(row.draft && !row.enabled); + assert_eq!(row.problems, problems); + } + #[test] fn a_file_without_accounts_keeps_the_first_admin() { let mut snap = configured().config_snapshot().unwrap(); diff --git a/crates/ot-store/src/sources.rs b/crates/ot-store/src/sources.rs index 97ce303..ba263a9 100644 --- a/crates/ot-store/src/sources.rs +++ b/crates/ot-store/src/sources.rs @@ -21,6 +21,11 @@ pub struct SourceRow { pub raw_subject: Option, pub created_at_ms: i64, pub updated_at_ms: i64, + /// Why the source may not run yet, as the server found when it was + /// saved: a JSON array, empty when it is complete. + pub problems: Value, + /// Whether it has problems: a draft cannot be enabled. + pub draft: bool, } #[derive(Debug, Clone, PartialEq, Serialize)] @@ -40,11 +45,13 @@ pub struct SourceWrite<'a> { pub codec: &'a str, pub priority: i64, pub spec: &'a Value, + /// Its problems (a JSON array; empty: complete). + pub problems: &'a Value, } -const COLUMNS: &str = "id, name, transport, codec, enabled, priority, revision, settings, raw_subject, created_at_ms, updated_at_ms"; +const COLUMNS: &str = "id, name, transport, codec, enabled, priority, revision, settings, raw_subject, created_at_ms, updated_at_ms, problems"; -fn row(r: &rusqlite::Row<'_>) -> rusqlite::Result<(SourceRow, String)> { +fn row(r: &rusqlite::Row<'_>) -> rusqlite::Result<(SourceRow, String, String)> { let spec: String = r.get(7)?; Ok(( SourceRow { @@ -59,16 +66,38 @@ fn row(r: &rusqlite::Row<'_>) -> rusqlite::Result<(SourceRow, String)> { raw_subject: r.get(8)?, created_at_ms: r.get(9)?, updated_at_ms: r.get(10)?, + problems: Value::Null, + draft: false, }, spec, + r.get(11)?, )) } -fn finish((mut row, spec): (SourceRow, String)) -> Result { +fn finish((mut row, spec, problems): (SourceRow, String, String)) -> Result { row.spec = serde_json::from_str(&spec)?; + row.problems = serde_json::from_str(&problems)?; + row.draft = is_draft(&row.problems); Ok(row) } +/// Whether a source with these problems is a draft: any problem listed. +fn is_draft(problems: &Value) -> bool { + problems.as_array().is_some_and(|a| !a.is_empty()) +} + +/// A source's stored problems, if it exists. +fn stored_problems(tx: &rusqlite::Transaction<'_>, id: &str) -> Result> { + tx.query_row( + "SELECT enabled, problems FROM sources WHERE id = ?1", + [id], + |r| Ok((r.get::<_, bool>(0)?, r.get::<_, String>(1)?)), + ) + .optional()? + .map(|(enabled, p)| Ok((enabled, serde_json::from_str(&p)?))) + .transpose() +} + impl Db { pub fn list_sources(&self) -> Result> { let mut stmt = self @@ -90,10 +119,19 @@ impl Db { .transpose() } - /// Create or update a source. Returns the saved row. + /// Create or update a source. Returns the saved row. An enabled source + /// is saved only complete (an edit never stops a running source). pub fn put_source(&mut self, w: &SourceWrite<'_>, actor: &str) -> Result { let before = self.get_source(w.id)?; self.write(|tx| { + if is_draft(w.problems) + && stored_problems(tx, w.id)?.is_some_and(|(enabled, _)| enabled) + { + return Err(StoreError::Conflict(format!( + "source {} is enabled; it can be saved only complete", + w.id + ))); + } let now = now_ms(); let (op, revision) = match &before { None => ("create_source", 1), @@ -110,19 +148,23 @@ impl Db { }; let decision_id = record_decision(tx, &decision, now)?; let spec = w.spec.to_string(); + let problems = w.problems.to_string(); if before.is_none() { tx.execute( "INSERT INTO sources (id, name, transport, codec, settings, priority, revision, - created_at_ms, updated_at_ms) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, 1, ?7, ?7)", - params![w.id, w.name, w.transport, w.codec, spec, w.priority, now], + created_at_ms, updated_at_ms, problems) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, 1, ?7, ?7, ?8)", + params![w.id, w.name, w.transport, w.codec, spec, w.priority, now, problems], )?; } else { tx.execute( "UPDATE sources SET name = ?2, transport = ?3, codec = ?4, settings = ?5, - priority = ?6, revision = ?7, updated_at_ms = ?8 + priority = ?6, revision = ?7, updated_at_ms = ?8, problems = ?9 WHERE id = ?1", - params![w.id, w.name, w.transport, w.codec, spec, w.priority, revision, now], + params![ + w.id, w.name, w.transport, w.codec, spec, w.priority, revision, now, + problems + ], )?; } tx.execute( @@ -136,14 +178,32 @@ impl Db { .ok_or_else(|| StoreError::NotFound(format!("source {}", w.id))) } + /// Enable or disable a source. `problems`, when given, is what checking + /// it again found now (a schema published since it was saved, say), + /// stored first: a draft cannot be enabled. pub fn set_source_enabled( &mut self, id: &str, enabled: bool, + problems: Option<&Value>, actor: &str, ) -> Result { self.write(|tx| { let now = now_ms(); + let Some((_, stored)) = stored_problems(tx, id)? else { + return Err(StoreError::NotFound(format!("source {id}"))); + }; + if let Some(p) = problems { + tx.execute( + "UPDATE sources SET problems = ?2 WHERE id = ?1", + params![id, p.to_string()], + )?; + } + if enabled && is_draft(problems.unwrap_or(&stored)) { + return Err(StoreError::Conflict(format!( + "source {id} is a draft; complete it before enabling it" + ))); + } let op = if enabled { "enable_source" } else { @@ -151,19 +211,31 @@ impl Db { }; let d = Decision::new(actor, op).evidence(json!({ "source": id })); record_decision(tx, &d, now)?; - let n = tx.execute( + tx.execute( "UPDATE sources SET enabled = ?2, updated_at_ms = ?3 WHERE id = ?1", params![id, enabled, now], )?; - if n == 0 { - return Err(StoreError::NotFound(format!("source {id}"))); - } Ok(()) })?; self.get_source(id)? .ok_or_else(|| StoreError::NotFound(format!("source {id}"))) } + /// Store what checking a source again found, without a new revision + /// (nothing about the source itself changed). + pub fn set_source_problems(&mut self, id: &str, problems: &Value) -> Result<()> { + self.write(|tx| { + let n = tx.execute( + "UPDATE sources SET problems = ?2 WHERE id = ?1", + params![id, problems.to_string()], + )?; + if n == 0 { + return Err(StoreError::NotFound(format!("source {id}"))); + } + Ok(()) + }) + } + /// Set or clear the raw-output subject. Setting one records the admin's /// consent to publish this source's raw tracks. pub fn set_raw_output( @@ -255,6 +327,8 @@ fn masked(spec: &Value) -> Value { mod tests { use super::*; + const COMPLETE: &Value = &Value::Array(Vec::new()); + fn write<'a>(spec: &'a Value) -> SourceWrite<'a> { SourceWrite { id: "ais", @@ -263,9 +337,45 @@ mod tests { codec: "json", priority: 100, spec, + problems: COMPLETE, } } + #[test] + fn a_draft_keeps_its_problems_and_cannot_be_enabled() { + let mut db = Db::open_in_memory().unwrap(); + let spec = json!({"v": 1}); + let problems = json!([{"kind": "incomplete", "stage": "source", "field": "url", + "message": "the URL is not set"}]); + let row = db + .put_source( + &SourceWrite { + problems: &problems, + ..write(&spec) + }, + "op:test", + ) + .unwrap(); + assert!(row.draft); + assert_eq!(row.problems, problems); + assert!(matches!( + db.set_source_enabled("ais", true, None, "op:test"), + Err(StoreError::Conflict(_)) + )); + assert!(!db.get_source("ais").unwrap().unwrap().enabled); + // Turning a draft off is fine; completing it lets it run. + db.set_source_enabled("ais", false, None, "op:test") + .unwrap(); + let row = db.put_source(&write(&spec), "op:test").unwrap(); + assert!(!row.draft); + assert_eq!(row.problems, json!([])); + assert!( + db.set_source_enabled("ais", true, None, "op:test") + .unwrap() + .enabled + ); + } + #[test] fn decisions_hide_a_sources_secrets_and_the_revision_keeps_them() { let mut db = Db::open_in_memory().unwrap(); @@ -318,7 +428,7 @@ mod tests { let v1 = db.sources_version().unwrap(); assert_ne!(v0, v1); - let row = db.set_source_enabled("ais", true, "op:test").unwrap(); + let row = db.set_source_enabled("ais", true, None, "op:test").unwrap(); assert!(row.enabled); let revs = db.source_revisions("ais").unwrap(); assert_eq!(revs.len(), 2); @@ -351,7 +461,7 @@ mod tests { ] ); assert!(matches!( - db.set_source_enabled("nope", true, "x"), + db.set_source_enabled("nope", true, None, "x"), Err(StoreError::NotFound(_)) )); } diff --git a/crates/ot-store/src/sqlite.rs b/crates/ot-store/src/sqlite.rs index d0a4ac6..0460c07 100644 --- a/crates/ot-store/src/sqlite.rs +++ b/crates/ot-store/src/sqlite.rs @@ -30,6 +30,7 @@ const MIGRATIONS: &[&str] = &[ include_str!("../migrations/0017_account_expiry.sql"), include_str!("../migrations/0018_sync_keys.sql"), include_str!("../migrations/0019_plugin_secret.sql"), + include_str!("../migrations/0020_source_problems.sql"), ]; #[derive(Debug, thiserror::Error)] @@ -567,9 +568,9 @@ mod tests { fn migrates_once_and_reopens() { let dir = tempfile::tempdir().unwrap(); let path = dir.path().join("ot.db"); - assert_eq!(Db::open(&path).unwrap().schema_version().unwrap(), 19); + assert_eq!(Db::open(&path).unwrap().schema_version().unwrap(), 20); // Re-opening applies nothing and keeps the version. - assert_eq!(Db::open(&path).unwrap().schema_version().unwrap(), 19); + assert_eq!(Db::open(&path).unwrap().schema_version().unwrap(), 20); } #[test] diff --git a/docs/reference.md b/docs/reference.md index 383c25b..5a00591 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -160,6 +160,14 @@ A source is a transport (`tcp_client`, `tcp_server`, `udp` with IPv4 or IPv6 mul `_frame` (an MQTT topic is `_frame.topic`, `_frame.topic_levels[1]` its second level). Secrets are written as `${env:NAME}` and resolved when the source starts. +A source that is not complete yet (a setting left empty, no mapping rules, a mapping problem, an +unpublished schema version) saves as a **draft**: its row has `"draft": true` and `problems`, each +`{kind, stage, field?, message}`, where `stage` is the pipeline stage it belongs to (`source`, +`decode`, `map`, `static_join`, `registry`, `affiliation`, `filter`, `tracker`, `throttle`, +`output`) and `null` for the source as a whole. A draft cannot be enabled. A source without a valid +id or name is not saved (`kind: "invalid"`), and an enabled source is saved only complete. A +refusal is a 422 whose body lists the `problems` beside the `error`. + `tcp_client`, `http_poll` (`https://`), `websocket` (`wss://`) and `mqtt` (`mqtts://`) take a `tls` object: `ca_file` (trusted instead of the system roots), `cert_file` and `key_file` (a client certificate for mutual TLS), `server_name` (verify against this name instead of the host) and diff --git a/ui/src/api/client.ts b/ui/src/api/client.ts index 30ceab7..b773ef6 100644 --- a/ui/src/api/client.ts +++ b/ui/src/api/client.ts @@ -388,9 +388,12 @@ export interface ConfigImported { export class ApiError extends Error { readonly status: number - constructor(status: number, message: string) { + /** For a source that cannot be saved or enabled: everything wrong with it, by stage. */ + readonly problems: SourceProblem[] + constructor(status: number, message: string, problems: SourceProblem[] = []) { super(message) this.status = status + this.problems = problems } } @@ -423,7 +426,7 @@ async function request(method: string, path: string, body?: unknown): Promise }) const text = await res.text() const parsed = text ? JSON.parse(text) : {} - if (!res.ok) throw new ApiError(res.status, parsed.error ?? res.statusText) + if (!res.ok) throw new ApiError(res.status, parsed.error ?? res.statusText, parsed.problems ?? []) return parsed as T } @@ -902,10 +905,38 @@ export interface SourceRow { raw_subject: string | null created_at_ms: number updated_at_ms: number + /** Why it may not run yet, as found when it was saved; empty when complete. */ + problems: SourceProblem[] + /** Whether it has problems: a draft cannot be enabled. */ + draft: boolean /** Live worker status; null when no worker runs it (disabled or stopped). */ status: SourceStatus | null } +/** A pipeline stage, in the order a frame meets them. */ +export type SourceStage = + | 'source' + | 'decode' + | 'map' + | 'static_join' + | 'registry' + | 'affiliation' + | 'filter' + | 'tracker' + | 'throttle' + | 'output' + +/** One reason a source may not run. */ +export interface SourceProblem { + /** `invalid`: the source cannot be saved. `incomplete`: it saves as a draft. */ + kind: 'invalid' | 'incomplete' + /** The stage it belongs to; null for the source as a whole (its id and name). */ + stage: SourceStage | null + /** The setting, where there is one (for example `host`). */ + field?: string + message: string +} + export interface MetricsResponse { source: string minutes: number diff --git a/ui/src/pages/sources/AddSourceWizard.tsx b/ui/src/pages/sources/AddSourceWizard.tsx index 30c7edf..d7195d1 100644 --- a/ui/src/pages/sources/AddSourceWizard.tsx +++ b/ui/src/pages/sources/AddSourceWizard.tsx @@ -122,11 +122,20 @@ export function AddSourceWizard({ onDone, onCancel }: { onDone: (id: string) => } } + // An incomplete source is saved as a draft, which cannot be enabled: once + // it exists, Save updates it instead of creating it again. + const [created, setCreated] = useState(false) + const save = async () => { setSaving(true) setSaveError(null) try { - await api.createSource(spec) + if (created) { + await api.saveSource(spec) + } else { + await api.createSource(spec) + setCreated(true) + } if (enable) await api.setEnabled(spec.id, true) toast({ variant: 'success', message: `Source ${spec.id} saved${enable ? ' and enabled' : ''}.` }) onDone(spec.id)