Skip to main content

max / makenotwork

37.3 KB · 1050 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 }
107
108 #[derive(Serialize)]
109 pub(crate) struct BuildView {
110 pub(crate) id: i64,
111 pub(crate) app: String,
112 pub(crate) version: String,
113 pub(crate) status: String,
114 pub(crate) created_at: String,
115 pub(crate) targets: Vec<TargetView>,
116 }
117
118 #[derive(Serialize)]
119 pub(crate) struct TargetView {
120 pub(crate) target: String,
121 pub(crate) status: String,
122 pub(crate) current_step: Option<String>,
123 pub(crate) error: Option<String>,
124 pub(crate) steps: Vec<StepView>,
125 }
126
127 #[derive(Serialize)]
128 pub(crate) struct StepView {
129 pub(crate) run_id: i64,
130 pub(crate) step: String,
131 pub(crate) status: String,
132 pub(crate) log_ref: Option<String>,
133 }
134
135 async fn get_state(State(s): State<AppState>) -> Result<Json<StateView>> {
136 let latest = sqlx::query(
137 "SELECT id, app, version, status, created_at FROM builds ORDER BY id DESC LIMIT 1",
138 )
139 .fetch_optional(&s.pool)
140 .await?;
141
142 let Some(b) = latest else {
143 return Ok(Json(StateView { build: None }));
144 };
145 Ok(Json(StateView {
146 build: Some(build_view(&s, &b).await?),
147 }))
148 }
149
150 /// Hydrate one `builds` row into its target x step matrix.
151 ///
152 /// Shared by `/state` and `/status.json` so the two cannot disagree about what
153 /// a build looks like.
154 ///
155 /// The matrix is the per-`(app, version)` RELEASE view, not one `builds` row's
156 /// targets: for each target, the latest `target_runs` row for this app+version
157 /// across EVERY build of that version. A `/retry` inserts a fresh single-target
158 /// build, so keying on `build_id` would collapse a 5-target release to the one
159 /// retried row and lose the answer to "did every target of this version ship".
160 /// Aggregating by `(app, version)` keeps the full matrix and folds the retry's
161 /// newer result into just that target's cell.
162 async fn build_view(s: &AppState, b: &sqlx::sqlite::SqliteRow) -> Result<BuildView> {
163 let build_id: i64 = b.get("id");
164 let app: String = b.get("app");
165 let version: String = b.get("version");
166
167 let target_rows = sqlx::query(
168 "SELECT id, target, status, current_step, error FROM target_runs tr
169 WHERE app = ?1 AND version = ?2
170 AND id = (SELECT MAX(id) FROM target_runs
171 WHERE app = ?1 AND version = ?2 AND target = tr.target)
172 ORDER BY target",
173 )
174 .bind(&app)
175 .bind(&version)
176 .fetch_all(&s.pool)
177 .await?;
178
179 let mut targets = Vec::with_capacity(target_rows.len());
180 for tr in target_rows {
181 let target_run_id: i64 = tr.get("id");
182 // Latest row per step for this target run.
183 let steps: Vec<StepView> = sqlx::query(
184 "SELECT id, step, status, log_ref FROM step_runs sr
185 WHERE target_run_id = ?1
186 AND id = (SELECT MAX(id) FROM step_runs
187 WHERE target_run_id = ?1 AND step = sr.step)
188 ORDER BY id",
189 )
190 .bind(target_run_id)
191 .fetch_all(&s.pool)
192 .await?
193 .into_iter()
194 .map(|r| StepView {
195 run_id: r.get("id"),
196 step: r.get("step"),
197 status: r.get("status"),
198 log_ref: r.get("log_ref"),
199 })
200 .collect();
201
202 targets.push(TargetView {
203 target: tr.get("target"),
204 status: tr.get("status"),
205 current_step: tr.get("current_step"),
206 error: tr.get("error"),
207 steps,
208 });
209 }
210
211 Ok(BuildView {
212 id: build_id,
213 app: b.get("app"),
214 version: b.get("version"),
215 status: b.get("status"),
216 created_at: b.get("created_at"),
217 targets,
218 })
219 }
220
221 /// `GET /status.json` -- every app's latest build, in the shared cross-service
222 /// payload every operator surface renders. See `crate::status`.
223 async fn get_status_json(State(s): State<AppState>) -> Result<Json<ops_status::Payload>> {
224 let view = status_view(&s).await?;
225 Ok(Json(crate::status::payload(&view, chrono::Utc::now())))
226 }
227
228 /// One release, answering "did every target of `{version}` ship?"
229 #[derive(Serialize)]
230 struct ReleaseView {
231 app: String,
232 version: String,
233 /// Every target the app's manifest declares for this release.
234 declared_targets: Vec<String>,
235 /// The per-`(app, version)` matrix: latest run per target across every build
236 /// (including retries) of this version.
237 build: BuildView,
238 /// Targets with a `releases` row at this version.
239 published_targets: Vec<String>,
240 /// Every declared target has a latest run of `ok` — the release built clean.
241 all_targets_green: bool,
242 /// `all_targets_green` AND every declared target is published.
243 complete: bool,
244 }
245
246 /// `GET /release/{app}/{version}` -- the release view for a SPECIFIC version, not
247 /// just the latest. Unlike `/state`, a retry can't hide it and an unrelated
248 /// newer build can't mask it; this is the durable "did 0.5.0 fully ship" query.
249 async fn get_release(
250 State(s): State<AppState>,
251 Path((app, version)): Path<(String, String)>,
252 ) -> Result<Json<ReleaseView>> {
253 let cfg = s
254 .topo
255 .app(&AppId::new(app.clone()))
256 .ok_or_else(|| Error::BadRequest(format!("unknown app `{app}`")))?;
257 let declared_targets: Vec<String> = cfg.targets.iter().map(ToString::to_string).collect();
258
259 // A representative build row for this exact version (newest wins for the
260 // build-level fields; the target matrix is aggregated across all of them).
261 let row = sqlx::query(
262 "SELECT id, app, version, status, created_at FROM builds
263 WHERE app = ? AND version = ? ORDER BY id DESC LIMIT 1",
264 )
265 .bind(&app)
266 .bind(&version)
267 .fetch_optional(&s.pool)
268 .await?
269 .ok_or(Error::NotFound)?;
270 let build = build_view(&s, &row).await?;
271
272 let published_targets = sqlx::query_scalar::<_, String>(
273 "SELECT DISTINCT target FROM releases WHERE app = ? AND version = ? ORDER BY target",
274 )
275 .bind(&app)
276 .bind(&version)
277 .fetch_all(&s.pool)
278 .await?;
279
280 // Green = every declared target has a latest run that is `ok`.
281 let all_targets_green = declared_targets.iter().all(|d| {
282 build
283 .targets
284 .iter()
285 .any(|t| &t.target == d && t.status == "ok")
286 });
287 let complete = all_targets_green
288 && declared_targets
289 .iter()
290 .all(|d| published_targets.contains(d));
291
292 Ok(Json(ReleaseView {
293 app,
294 version,
295 declared_targets,
296 build,
297 published_targets,
298 all_targets_green,
299 complete,
300 }))
301 }
302
303 /// One [`AppStatusView`] per app in the topology, name-ordered.
304 ///
305 /// Every declared app appears whether or not it has ever been built: an app
306 /// missing from the surface is indistinguishable from an app that is fine, and
307 /// the whole point of the viewer is that absence of evidence must be visible.
308 pub(crate) async fn status_view(s: &AppState) -> Result<Vec<AppStatusView>> {
309 let mut names: Vec<&String> = s.topo.app.keys().collect();
310 names.sort();
311
312 let mut apps = Vec::with_capacity(names.len());
313 for name in names {
314 let cfg = &s.topo.app[name];
315
316 let latest = sqlx::query(
317 "SELECT id, app, version, status, created_at FROM builds
318 WHERE app = ? ORDER BY id DESC LIMIT 1",
319 )
320 .bind(name)
321 .fetch_optional(&s.pool)
322 .await?;
323
324 let build = match latest {
325 Some(row) => Some(build_view(s, &row).await?),
326 None => None,
327 };
328
329 let published_targets = match &build {
330 Some(b) => {
331 sqlx::query_scalar::<_, String>(
332 "SELECT DISTINCT target FROM releases
333 WHERE app = ? AND version = ? ORDER BY target",
334 )
335 .bind(name)
336 .bind(&b.version)
337 .fetch_all(&s.pool)
338 .await?
339 }
340 None => Vec::new(),
341 };
342
343 apps.push(AppStatusView {
344 app: name.clone(),
345 kind: cfg.kind,
346 declared_targets: cfg.targets.iter().map(ToString::to_string).collect(),
347 build,
348 published_targets,
349 });
350 }
351 Ok(apps)
352 }
353
354 #[derive(Deserialize, Default)]
355 struct BuildBody {
356 app: String,
357 #[serde(default)]
358 version: Option<String>,
359 #[serde(default)]
360 targets: Vec<String>,
361 }
362
363 async fn build(
364 State(s): State<AppState>,
365 Json(body): Json<BuildBody>,
366 ) -> Result<Json<serde_json::Value>> {
367 let app = AppId::new(body.app);
368 let targets = parse_targets(body.targets)?;
369 let version = runner::resolve_version(&s, &app, body.version)?;
370 let targets = runner::resolve_targets(&s, &app, targets)?;
371 let build_id = runner::start_build(s, app, version.clone(), targets)
372 .await
373 .map_err(Error::Other)?;
374 Ok(Json(
375 serde_json::json!({ "accepted": true, "build_id": build_id, "version": version.to_string() }),
376 ))
377 }
378
379 #[derive(Deserialize)]
380 struct RetryBody {
381 app: String,
382 target: String,
383 #[serde(default)]
384 version: Option<String>,
385 }
386
387 async fn retry(
388 State(s): State<AppState>,
389 Json(body): Json<RetryBody>,
390 ) -> Result<Json<serde_json::Value>> {
391 let app = AppId::new(body.app);
392 let target: Target = body.target.parse().map_err(Error::BadRequest)?;
393 let version = runner::resolve_version(&s, &app, body.version)?;
394 let targets = runner::resolve_targets(&s, &app, vec![target])?;
395 let build_id = runner::start_build(s, app, version.clone(), targets)
396 .await
397 .map_err(Error::Other)?;
398 Ok(Json(
399 serde_json::json!({ "accepted": true, "build_id": build_id, "target": target.to_string() }),
400 ))
401 }
402
403 fn parse_targets(raw: Vec<String>) -> Result<Vec<Target>> {
404 raw.into_iter()
405 .map(|t| t.parse::<Target>().map_err(Error::BadRequest))
406 .collect()
407 }
408
409 async fn get_step_log(
410 State(s): State<AppState>,
411 Path((app, version, target, step)): Path<(String, String, String, String)>,
412 ) -> Result<axum::response::Response> {
413 fn safe(seg: &str) -> bool {
414 !seg.is_empty() && !seg.contains('/') && !seg.contains('\\') && seg != "." && seg != ".."
415 }
416 if ![&app, &version, &target, &step]
417 .into_iter()
418 .all(|s| safe(s))
419 {
420 return Err(Error::NotFound);
421 }
422 let path = s
423 .cfg
424 .logs_root
425 .join(&app)
426 .join(&version)
427 .join(&target)
428 .join(format!("{step}.log"));
429 // Stream the log in chunks rather than reading the whole (potentially large,
430 // verbose-build) file into memory.
431 let file = match tokio::fs::File::open(&path).await {
432 Ok(f) => f,
433 Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Err(Error::NotFound),
434 Err(e) => return Err(Error::Other(e.into())),
435 };
436 let (tx, rx) =
437 tokio::sync::mpsc::channel::<std::result::Result<axum::body::Bytes, std::io::Error>>(8);
438 tokio::spawn(async move {
439 use tokio::io::AsyncReadExt;
440 let mut file = file;
441 let mut buf = vec![0u8; 64 * 1024];
442 loop {
443 match file.read(&mut buf).await {
444 Ok(0) => break,
445 Ok(n) => {
446 if tx
447 .send(Ok(axum::body::Bytes::copy_from_slice(&buf[..n])))
448 .await
449 .is_err()
450 {
451 break;
452 }
453 }
454 Err(e) => {
455 let _ = tx.send(Err(e)).await;
456 break;
457 }
458 }
459 }
460 });
461 let body = axum::body::Body::from_stream(tokio_stream::wrappers::ReceiverStream::new(rx));
462 Ok((
463 [(
464 axum::http::header::CONTENT_TYPE,
465 "text/plain; charset=utf-8",
466 )],
467 body,
468 )
469 .into_response())
470 }
471
472 async fn events_ws(ws: WebSocketUpgrade, State(s): State<AppState>) -> impl IntoResponse {
473 use axum::extract::ws::Message;
474 use tokio::sync::broadcast::error::RecvError;
475
476 ws.on_upgrade(move |mut socket| async move {
477 // Subscribe to both channels and merge them: a lag on the high-rate log
478 // stream emits its own `lagged` frame without dropping anything on the
479 // status stream (and vice versa), so a busy build's chunk firehose can't
480 // evict a TargetFailed/PublishOk the operator needs to see.
481 let mut status_rx = s.events.subscribe_status();
482 let mut logs_rx = s.events.subscribe_logs();
483 loop {
484 let recv = tokio::select! {
485 r = status_rx.recv() => r,
486 r = logs_rx.recv() => r,
487 };
488 match recv {
489 Ok(env) => {
490 let json = match serde_json::to_string(&env) {
491 Ok(s) => s,
492 Err(e) => {
493 tracing::warn!(error = %e, "events ws: serialize failed");
494 continue;
495 }
496 };
497 if socket.send(Message::Text(json.into())).await.is_err() {
498 break;
499 }
500 }
501 Err(RecvError::Lagged(n)) => {
502 let _ = socket
503 .send(Message::Text(
504 format!(r#"{{"kind":"lagged","skipped":{n}}}"#).into(),
505 ))
506 .await;
507 }
508 Err(RecvError::Closed) => break,
509 }
510 }
511 })
512 }
513
514 #[cfg(test)]
515 mod tests {
516 use super::*;
517 use crate::config::Config;
518 use crate::ota::OtaRegistry;
519 use crate::topology::Topology;
520 use axum::body::Body;
521 use axum::http::{Request, StatusCode};
522 use http_body_util::BodyExt;
523 use std::collections::HashMap;
524 use std::sync::Arc;
525 use tokio::sync::Mutex;
526 use tower::ServiceExt;
527
528 async fn test_state(root: &std::path::Path) -> AppState {
529 let cfg = Config::for_tests(root);
530 let pool = crate::db::open(&cfg.db_path).await.unwrap();
531 let repo = root.join("goingson");
532 std::fs::create_dir_all(&repo).unwrap();
533 std::fs::write(repo.join("bento.toml"), "targets = [\"linux/x86_64\"]\n").unwrap();
534 let topo = Topology::from_str_for_tests(&format!(
535 r#"
536 [[host]]
537 name = "fw13"
538 ssh = "local"
539 targets = ["linux/x86_64"]
540
541 [app.goingson]
542 repo = "{}"
543 "#,
544 repo.display()
545 ))
546 .unwrap();
547 let executors = Arc::new(crate::state::build_executors(&topo));
548 let syncs = Arc::new(crate::state::build_syncs(&topo));
549 let host_locks = crate::state::build_host_locks(&topo);
550 AppState {
551 pool,
552 topo: Arc::new(topo),
553 cfg: Arc::new(cfg),
554 prom: crate::metrics::test_handle(),
555 events: crate::events::channel(),
556 ota: Arc::new(OtaRegistry::standard("https://makenot.work")),
557 executors,
558 syncs,
559 active: Arc::new(Mutex::new(HashMap::new())),
560 api_token: None,
561 host_locks,
562 }
563 }
564
565 async fn body_string(resp: axum::response::Response) -> String {
566 let bytes = resp.into_body().collect().await.unwrap().to_bytes();
567 String::from_utf8(bytes.to_vec()).unwrap()
568 }
569
570 #[tokio::test]
571 async fn state_is_empty_initially() {
572 let tmp = tempfile::tempdir().unwrap();
573 let app = router(test_state(tmp.path()).await);
574 let resp = app
575 .oneshot(
576 Request::builder()
577 .uri("/state")
578 .body(Body::empty())
579 .unwrap(),
580 )
581 .await
582 .unwrap();
583 assert_eq!(resp.status(), StatusCode::OK);
584 assert_eq!(body_string(resp).await, r#"{"build":null}"#);
585 }
586
587 #[tokio::test]
588 async fn status_json_lists_every_declared_app_before_any_build() {
589 // The mapping is tested in `crate::status`. This asserts the route is
590 // wired and that an app with no build history still reaches the
591 // surface -- an app missing from the viewer is indistinguishable from
592 // an app that is fine.
593 let tmp = tempfile::tempdir().unwrap();
594 let app = router(test_state(tmp.path()).await);
595 let resp = app
596 .oneshot(
597 Request::builder()
598 .uri("/status.json")
599 .body(Body::empty())
600 .unwrap(),
601 )
602 .await
603 .unwrap();
604 assert_eq!(resp.status(), StatusCode::OK);
605
606 let payload: ops_status::Payload = serde_json::from_str(&body_string(resp).await).unwrap();
607 assert_eq!(payload.source, crate::status::SOURCE);
608 assert_eq!(payload.schema, ops_status::SCHEMA_VERSION);
609 assert_eq!(payload.validate(), Ok(()));
610
611 let goingson = payload.node("app:goingson").expect("declared app appears");
612 assert_eq!(goingson.status, ops_status::Status::Pending);
613 assert!(payload.actions.contains_key("build-goingson"));
614 assert!(payload.node("target:goingson:linux/x86_64").is_some());
615 }
616
617 // ---- CF2: bearer-token auth on build triggers ----
618
619 #[tokio::test]
620 async fn build_route_requires_bearer_when_token_set() {
621 let tmp = tempfile::tempdir().unwrap();
622 let mut state = test_state(tmp.path()).await;
623 state.api_token = Some(std::sync::Arc::from("s3cr3t"));
624 let app = router(state);
625
626 let build_req = || {
627 Request::builder()
628 .method("POST")
629 .uri("/build")
630 .header("content-type", "application/json")
631 .body(Body::from(r#"{"app":"goingson"}"#))
632 .unwrap()
633 };
634
635 // No token -> 401.
636 let resp = app.clone().oneshot(build_req()).await.unwrap();
637 assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
638
639 // Wrong token -> 401.
640 let mut bad = build_req();
641 bad.headers_mut()
642 .insert("authorization", "Bearer nope".parse().unwrap());
643 let resp = app.clone().oneshot(bad).await.unwrap();
644 assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
645
646 // Correct token -> passes auth (NOT 401; the build itself is accepted).
647 let mut good = build_req();
648 good.headers_mut()
649 .insert("authorization", "Bearer s3cr3t".parse().unwrap());
650 let resp = app.clone().oneshot(good).await.unwrap();
651 assert_ne!(resp.status(), StatusCode::UNAUTHORIZED);
652 }
653
654 #[tokio::test]
655 async fn read_route_open_when_token_set() {
656 let tmp = tempfile::tempdir().unwrap();
657 let mut state = test_state(tmp.path()).await;
658 state.api_token = Some(std::sync::Arc::from("s3cr3t"));
659 let app = router(state);
660 let resp = app
661 .oneshot(
662 Request::builder()
663 .uri("/state")
664 .body(Body::empty())
665 .unwrap(),
666 )
667 .await
668 .unwrap();
669 assert_eq!(resp.status(), StatusCode::OK);
670 }
671
672 #[tokio::test]
673 async fn step_log_rejects_traversal() {
674 let tmp = tempfile::tempdir().unwrap();
675 let app = router(test_state(tmp.path()).await);
676 let resp = app
677 .oneshot(
678 Request::builder()
679 .uri("/logs/goingson/0.4.1/..%2f..%2fetc/passwd")
680 .body(Body::empty())
681 .unwrap(),
682 )
683 .await
684 .unwrap();
685 assert_eq!(resp.status(), StatusCode::NOT_FOUND);
686 }
687
688 #[tokio::test]
689 async fn step_log_streams_existing_file() {
690 let tmp = tempfile::tempdir().unwrap();
691 let state = test_state(tmp.path()).await;
692 // Write a log at the path the route resolves.
693 let dir = state
694 .cfg
695 .logs_root
696 .join("goingson")
697 .join("0.4.1")
698 .join("linux-x86_64");
699 std::fs::create_dir_all(&dir).unwrap();
700 std::fs::write(dir.join("build.log"), b"compiling\nlinked\n").unwrap();
701 let app = router(state);
702 let resp = app
703 .oneshot(
704 Request::builder()
705 .uri("/logs/goingson/0.4.1/linux-x86_64/build")
706 .body(Body::empty())
707 .unwrap(),
708 )
709 .await
710 .unwrap();
711 assert_eq!(resp.status(), StatusCode::OK);
712 assert_eq!(body_string(resp).await, "compiling\nlinked\n");
713 }
714
715 #[tokio::test]
716 async fn bad_request_body_is_a_json_error_envelope() {
717 let tmp = tempfile::tempdir().unwrap();
718 let app = router(test_state(tmp.path()).await);
719 let resp = app
720 .oneshot(
721 Request::builder()
722 .method("POST")
723 .uri("/build")
724 .header("content-type", "application/json")
725 .body(Body::from(r#"{"app":"nope","targets":["linux/x86_64"]}"#))
726 .unwrap(),
727 )
728 .await
729 .unwrap();
730 assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
731 // JSON envelope, message preserved.
732 let body = body_string(resp).await;
733 let v: serde_json::Value = serde_json::from_str(&body).unwrap();
734 assert!(v["error"].as_str().unwrap().contains("unknown app"));
735 }
736
737 #[tokio::test]
738 async fn build_rejects_unshipped_target_as_bad_request() {
739 // windows/x86_64 is a valid target but the test app doesn't ship it.
740 // That's a client error (400), not a server error (500).
741 let tmp = tempfile::tempdir().unwrap();
742 let app = router(test_state(tmp.path()).await);
743 let resp = app
744 .oneshot(
745 Request::builder()
746 .method("POST")
747 .uri("/build")
748 .header("content-type", "application/json")
749 // Explicit version so resolve_version doesn't read a
750 // (nonexistent) tauri.conf first — we're testing the target
751 // validation specifically.
752 .body(Body::from(
753 r#"{"app":"goingson","version":"0.4.1","targets":["windows/x86_64"]}"#,
754 ))
755 .unwrap(),
756 )
757 .await
758 .unwrap();
759 assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
760 assert!(
761 body_string(resp).await.contains("does not ship target"),
762 "names the problem"
763 );
764 }
765
766 #[tokio::test]
767 async fn build_rejects_unknown_app_as_bad_request() {
768 let tmp = tempfile::tempdir().unwrap();
769 let app = router(test_state(tmp.path()).await);
770 let resp = app
771 .oneshot(
772 Request::builder()
773 .method("POST")
774 .uri("/build")
775 .header("content-type", "application/json")
776 .body(Body::from(r#"{"app":"nope","targets":["linux/x86_64"]}"#))
777 .unwrap(),
778 )
779 .await
780 .unwrap();
781 assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
782 assert!(
783 body_string(resp).await.contains("unknown app"),
784 "names the problem"
785 );
786 }
787
788 #[tokio::test]
789 async fn build_rejects_malformed_target_as_bad_request() {
790 // A non-parseable target string is also a 400 (via parse_targets).
791 let tmp = tempfile::tempdir().unwrap();
792 let app = router(test_state(tmp.path()).await);
793 let resp = app
794 .oneshot(
795 Request::builder()
796 .method("POST")
797 .uri("/build")
798 .header("content-type", "application/json")
799 .body(Body::from(
800 r#"{"app":"goingson","targets":["not-a-target"]}"#,
801 ))
802 .unwrap(),
803 )
804 .await
805 .unwrap();
806 assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
807 }
808
809 /// A two-target app + a two-host topology, for the release-view aggregation.
810 async fn two_target_state(root: &std::path::Path) -> AppState {
811 let cfg = Config::for_tests(root);
812 let pool = crate::db::open(&cfg.db_path).await.unwrap();
813 let repo = root.join("demo");
814 std::fs::create_dir_all(&repo).unwrap();
815 std::fs::write(
816 repo.join("bento.toml"),
817 "targets = [\"linux/x86_64\", \"macos/aarch64\"]\n",
818 )
819 .unwrap();
820 let topo = Topology::from_str_for_tests(&format!(
821 "[[host]]\nname = \"fw13\"\nssh = \"local\"\ntargets = [\"linux/x86_64\"]\n\
822 [[host]]\nname = \"mbp\"\nssh = \"mbp\"\ntargets = [\"macos/aarch64\"]\n\
823 [app.demo]\nrepo = \"{}\"\n",
824 repo.display()
825 ))
826 .unwrap();
827 let executors = Arc::new(crate::state::build_executors(&topo));
828 let syncs = Arc::new(crate::state::build_syncs(&topo));
829 let host_locks = crate::state::build_host_locks(&topo);
830 AppState {
831 pool,
832 topo: Arc::new(topo),
833 cfg: Arc::new(cfg),
834 prom: crate::metrics::test_handle(),
835 events: crate::events::channel(),
836 ota: Arc::new(OtaRegistry::standard("https://makenot.work")),
837 executors,
838 syncs,
839 active: Arc::new(Mutex::new(HashMap::new())),
840 api_token: None,
841 host_locks,
842 }
843 }
844
845 async fn insert_target_run(pool: &sqlx::SqlitePool, ver: &str, target: &str, status: &str) {
846 let bid: i64 = sqlx::query_scalar(
847 "INSERT INTO builds (app, version, status, created_at) VALUES ('demo', ?, ?, '2026-07-23T00:00:00Z') RETURNING id",
848 )
849 .bind(ver)
850 .bind(status)
851 .fetch_one(pool)
852 .await
853 .unwrap();
854 sqlx::query(
855 "INSERT INTO target_runs (build_id, app, version, target, status, started_at)
856 VALUES (?, 'demo', ?, ?, ?, '2026-07-23T00:00:00Z')",
857 )
858 .bind(bid)
859 .bind(ver)
860 .bind(target)
861 .bind(status)
862 .execute(pool)
863 .await
864 .unwrap();
865 }
866
867 /// The audit's H2: `/retry` inserts a whole new single-target build, so a
868 /// build-id-keyed view collapses a multi-target release to the retried row.
869 /// The release view aggregates by (app, version), so a green retry of one
870 /// target folds into that cell while the other target stays visible.
871 #[tokio::test]
872 async fn release_view_survives_a_single_target_retry() {
873 let tmp = tempfile::tempdir().unwrap();
874 let state = two_target_state(tmp.path()).await;
875 let pool = state.pool.clone();
876
877 // Original release build: linux ok, macos failed.
878 insert_target_run(&pool, "0.5.0", "linux/x86_64", "ok").await;
879 insert_target_run(&pool, "0.5.0", "macos/aarch64", "failed").await;
880 // Retry macos only — a fresh single-target build that succeeds.
881 insert_target_run(&pool, "0.5.0", "macos/aarch64", "ok").await;
882
883 let app = router(state);
884 let resp = app
885 .oneshot(
886 Request::builder()
887 .uri("/release/demo/0.5.0")
888 .body(Body::empty())
889 .unwrap(),
890 )
891 .await
892 .unwrap();
893 assert_eq!(resp.status(), StatusCode::OK);
894 let v: serde_json::Value = serde_json::from_str(&body_string(resp).await).unwrap();
895
896 // Both targets are present (the matrix did not collapse to the retry),
897 // and macos shows the newer `ok`, not the earlier `failed`.
898 let targets = v["build"]["targets"].as_array().unwrap();
899 assert_eq!(targets.len(), 2, "full matrix preserved across the retry");
900 let macos = targets
901 .iter()
902 .find(|t| t["target"] == "macos/aarch64")
903 .unwrap();
904 assert_eq!(macos["status"], "ok", "retry result wins the cell");
905 assert_eq!(v["all_targets_green"], true);
906 }
907
908 #[tokio::test]
909 async fn release_view_404s_for_an_unbuilt_version() {
910 let tmp = tempfile::tempdir().unwrap();
911 let app = router(two_target_state(tmp.path()).await);
912 let resp = app
913 .oneshot(
914 Request::builder()
915 .uri("/release/demo/9.9.9")
916 .body(Body::empty())
917 .unwrap(),
918 )
919 .await
920 .unwrap();
921 assert_eq!(resp.status(), StatusCode::NOT_FOUND);
922 }
923
924 // ---- /retry handler (the route itself, distinct from the release-view
925 // aggregation a retry produces above) ----
926
927 fn retry_req(json: &str) -> Request<Body> {
928 Request::builder()
929 .method("POST")
930 .uri("/retry")
931 .header("content-type", "application/json")
932 .body(Body::from(json.to_string()))
933 .unwrap()
934 }
935
936 #[tokio::test]
937 async fn retry_accepts_a_single_shipped_target() {
938 let tmp = tempfile::tempdir().unwrap();
939 let app = router(test_state(tmp.path()).await);
940 // Explicit version so the handler doesn't read a (nonexistent) version
941 // file; the point here is the accept path, not version resolution.
942 let resp = app
943 .oneshot(retry_req(
944 r#"{"app":"goingson","target":"linux/x86_64","version":"0.4.1"}"#,
945 ))
946 .await
947 .unwrap();
948 assert_eq!(resp.status(), StatusCode::OK);
949 let v: serde_json::Value = serde_json::from_str(&body_string(resp).await).unwrap();
950 assert_eq!(v["accepted"], true);
951 assert_eq!(v["target"], "linux/x86_64", "echoes the retried target");
952 assert!(
953 v["build_id"].as_i64().is_some(),
954 "a fresh single-target build id is returned"
955 );
956 }
957
958 #[tokio::test]
959 async fn retry_rejects_an_unshipped_target_as_bad_request() {
960 // goingson does not ship windows; a retry of it is a 400, not a 500.
961 let tmp = tempfile::tempdir().unwrap();
962 let app = router(test_state(tmp.path()).await);
963 let resp = app
964 .oneshot(retry_req(
965 r#"{"app":"goingson","target":"windows/x86_64","version":"0.4.1"}"#,
966 ))
967 .await
968 .unwrap();
969 assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
970 assert!(body_string(resp).await.contains("does not ship target"));
971 }
972
973 #[tokio::test]
974 async fn retry_rejects_a_malformed_target_as_bad_request() {
975 let tmp = tempfile::tempdir().unwrap();
976 let app = router(test_state(tmp.path()).await);
977 let resp = app
978 .oneshot(retry_req(
979 r#"{"app":"goingson","target":"not-a-target","version":"0.4.1"}"#,
980 ))
981 .await
982 .unwrap();
983 assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
984 }
985
986 #[tokio::test]
987 async fn retry_rejects_an_unknown_app_as_bad_request() {
988 let tmp = tempfile::tempdir().unwrap();
989 let app = router(test_state(tmp.path()).await);
990 let resp = app
991 .oneshot(retry_req(r#"{"app":"nope","target":"linux/x86_64"}"#))
992 .await
993 .unwrap();
994 assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
995 assert!(body_string(resp).await.contains("unknown app"));
996 }
997
998 // ---- /events websocket: the HTTP->WS wiring (upgrade + serialize + send).
999 // The channel merge and lagged framing themselves are covered in
1000 // `crate::events`; this asserts a real client on `/events` receives an
1001 // emitted event as the flat-`kind` JSON the TUI parses. ----
1002
1003 #[tokio::test]
1004 async fn events_ws_streams_emitted_events_as_flat_json() {
1005 use futures_util::StreamExt;
1006 use tokio_tungstenite::tungstenite::Message as WsMessage;
1007
1008 let tmp = tempfile::tempdir().unwrap();
1009 let state = test_state(tmp.path()).await;
1010 let events = state.events.clone();
1011 let app = router(state);
1012
1013 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
1014 let addr = listener.local_addr().unwrap();
1015 tokio::spawn(async move {
1016 axum::serve(listener, app).await.unwrap();
1017 });
1018
1019 let (mut ws, _resp) = tokio_tungstenite::connect_async(format!("ws://{addr}/events"))
1020 .await
1021 .expect("client connects to /events");
1022
1023 // The handler subscribes to the bus inside `on_upgrade`, which can land
1024 // just after the handshake returns; a broadcast has no backlog for a
1025 // late subscriber, so re-emit until the first frame arrives rather than
1026 // racing that window with a single send.
1027 let ev = crate::events::Event::PublishFailed {
1028 app: AppId::new("goingson"),
1029 target: "macos/aarch64".parse().unwrap(),
1030 channel: "stable".into(),
1031 error: "boom".into(),
1032 };
1033 let mut frame = None;
1034 for _ in 0..100 {
1035 crate::events::emit(&events, ev.clone());
1036 if let Ok(Some(Ok(WsMessage::Text(t)))) =
1037 tokio::time::timeout(std::time::Duration::from_millis(50), ws.next()).await
1038 {
1039 frame = Some(t);
1040 break;
1041 }
1042 }
1043 let frame = frame.expect("received an event frame within the retry budget");
1044 let v: serde_json::Value = serde_json::from_str(&frame).unwrap();
1045 assert_eq!(v["kind"], "publish_failed", "flat kind-tagged wire shape");
1046 assert_eq!(v["channel"], "stable");
1047 assert!(v.get("event").is_none(), "not nested under `event`");
1048 }
1049 }
1050