Skip to main content

max / makenotwork

43.6 KB · 1209 lines History Blame Raw
1 //! HTTP + WS contract (mirrors Sando's shape).
2 //!
3 //! - `GET /state` — latest build + its target x step matrix (loose JSON).
4 //! - `POST /build` — `{app, version?, targets?[]}` kick a build.
5 //! - `POST /retry` — `{app, target, version?}` re-run one target.
6 //! - `GET /logs/{app}/{version}/{target}/{step}` — post-mortem step log.
7 //! - `GET /events` — WS; broadcasts `EventEnvelope` JSON frames.
8 //! - `GET /metrics` — Prometheus.
9
10 use crate::domain::{AppId, Target};
11 use crate::error::{Error, Result};
12 use crate::runner;
13 use crate::state::AppState;
14 use axum::extract::{Path, State, WebSocketUpgrade};
15 use axum::response::IntoResponse;
16 use axum::routing::{get, post};
17 use axum::{Json, Router};
18 use serde::{Deserialize, Serialize};
19 use sqlx::Row;
20
21 pub fn router(state: AppState) -> Router {
22 let prom = state.prom.clone();
23 let token = state.api_token.clone();
24
25 // Build triggers require a bearer token (when configured); reads stay open
26 // so the TUI's state/event/log polling needs no credential (CF2).
27 let mutating = Router::new()
28 .route("/build", post(build))
29 .route("/retry", post(retry))
30 .route_layer(axum::middleware::from_fn(move |req, next| {
31 require_bearer(token.clone(), req, next)
32 }));
33
34 let open = Router::new()
35 .route("/state", get(get_state))
36 .route("/status.json", get(get_status_json))
37 .route("/release/{app}/{version}", get(get_release))
38 .route("/logs/{app}/{version}/{target}/{step}", get(get_step_log))
39 .route("/events", get(events_ws));
40
41 Router::new()
42 .merge(mutating)
43 .merge(open)
44 .with_state(state)
45 // `/metrics` is intentionally open (Prometheus scrape convention): it
46 // exposes only build/host telemetry — counts, durations, app/host names —
47 // and never a secret or token (no secret reaches a metric label). The
48 // daemon binds the tailnet only, so the scrape surface is tailnet-internal.
49 .route("/metrics", get(crate::metrics::render).with_state(prom))
50 // Build/retry bodies are small JSON; cap request bodies well below
51 // axum's 2 MB default so a flood of oversized POSTs can't buffer freely.
52 .layer(axum::extract::DefaultBodyLimit::max(64 * 1024))
53 }
54
55 /// Bearer-token gate for the build triggers. No token configured -> pass
56 /// (main() only allows that on a loopback bind). Constant-time comparison.
57 async fn require_bearer(
58 token: Option<std::sync::Arc<str>>,
59 req: axum::extract::Request,
60 next: axum::middleware::Next,
61 ) -> axum::response::Response {
62 let Some(expected) = token.as_deref() else {
63 return next.run(req).await;
64 };
65 let path = req.uri().path().to_string();
66 let ok = req
67 .headers()
68 .get(axum::http::header::AUTHORIZATION)
69 .and_then(|h| h.to_str().ok())
70 .and_then(|h| h.strip_prefix("Bearer "))
71 .is_some_and(|t| ops_core::daemon::ct_eq(t, expected));
72 if ok {
73 next.run(req).await
74 } else {
75 tracing::warn!(path = %path, "rejected unauthenticated request to a build trigger");
76 (
77 axum::http::StatusCode::UNAUTHORIZED,
78 "missing or invalid bearer token\n",
79 )
80 .into_response()
81 }
82 }
83
84 #[derive(Serialize)]
85 struct StateView {
86 build: Option<BuildView>,
87 }
88
89 /// One app's latest build, plus what the topology says it ships.
90 ///
91 /// `/state` answers with the newest build row across every app, which means a
92 /// release that half-failed on one app becomes invisible the moment any other
93 /// app builds. `/status.json` reads per app instead, so nothing can be hidden
94 /// by a later, unrelated build.
95 pub(crate) struct AppStatusView {
96 pub(crate) app: String,
97 pub(crate) kind: crate::topology::Kind,
98 /// Every target this app ships, from its in-repo manifest -- including ones
99 /// the latest build never ran.
100 pub(crate) declared_targets: Vec<String>,
101 /// `None` until this app has ever been built.
102 pub(crate) build: Option<BuildView>,
103 /// Targets with a `releases` row at the build's version. Empty when
104 /// publishing is not in play for this app.
105 pub(crate) published_targets: Vec<String>,
106 /// What MNW will actually serve for this app at the build's version.
107 ///
108 /// `None` for anything that does not reach users through MNW: libraries go
109 /// to crates.io and services are installed onto hosts, so neither has an
110 /// OTA manifest and neither should be judged against one.
111 pub(crate) distribution: Option<crate::ota::Distribution>,
112 }
113
114 #[derive(Serialize)]
115 pub(crate) struct BuildView {
116 pub(crate) id: i64,
117 pub(crate) app: String,
118 pub(crate) version: String,
119 pub(crate) status: String,
120 pub(crate) created_at: String,
121 pub(crate) targets: Vec<TargetView>,
122 }
123
124 #[derive(Serialize)]
125 pub(crate) struct TargetView {
126 pub(crate) target: String,
127 pub(crate) status: String,
128 pub(crate) current_step: Option<String>,
129 pub(crate) error: Option<String>,
130 pub(crate) steps: Vec<StepView>,
131 }
132
133 #[derive(Serialize)]
134 pub(crate) struct StepView {
135 pub(crate) run_id: i64,
136 pub(crate) step: String,
137 pub(crate) status: String,
138 pub(crate) log_ref: Option<String>,
139 }
140
141 async fn get_state(State(s): State<AppState>) -> Result<Json<StateView>> {
142 let latest = sqlx::query(
143 "SELECT id, app, version, status, created_at FROM builds ORDER BY id DESC LIMIT 1",
144 )
145 .fetch_optional(&s.pool)
146 .await?;
147
148 let Some(b) = latest else {
149 return Ok(Json(StateView { build: None }));
150 };
151 Ok(Json(StateView {
152 build: Some(build_view(&s, &b).await?),
153 }))
154 }
155
156 /// Hydrate one `builds` row into its target x step matrix.
157 ///
158 /// Shared by `/state` and `/status.json` so the two cannot disagree about what
159 /// a build looks like.
160 ///
161 /// The matrix is the per-`(app, version)` RELEASE view, not one `builds` row's
162 /// targets: for each target, the latest `target_runs` row for this app+version
163 /// across EVERY build of that version. A `/retry` inserts a fresh single-target
164 /// build, so keying on `build_id` would collapse a 5-target release to the one
165 /// retried row and lose the answer to "did every target of this version ship".
166 /// Aggregating by `(app, version)` keeps the full matrix and folds the retry's
167 /// newer result into just that target's cell.
168 async fn build_view(s: &AppState, b: &sqlx::sqlite::SqliteRow) -> Result<BuildView> {
169 let build_id: i64 = b.get("id");
170 let app: String = b.get("app");
171 let version: String = b.get("version");
172
173 let target_rows = sqlx::query(
174 "SELECT id, target, status, current_step, error FROM target_runs tr
175 WHERE app = ?1 AND version = ?2
176 AND id = (SELECT MAX(id) FROM target_runs
177 WHERE app = ?1 AND version = ?2 AND target = tr.target)
178 ORDER BY target",
179 )
180 .bind(&app)
181 .bind(&version)
182 .fetch_all(&s.pool)
183 .await?;
184
185 let mut targets = Vec::with_capacity(target_rows.len());
186 for tr in target_rows {
187 let target_run_id: i64 = tr.get("id");
188 // Latest row per step for this target run.
189 let steps: Vec<StepView> = sqlx::query(
190 "SELECT id, step, status, log_ref FROM step_runs sr
191 WHERE target_run_id = ?1
192 AND id = (SELECT MAX(id) FROM step_runs
193 WHERE target_run_id = ?1 AND step = sr.step)
194 ORDER BY id",
195 )
196 .bind(target_run_id)
197 .fetch_all(&s.pool)
198 .await?
199 .into_iter()
200 .map(|r| StepView {
201 run_id: r.get("id"),
202 step: r.get("step"),
203 status: r.get("status"),
204 log_ref: r.get("log_ref"),
205 })
206 .collect();
207
208 targets.push(TargetView {
209 target: tr.get("target"),
210 status: tr.get("status"),
211 current_step: tr.get("current_step"),
212 error: tr.get("error"),
213 steps,
214 });
215 }
216
217 Ok(BuildView {
218 id: build_id,
219 app: b.get("app"),
220 version: b.get("version"),
221 status: b.get("status"),
222 created_at: b.get("created_at"),
223 targets,
224 })
225 }
226
227 /// `GET /status.json` -- every app's latest build, in the shared cross-service
228 /// payload every operator surface renders. See `crate::status`.
229 async fn get_status_json(State(s): State<AppState>) -> Result<Json<ops_status::Payload>> {
230 let view = status_view(&s).await?;
231 Ok(Json(crate::status::payload(&view, chrono::Utc::now())))
232 }
233
234 /// One release, answering "did every target of `{version}` ship?"
235 #[derive(Serialize)]
236 struct ReleaseView {
237 app: String,
238 version: String,
239 /// Every target the app's manifest declares for this release.
240 declared_targets: Vec<String>,
241 /// The per-`(app, version)` matrix: latest run per target across every build
242 /// (including retries) of this version.
243 build: BuildView,
244 /// Targets with a `releases` row at this version.
245 published_targets: Vec<String>,
246 /// Every declared target has a latest run of `ok` — the release built clean.
247 all_targets_green: bool,
248 /// `all_targets_green` AND every declared target is published.
249 complete: bool,
250 }
251
252 /// `GET /release/{app}/{version}` -- the release view for a SPECIFIC version, not
253 /// just the latest. Unlike `/state`, a retry can't hide it and an unrelated
254 /// newer build can't mask it; this is the durable "did 0.5.0 fully ship" query.
255 async fn get_release(
256 State(s): State<AppState>,
257 Path((app, version)): Path<(String, String)>,
258 ) -> Result<Json<ReleaseView>> {
259 let cfg = s
260 .topo
261 .app(&AppId::new(app.clone()))
262 .ok_or_else(|| Error::BadRequest(format!("unknown app `{app}`")))?;
263 let declared_targets: Vec<String> = cfg.targets.iter().map(ToString::to_string).collect();
264
265 // A representative build row for this exact version (newest wins for the
266 // build-level fields; the target matrix is aggregated across all of them).
267 let row = sqlx::query(
268 "SELECT id, app, version, status, created_at FROM builds
269 WHERE app = ? AND version = ? ORDER BY id DESC LIMIT 1",
270 )
271 .bind(&app)
272 .bind(&version)
273 .fetch_optional(&s.pool)
274 .await?
275 .ok_or(Error::NotFound)?;
276 let build = build_view(&s, &row).await?;
277
278 let published_targets = sqlx::query_scalar::<_, String>(
279 "SELECT DISTINCT target FROM releases WHERE app = ? AND version = ? ORDER BY target",
280 )
281 .bind(&app)
282 .bind(&version)
283 .fetch_all(&s.pool)
284 .await?;
285
286 // Green = every declared target has a latest run that is `ok`.
287 let all_targets_green = declared_targets.iter().all(|d| {
288 build
289 .targets
290 .iter()
291 .any(|t| &t.target == d && t.status == "ok")
292 });
293 let complete = all_targets_green
294 && declared_targets
295 .iter()
296 .all(|d| published_targets.contains(d));
297
298 Ok(Json(ReleaseView {
299 app,
300 version,
301 declared_targets,
302 build,
303 published_targets,
304 all_targets_green,
305 complete,
306 }))
307 }
308
309 /// One [`AppStatusView`] per app in the topology, name-ordered.
310 ///
311 /// Every declared app appears whether or not it has ever been built: an app
312 /// missing from the surface is indistinguishable from an app that is fine, and
313 /// the whole point of the viewer is that absence of evidence must be visible.
314 pub(crate) async fn status_view(s: &AppState) -> Result<Vec<AppStatusView>> {
315 let mut names: Vec<&String> = s.topo.app.keys().collect();
316 names.sort();
317
318 let mut apps = Vec::with_capacity(names.len());
319 for name in names {
320 let cfg = &s.topo.app[name];
321
322 let latest = sqlx::query(
323 "SELECT id, app, version, status, created_at FROM builds
324 WHERE app = ? ORDER BY id DESC LIMIT 1",
325 )
326 .bind(name)
327 .fetch_optional(&s.pool)
328 .await?;
329
330 let build = match latest {
331 Some(row) => Some(build_view(s, &row).await?),
332 None => None,
333 };
334
335 let published_targets = match &build {
336 Some(b) => {
337 sqlx::query_scalar::<_, String>(
338 "SELECT DISTINCT target FROM releases
339 WHERE app = ? AND version = ? ORDER BY target",
340 )
341 .bind(name)
342 .bind(&b.version)
343 .fetch_all(&s.pool)
344 .await?
345 }
346 None => Vec::new(),
347 };
348
349 let distribution = match (&build, cfg.kind) {
350 // Only apps reach users through MNW's OTA endpoint. A library's
351 // distribution question is answered by crates.io and a service's by
352 // whether it is installed, so probing MNW for either would invent a
353 // red that means nothing.
354 (Some(b), crate::topology::Kind::App) => {
355 Some(distribution_for(s, name, &b.version, &cfg.targets).await)
356 }
357 _ => None,
358 };
359
360 apps.push(AppStatusView {
361 app: name.clone(),
362 kind: cfg.kind,
363 declared_targets: cfg.targets.iter().map(ToString::to_string).collect(),
364 build,
365 published_targets,
366 distribution,
367 });
368 }
369 Ok(apps)
370 }
371
372 /// The cached distribution answer for one `(app, version)`, probing MNW only
373 /// when there is no fresh one.
374 async fn distribution_for(
375 s: &AppState,
376 app: &str,
377 version: &str,
378 targets: &[crate::domain::Target],
379 ) -> crate::ota::Distribution {
380 let key = (app.to_string(), version.to_string());
381
382 {
383 let cache = s.distribution.lock().await;
384 if let Some((taken, dist)) = cache.get(&key)
385 && taken.elapsed() < crate::state::DISTRIBUTION_TTL
386 {
387 return dist.clone();
388 }
389 }
390
391 // Deliberately not holding the lock across the probe: a slow MNW would
392 // otherwise serialize every poll behind one request.
393 let slug = crate::ota::mnw_slug(app);
394 let dist =
395 crate::ota::probe_distribution(&s.http, &s.mnw_base_url, &slug, version, targets).await;
396
397 s.distribution
398 .lock()
399 .await
400 .insert(key, (std::time::Instant::now(), dist.clone()));
401 dist
402 }
403
404 #[derive(Deserialize, Default)]
405 struct BuildBody {
406 app: String,
407 #[serde(default)]
408 version: Option<String>,
409 #[serde(default)]
410 targets: Vec<String>,
411 }
412
413 async fn build(
414 State(s): State<AppState>,
415 Json(body): Json<BuildBody>,
416 ) -> Result<Json<serde_json::Value>> {
417 let app = AppId::new(body.app);
418 let targets = parse_targets(body.targets)?;
419 let version = runner::resolve_version(&s, &app, body.version)?;
420 let targets = runner::resolve_targets(&s, &app, targets)?;
421 let build_id = runner::start_build(s, app, version.clone(), targets)
422 .await
423 .map_err(Error::Other)?;
424 Ok(Json(
425 serde_json::json!({ "accepted": true, "build_id": build_id, "version": version.to_string() }),
426 ))
427 }
428
429 #[derive(Deserialize)]
430 struct RetryBody {
431 app: String,
432 target: String,
433 #[serde(default)]
434 version: Option<String>,
435 }
436
437 async fn retry(
438 State(s): State<AppState>,
439 Json(body): Json<RetryBody>,
440 ) -> Result<Json<serde_json::Value>> {
441 let app = AppId::new(body.app);
442 let target: Target = body.target.parse().map_err(Error::BadRequest)?;
443 let version = runner::resolve_version(&s, &app, body.version)?;
444 let targets = runner::resolve_targets(&s, &app, vec![target])?;
445 let build_id = runner::start_build(s, app, version.clone(), targets)
446 .await
447 .map_err(Error::Other)?;
448 Ok(Json(
449 serde_json::json!({ "accepted": true, "build_id": build_id, "target": target.to_string() }),
450 ))
451 }
452
453 fn parse_targets(raw: Vec<String>) -> Result<Vec<Target>> {
454 raw.into_iter()
455 .map(|t| t.parse::<Target>().map_err(Error::BadRequest))
456 .collect()
457 }
458
459 #[derive(Deserialize)]
460 pub(crate) struct StepLogQuery {
461 /// Which run of this step to serve. Omitted, the newest wins.
462 run_id: Option<i64>,
463 }
464
465 /// The file holding `step`'s output in `dir`.
466 ///
467 /// Logs are named `<step>.<run_id>.log` so re-running an app at a version it
468 /// already built writes a new file instead of appending to the old one. The URL
469 /// carries no run id, so without `?run_id=` this picks the highest, the run a
470 /// reader asking for "the build log" means. An unsuffixed `<step>.log` is still
471 /// served when nothing newer exists.
472 async fn resolve_step_log(
473 dir: &std::path::Path,
474 step: &str,
475 run_id: Option<i64>,
476 ) -> Option<std::path::PathBuf> {
477 if let Some(id) = run_id {
478 let exact = dir.join(format!("{step}.{id}.log"));
479 return tokio::fs::try_exists(&exact).await.ok()?.then_some(exact);
480 }
481 let mut entries = tokio::fs::read_dir(dir).await.ok()?;
482 let mut newest: Option<(i64, std::path::PathBuf)> = None;
483 while let Ok(Some(e)) = entries.next_entry().await {
484 let name = e.file_name();
485 let Some(name) = name.to_str() else { continue };
486 let Some(rest) = name
487 .strip_prefix(step)
488 .and_then(|r| r.strip_prefix('.'))
489 .and_then(|r| r.strip_suffix(".log"))
490 else {
491 continue;
492 };
493 let Ok(id) = rest.parse::<i64>() else {
494 continue;
495 };
496 if newest.as_ref().is_none_or(|(best, _)| id > *best) {
497 newest = Some((id, e.path()));
498 }
499 }
500 match newest {
501 Some((_, path)) => Some(path),
502 // Legacy name, from before the run id was part of it.
503 None => {
504 let legacy = dir.join(format!("{step}.log"));
505 tokio::fs::try_exists(&legacy).await.ok()?.then_some(legacy)
506 }
507 }
508 }
509
510 async fn get_step_log(
511 State(s): State<AppState>,
512 Path((app, version, target, step)): Path<(String, String, String, String)>,
513 axum::extract::Query(q): axum::extract::Query<StepLogQuery>,
514 ) -> Result<axum::response::Response> {
515 fn safe(seg: &str) -> bool {
516 !seg.is_empty() && !seg.contains('/') && !seg.contains('\\') && seg != "." && seg != ".."
517 }
518 if ![&app, &version, &target, &step]
519 .into_iter()
520 .all(|s| safe(s))
521 {
522 return Err(Error::NotFound);
523 }
524 let dir = s.cfg.logs_root.join(&app).join(&version).join(&target);
525 let path = resolve_step_log(&dir, &step, q.run_id)
526 .await
527 .ok_or(Error::NotFound)?;
528 // Stream the log in chunks rather than reading the whole (potentially large,
529 // verbose-build) file into memory.
530 let file = match tokio::fs::File::open(&path).await {
531 Ok(f) => f,
532 Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Err(Error::NotFound),
533 Err(e) => return Err(Error::Other(e.into())),
534 };
535 let (tx, rx) =
536 tokio::sync::mpsc::channel::<std::result::Result<axum::body::Bytes, std::io::Error>>(8);
537 tokio::spawn(async move {
538 use tokio::io::AsyncReadExt;
539 let mut file = file;
540 let mut buf = vec![0u8; 64 * 1024];
541 loop {
542 match file.read(&mut buf).await {
543 Ok(0) => break,
544 Ok(n) => {
545 if tx
546 .send(Ok(axum::body::Bytes::copy_from_slice(&buf[..n])))
547 .await
548 .is_err()
549 {
550 break;
551 }
552 }
553 Err(e) => {
554 let _ = tx.send(Err(e)).await;
555 break;
556 }
557 }
558 }
559 });
560 let body = axum::body::Body::from_stream(tokio_stream::wrappers::ReceiverStream::new(rx));
561 Ok((
562 [(
563 axum::http::header::CONTENT_TYPE,
564 "text/plain; charset=utf-8",
565 )],
566 body,
567 )
568 .into_response())
569 }
570
571 async fn events_ws(ws: WebSocketUpgrade, State(s): State<AppState>) -> impl IntoResponse {
572 use axum::extract::ws::Message;
573 use tokio::sync::broadcast::error::RecvError;
574
575 ws.on_upgrade(move |mut socket| async move {
576 // Subscribe to both channels and merge them: a lag on the high-rate log
577 // stream emits its own `lagged` frame without dropping anything on the
578 // status stream (and vice versa), so a busy build's chunk firehose can't
579 // evict a TargetFailed/PublishOk the operator needs to see.
580 let mut status_rx = s.events.subscribe_status();
581 let mut logs_rx = s.events.subscribe_logs();
582 loop {
583 let recv = tokio::select! {
584 r = status_rx.recv() => r,
585 r = logs_rx.recv() => r,
586 };
587 match recv {
588 Ok(env) => {
589 let json = match serde_json::to_string(&env) {
590 Ok(s) => s,
591 Err(e) => {
592 tracing::warn!(error = %e, "events ws: serialize failed");
593 continue;
594 }
595 };
596 if socket.send(Message::Text(json.into())).await.is_err() {
597 break;
598 }
599 }
600 Err(RecvError::Lagged(n)) => {
601 let _ = socket
602 .send(Message::Text(
603 format!(r#"{{"kind":"lagged","skipped":{n}}}"#).into(),
604 ))
605 .await;
606 }
607 Err(RecvError::Closed) => break,
608 }
609 }
610 })
611 }
612
613 #[cfg(test)]
614 mod tests {
615 use super::*;
616 use crate::config::Config;
617 use crate::ota::OtaRegistry;
618 use crate::topology::Topology;
619 use axum::body::Body;
620 use axum::http::{Request, StatusCode};
621 use http_body_util::BodyExt;
622 use std::collections::HashMap;
623 use std::sync::Arc;
624 use tokio::sync::Mutex;
625 use tower::ServiceExt;
626
627 async fn test_state(root: &std::path::Path) -> AppState {
628 let cfg = Config::for_tests(root);
629 let pool = crate::db::open(&cfg.db_path).await.unwrap();
630 let repo = root.join("goingson");
631 std::fs::create_dir_all(&repo).unwrap();
632 std::fs::write(repo.join("bento.toml"), "targets = [\"linux/x86_64\"]\n").unwrap();
633 let topo = Topology::from_str_for_tests(&format!(
634 r#"
635 [[host]]
636 name = "fw13"
637 ssh = "local"
638 targets = ["linux/x86_64"]
639
640 [app.goingson]
641 repo = "{}"
642 "#,
643 repo.display()
644 ))
645 .unwrap();
646 let executors = Arc::new(crate::state::build_executors(&topo));
647 let syncs = Arc::new(crate::state::build_syncs(&topo));
648 let host_locks = crate::state::build_host_locks(&topo);
649 AppState {
650 pool,
651 topo: Arc::new(topo),
652 cfg: Arc::new(cfg),
653 prom: crate::metrics::test_handle(),
654 events: crate::events::channel(),
655 ota: Arc::new(OtaRegistry::standard("https://makenot.work")),
656 executors,
657 syncs,
658 active: Arc::new(Mutex::new(HashMap::new())),
659 api_token: None,
660 host_locks,
661 distribution: Arc::new(Mutex::new(HashMap::new())),
662 http: crate::tls::builder().build().unwrap(),
663 // Port 1 refuses instantly. A test must never probe production, and
664 // a refusal is also faster than any timeout would be.
665 mnw_base_url: "http://127.0.0.1:1".into(),
666 }
667 }
668
669 async fn body_string(resp: axum::response::Response) -> String {
670 let bytes = resp.into_body().collect().await.unwrap().to_bytes();
671 String::from_utf8(bytes.to_vec()).unwrap()
672 }
673
674 #[tokio::test]
675 async fn state_is_empty_initially() {
676 let tmp = tempfile::tempdir().unwrap();
677 let app = router(test_state(tmp.path()).await);
678 let resp = app
679 .oneshot(
680 Request::builder()
681 .uri("/state")
682 .body(Body::empty())
683 .unwrap(),
684 )
685 .await
686 .unwrap();
687 assert_eq!(resp.status(), StatusCode::OK);
688 assert_eq!(body_string(resp).await, r#"{"build":null}"#);
689 }
690
691 #[tokio::test]
692 async fn status_json_lists_every_declared_app_before_any_build() {
693 // The mapping is tested in `crate::status`. This asserts the route is
694 // wired and that an app with no build history still reaches the
695 // surface -- an app missing from the viewer is indistinguishable from
696 // an app that is fine.
697 let tmp = tempfile::tempdir().unwrap();
698 let app = router(test_state(tmp.path()).await);
699 let resp = app
700 .oneshot(
701 Request::builder()
702 .uri("/status.json")
703 .body(Body::empty())
704 .unwrap(),
705 )
706 .await
707 .unwrap();
708 assert_eq!(resp.status(), StatusCode::OK);
709
710 let payload: ops_status::Payload = serde_json::from_str(&body_string(resp).await).unwrap();
711 assert_eq!(payload.source, crate::status::SOURCE);
712 assert_eq!(payload.schema, ops_status::SCHEMA_VERSION);
713 assert_eq!(payload.validate(), Ok(()));
714
715 let goingson = payload.node("app:goingson").expect("declared app appears");
716 assert_eq!(goingson.status, ops_status::Status::Pending);
717 assert!(payload.actions.contains_key("build-goingson"));
718 assert!(payload.node("target:goingson:linux/x86_64").is_some());
719 }
720
721 // ---- CF2: bearer-token auth on build triggers ----
722
723 #[tokio::test]
724 async fn build_route_requires_bearer_when_token_set() {
725 let tmp = tempfile::tempdir().unwrap();
726 let mut state = test_state(tmp.path()).await;
727 state.api_token = Some(std::sync::Arc::from("s3cr3t"));
728 let app = router(state);
729
730 let build_req = || {
731 Request::builder()
732 .method("POST")
733 .uri("/build")
734 .header("content-type", "application/json")
735 .body(Body::from(r#"{"app":"goingson"}"#))
736 .unwrap()
737 };
738
739 // No token -> 401.
740 let resp = app.clone().oneshot(build_req()).await.unwrap();
741 assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
742
743 // Wrong token -> 401.
744 let mut bad = build_req();
745 bad.headers_mut()
746 .insert("authorization", "Bearer nope".parse().unwrap());
747 let resp = app.clone().oneshot(bad).await.unwrap();
748 assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
749
750 // Correct token -> passes auth (NOT 401; the build itself is accepted).
751 let mut good = build_req();
752 good.headers_mut()
753 .insert("authorization", "Bearer s3cr3t".parse().unwrap());
754 let resp = app.clone().oneshot(good).await.unwrap();
755 assert_ne!(resp.status(), StatusCode::UNAUTHORIZED);
756 }
757
758 #[tokio::test]
759 async fn read_route_open_when_token_set() {
760 let tmp = tempfile::tempdir().unwrap();
761 let mut state = test_state(tmp.path()).await;
762 state.api_token = Some(std::sync::Arc::from("s3cr3t"));
763 let app = router(state);
764 let resp = app
765 .oneshot(
766 Request::builder()
767 .uri("/state")
768 .body(Body::empty())
769 .unwrap(),
770 )
771 .await
772 .unwrap();
773 assert_eq!(resp.status(), StatusCode::OK);
774 }
775
776 #[tokio::test]
777 async fn step_log_rejects_traversal() {
778 let tmp = tempfile::tempdir().unwrap();
779 let app = router(test_state(tmp.path()).await);
780 let resp = app
781 .oneshot(
782 Request::builder()
783 .uri("/logs/goingson/0.4.1/..%2f..%2fetc/passwd")
784 .body(Body::empty())
785 .unwrap(),
786 )
787 .await
788 .unwrap();
789 assert_eq!(resp.status(), StatusCode::NOT_FOUND);
790 }
791
792 /// A log written before the run id entered the filename is still served.
793 #[tokio::test]
794 async fn step_log_streams_existing_file() {
795 let tmp = tempfile::tempdir().unwrap();
796 let state = test_state(tmp.path()).await;
797 let dir = state
798 .cfg
799 .logs_root
800 .join("goingson")
801 .join("0.4.1")
802 .join("linux-x86_64");
803 std::fs::create_dir_all(&dir).unwrap();
804 std::fs::write(dir.join("build.log"), b"compiling\nlinked\n").unwrap();
805 let app = router(state);
806 let resp = app
807 .oneshot(
808 Request::builder()
809 .uri("/logs/goingson/0.4.1/linux-x86_64/build")
810 .body(Body::empty())
811 .unwrap(),
812 )
813 .await
814 .unwrap();
815 assert_eq!(resp.status(), StatusCode::OK);
816 assert_eq!(body_string(resp).await, "compiling\nlinked\n");
817 }
818
819 /// Two runs of the same app/version/target/step do not share a file, so the
820 /// route has to choose: newest by default, exact on `?run_id=`. An unsuffixed
821 /// file must lose to any run-id-keyed one, or a stale log outranks every run
822 /// since.
823 #[tokio::test]
824 async fn step_log_picks_newest_run_and_honours_run_id() {
825 let tmp = tempfile::tempdir().unwrap();
826 let state = test_state(tmp.path()).await;
827 let dir = state
828 .cfg
829 .logs_root
830 .join("goingson")
831 .join("0.4.1")
832 .join("linux-x86_64");
833 std::fs::create_dir_all(&dir).unwrap();
834 std::fs::write(dir.join("build.log"), b"legacy\n").unwrap();
835 std::fs::write(dir.join("build.55.log"), b"run 55\n").unwrap();
836 std::fs::write(dir.join("build.201.log"), b"run 201\n").unwrap();
837 // Same prefix, different step: must not be mistaken for a run of `build`.
838 std::fs::write(dir.join("build_extra.9.log"), b"other\n").unwrap();
839
840 let fetch = |uri: &'static str| {
841 let app = router(state.clone());
842 async move {
843 app.oneshot(Request::builder().uri(uri).body(Body::empty()).unwrap())
844 .await
845 .unwrap()
846 }
847 };
848
849 let resp = fetch("/logs/goingson/0.4.1/linux-x86_64/build").await;
850 assert_eq!(resp.status(), StatusCode::OK);
851 assert_eq!(
852 body_string(resp).await,
853 "run 201\n",
854 "highest run id wins, and 201 must beat 55 numerically rather than lexically"
855 );
856
857 let resp = fetch("/logs/goingson/0.4.1/linux-x86_64/build?run_id=55").await;
858 assert_eq!(resp.status(), StatusCode::OK);
859 assert_eq!(body_string(resp).await, "run 55\n");
860
861 let resp = fetch("/logs/goingson/0.4.1/linux-x86_64/build?run_id=999").await;
862 assert_eq!(
863 resp.status(),
864 StatusCode::NOT_FOUND,
865 "an unknown run id is a miss, not a silent fall back to the newest"
866 );
867 }
868
869 #[tokio::test]
870 async fn bad_request_body_is_a_json_error_envelope() {
871 let tmp = tempfile::tempdir().unwrap();
872 let app = router(test_state(tmp.path()).await);
873 let resp = app
874 .oneshot(
875 Request::builder()
876 .method("POST")
877 .uri("/build")
878 .header("content-type", "application/json")
879 .body(Body::from(r#"{"app":"nope","targets":["linux/x86_64"]}"#))
880 .unwrap(),
881 )
882 .await
883 .unwrap();
884 assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
885 // JSON envelope, message preserved.
886 let body = body_string(resp).await;
887 let v: serde_json::Value = serde_json::from_str(&body).unwrap();
888 assert!(v["error"].as_str().unwrap().contains("unknown app"));
889 }
890
891 #[tokio::test]
892 async fn build_rejects_unshipped_target_as_bad_request() {
893 // windows/x86_64 is a valid target but the test app doesn't ship it.
894 // That's a client error (400), not a server error (500).
895 let tmp = tempfile::tempdir().unwrap();
896 let app = router(test_state(tmp.path()).await);
897 let resp = app
898 .oneshot(
899 Request::builder()
900 .method("POST")
901 .uri("/build")
902 .header("content-type", "application/json")
903 // Explicit version so resolve_version doesn't read a
904 // (nonexistent) tauri.conf first — we're testing the target
905 // validation specifically.
906 .body(Body::from(
907 r#"{"app":"goingson","version":"0.4.1","targets":["windows/x86_64"]}"#,
908 ))
909 .unwrap(),
910 )
911 .await
912 .unwrap();
913 assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
914 assert!(
915 body_string(resp).await.contains("does not ship target"),
916 "names the problem"
917 );
918 }
919
920 #[tokio::test]
921 async fn build_rejects_unknown_app_as_bad_request() {
922 let tmp = tempfile::tempdir().unwrap();
923 let app = router(test_state(tmp.path()).await);
924 let resp = app
925 .oneshot(
926 Request::builder()
927 .method("POST")
928 .uri("/build")
929 .header("content-type", "application/json")
930 .body(Body::from(r#"{"app":"nope","targets":["linux/x86_64"]}"#))
931 .unwrap(),
932 )
933 .await
934 .unwrap();
935 assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
936 assert!(
937 body_string(resp).await.contains("unknown app"),
938 "names the problem"
939 );
940 }
941
942 #[tokio::test]
943 async fn build_rejects_malformed_target_as_bad_request() {
944 // A non-parseable target string is also a 400 (via parse_targets).
945 let tmp = tempfile::tempdir().unwrap();
946 let app = router(test_state(tmp.path()).await);
947 let resp = app
948 .oneshot(
949 Request::builder()
950 .method("POST")
951 .uri("/build")
952 .header("content-type", "application/json")
953 .body(Body::from(
954 r#"{"app":"goingson","targets":["not-a-target"]}"#,
955 ))
956 .unwrap(),
957 )
958 .await
959 .unwrap();
960 assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
961 }
962
963 /// A two-target app + a two-host topology, for the release-view aggregation.
964 async fn two_target_state(root: &std::path::Path) -> AppState {
965 let cfg = Config::for_tests(root);
966 let pool = crate::db::open(&cfg.db_path).await.unwrap();
967 let repo = root.join("demo");
968 std::fs::create_dir_all(&repo).unwrap();
969 std::fs::write(
970 repo.join("bento.toml"),
971 "targets = [\"linux/x86_64\", \"macos/aarch64\"]\n",
972 )
973 .unwrap();
974 let topo = Topology::from_str_for_tests(&format!(
975 "[[host]]\nname = \"fw13\"\nssh = \"local\"\ntargets = [\"linux/x86_64\"]\n\
976 [[host]]\nname = \"mbp\"\nssh = \"mbp\"\ntargets = [\"macos/aarch64\"]\n\
977 [app.demo]\nrepo = \"{}\"\n",
978 repo.display()
979 ))
980 .unwrap();
981 let executors = Arc::new(crate::state::build_executors(&topo));
982 let syncs = Arc::new(crate::state::build_syncs(&topo));
983 let host_locks = crate::state::build_host_locks(&topo);
984 AppState {
985 pool,
986 topo: Arc::new(topo),
987 cfg: Arc::new(cfg),
988 prom: crate::metrics::test_handle(),
989 events: crate::events::channel(),
990 ota: Arc::new(OtaRegistry::standard("https://makenot.work")),
991 executors,
992 syncs,
993 active: Arc::new(Mutex::new(HashMap::new())),
994 api_token: None,
995 host_locks,
996 distribution: Arc::new(Mutex::new(HashMap::new())),
997 http: crate::tls::builder().build().unwrap(),
998 // Port 1 refuses instantly. A test must never probe production, and
999 // a refusal is also faster than any timeout would be.
1000 mnw_base_url: "http://127.0.0.1:1".into(),
1001 }
1002 }
1003
1004 async fn insert_target_run(pool: &sqlx::SqlitePool, ver: &str, target: &str, status: &str) {
1005 let bid: i64 = sqlx::query_scalar(
1006 "INSERT INTO builds (app, version, status, created_at) VALUES ('demo', ?, ?, '2026-07-23T00:00:00Z') RETURNING id",
1007 )
1008 .bind(ver)
1009 .bind(status)
1010 .fetch_one(pool)
1011 .await
1012 .unwrap();
1013 sqlx::query(
1014 "INSERT INTO target_runs (build_id, app, version, target, status, started_at)
1015 VALUES (?, 'demo', ?, ?, ?, '2026-07-23T00:00:00Z')",
1016 )
1017 .bind(bid)
1018 .bind(ver)
1019 .bind(target)
1020 .bind(status)
1021 .execute(pool)
1022 .await
1023 .unwrap();
1024 }
1025
1026 /// The audit's H2: `/retry` inserts a whole new single-target build, so a
1027 /// build-id-keyed view collapses a multi-target release to the retried row.
1028 /// The release view aggregates by (app, version), so a green retry of one
1029 /// target folds into that cell while the other target stays visible.
1030 #[tokio::test]
1031 async fn release_view_survives_a_single_target_retry() {
1032 let tmp = tempfile::tempdir().unwrap();
1033 let state = two_target_state(tmp.path()).await;
1034 let pool = state.pool.clone();
1035
1036 // Original release build: linux ok, macos failed.
1037 insert_target_run(&pool, "0.5.0", "linux/x86_64", "ok").await;
1038 insert_target_run(&pool, "0.5.0", "macos/aarch64", "failed").await;
1039 // Retry macos only — a fresh single-target build that succeeds.
1040 insert_target_run(&pool, "0.5.0", "macos/aarch64", "ok").await;
1041
1042 let app = router(state);
1043 let resp = app
1044 .oneshot(
1045 Request::builder()
1046 .uri("/release/demo/0.5.0")
1047 .body(Body::empty())
1048 .unwrap(),
1049 )
1050 .await
1051 .unwrap();
1052 assert_eq!(resp.status(), StatusCode::OK);
1053 let v: serde_json::Value = serde_json::from_str(&body_string(resp).await).unwrap();
1054
1055 // Both targets are present (the matrix did not collapse to the retry),
1056 // and macos shows the newer `ok`, not the earlier `failed`.
1057 let targets = v["build"]["targets"].as_array().unwrap();
1058 assert_eq!(targets.len(), 2, "full matrix preserved across the retry");
1059 let macos = targets
1060 .iter()
1061 .find(|t| t["target"] == "macos/aarch64")
1062 .unwrap();
1063 assert_eq!(macos["status"], "ok", "retry result wins the cell");
1064 assert_eq!(v["all_targets_green"], true);
1065 }
1066
1067 #[tokio::test]
1068 async fn release_view_404s_for_an_unbuilt_version() {
1069 let tmp = tempfile::tempdir().unwrap();
1070 let app = router(two_target_state(tmp.path()).await);
1071 let resp = app
1072 .oneshot(
1073 Request::builder()
1074 .uri("/release/demo/9.9.9")
1075 .body(Body::empty())
1076 .unwrap(),
1077 )
1078 .await
1079 .unwrap();
1080 assert_eq!(resp.status(), StatusCode::NOT_FOUND);
1081 }
1082
1083 // ---- /retry handler (the route itself, distinct from the release-view
1084 // aggregation a retry produces above) ----
1085
1086 fn retry_req(json: &str) -> Request<Body> {
1087 Request::builder()
1088 .method("POST")
1089 .uri("/retry")
1090 .header("content-type", "application/json")
1091 .body(Body::from(json.to_string()))
1092 .unwrap()
1093 }
1094
1095 #[tokio::test]
1096 async fn retry_accepts_a_single_shipped_target() {
1097 let tmp = tempfile::tempdir().unwrap();
1098 let app = router(test_state(tmp.path()).await);
1099 // Explicit version so the handler doesn't read a (nonexistent) version
1100 // file; the point here is the accept path, not version resolution.
1101 let resp = app
1102 .oneshot(retry_req(
1103 r#"{"app":"goingson","target":"linux/x86_64","version":"0.4.1"}"#,
1104 ))
1105 .await
1106 .unwrap();
1107 assert_eq!(resp.status(), StatusCode::OK);
1108 let v: serde_json::Value = serde_json::from_str(&body_string(resp).await).unwrap();
1109 assert_eq!(v["accepted"], true);
1110 assert_eq!(v["target"], "linux/x86_64", "echoes the retried target");
1111 assert!(
1112 v["build_id"].as_i64().is_some(),
1113 "a fresh single-target build id is returned"
1114 );
1115 }
1116
1117 #[tokio::test]
1118 async fn retry_rejects_an_unshipped_target_as_bad_request() {
1119 // goingson does not ship windows; a retry of it is a 400, not a 500.
1120 let tmp = tempfile::tempdir().unwrap();
1121 let app = router(test_state(tmp.path()).await);
1122 let resp = app
1123 .oneshot(retry_req(
1124 r#"{"app":"goingson","target":"windows/x86_64","version":"0.4.1"}"#,
1125 ))
1126 .await
1127 .unwrap();
1128 assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
1129 assert!(body_string(resp).await.contains("does not ship target"));
1130 }
1131
1132 #[tokio::test]
1133 async fn retry_rejects_a_malformed_target_as_bad_request() {
1134 let tmp = tempfile::tempdir().unwrap();
1135 let app = router(test_state(tmp.path()).await);
1136 let resp = app
1137 .oneshot(retry_req(
1138 r#"{"app":"goingson","target":"not-a-target","version":"0.4.1"}"#,
1139 ))
1140 .await
1141 .unwrap();
1142 assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
1143 }
1144
1145 #[tokio::test]
1146 async fn retry_rejects_an_unknown_app_as_bad_request() {
1147 let tmp = tempfile::tempdir().unwrap();
1148 let app = router(test_state(tmp.path()).await);
1149 let resp = app
1150 .oneshot(retry_req(r#"{"app":"nope","target":"linux/x86_64"}"#))
1151 .await
1152 .unwrap();
1153 assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
1154 assert!(body_string(resp).await.contains("unknown app"));
1155 }
1156
1157 // ---- /events websocket: the HTTP->WS wiring (upgrade + serialize + send).
1158 // The channel merge and lagged framing themselves are covered in
1159 // `crate::events`; this asserts a real client on `/events` receives an
1160 // emitted event as the flat-`kind` JSON the TUI parses. ----
1161
1162 #[tokio::test]
1163 async fn events_ws_streams_emitted_events_as_flat_json() {
1164 use futures_util::StreamExt;
1165 use tokio_tungstenite::tungstenite::Message as WsMessage;
1166
1167 let tmp = tempfile::tempdir().unwrap();
1168 let state = test_state(tmp.path()).await;
1169 let events = state.events.clone();
1170 let app = router(state);
1171
1172 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
1173 let addr = listener.local_addr().unwrap();
1174 tokio::spawn(async move {
1175 axum::serve(listener, app).await.unwrap();
1176 });
1177
1178 let (mut ws, _resp) = tokio_tungstenite::connect_async(format!("ws://{addr}/events"))
1179 .await
1180 .expect("client connects to /events");
1181
1182 // The handler subscribes to the bus inside `on_upgrade`, which can land
1183 // just after the handshake returns; a broadcast has no backlog for a
1184 // late subscriber, so re-emit until the first frame arrives rather than
1185 // racing that window with a single send.
1186 let ev = crate::events::Event::PublishFailed {
1187 app: AppId::new("goingson"),
1188 target: "macos/aarch64".parse().unwrap(),
1189 channel: "stable".into(),
1190 error: "boom".into(),
1191 };
1192 let mut frame = None;
1193 for _ in 0..100 {
1194 crate::events::emit(&events, ev.clone());
1195 if let Ok(Some(Ok(WsMessage::Text(t)))) =
1196 tokio::time::timeout(std::time::Duration::from_millis(50), ws.next()).await
1197 {
1198 frame = Some(t);
1199 break;
1200 }
1201 }
1202 let frame = frame.expect("received an event frame within the retry budget");
1203 let v: serde_json::Value = serde_json::from_str(&frame).unwrap();
1204 assert_eq!(v["kind"], "publish_failed", "flat kind-tagged wire shape");
1205 assert_eq!(v["channel"], "stable");
1206 assert!(v.get("event").is_none(), "not nested under `event`");
1207 }
1208 }
1209