Skip to main content

max / makenotwork

11.6 KB · 356 lines History Blame Raw
1 use crate::error::Result;
2 use crate::state::AppState;
3 use axum::extract::{Path, State, WebSocketUpgrade};
4 use axum::response::IntoResponse;
5 use axum::routing::{get, post};
6 use axum::{Json, Router};
7 use serde::{Deserialize, Serialize};
8 use sqlx::Row;
9
10 pub fn router(state: AppState) -> Router {
11 let prom = state.prom.clone();
12 Router::new()
13 .route("/state", get(get_state))
14 .route("/promote/{tier}", post(promote))
15 .route("/rollback/{tier}", post(rollback))
16 .route("/rebuild", post(rebuild))
17 .route("/backup/fetch", post(backup_fetch))
18 .route("/events", get(events_ws))
19 .with_state(state)
20 .route("/metrics", get(crate::metrics::render).with_state(prom))
21 }
22
23 #[derive(Serialize)]
24 struct StateView {
25 tiers: Vec<TierView>,
26 }
27
28 #[derive(Serialize)]
29 struct TierView {
30 name: String,
31 ord: i64,
32 provisioned: bool,
33 canary: String,
34 current_version: Option<String>,
35 previous_version: Option<String>,
36 burn_in_started_at: Option<String>,
37 nodes: Vec<String>,
38 gates: Vec<GateView>,
39 }
40
41 #[derive(Serialize)]
42 struct GateView {
43 kind: String,
44 passed: Option<bool>,
45 finished_at: Option<String>,
46 detail: Option<String>,
47 }
48
49 async fn get_state(State(s): State<AppState>) -> Result<Json<StateView>> {
50 let rows = sqlx::query(
51 "SELECT t.name, t.ord, t.provisioned, t.canary,
52 ts.current_version, ts.previous_version, ts.burn_in_started_at
53 FROM tiers t
54 LEFT JOIN tier_state ts ON ts.tier = t.name
55 ORDER BY t.ord",
56 )
57 .fetch_all(&s.pool)
58 .await?;
59
60 let mut tiers = Vec::with_capacity(rows.len());
61 for r in rows {
62 let name: String = r.get("name");
63 let current_version: Option<String> = r.get("current_version");
64
65 let nodes: Vec<String> = sqlx::query_scalar("SELECT name FROM nodes WHERE tier = ? ORDER BY name")
66 .bind(&name)
67 .fetch_all(&s.pool)
68 .await?;
69
70 let gates: Vec<GateView> = if let Some(ver) = current_version.as_ref() {
71 // Most recent gate_runs row per gate_kind for (tier, current_version).
72 sqlx::query(
73 "SELECT gate_kind, passed, finished_at, detail
74 FROM gate_runs g
75 WHERE tier = ?1 AND version = ?2
76 AND id = (SELECT MAX(id) FROM gate_runs
77 WHERE tier = ?1 AND version = ?2 AND gate_kind = g.gate_kind)
78 ORDER BY gate_kind",
79 )
80 .bind(&name)
81 .bind(ver)
82 .fetch_all(&s.pool)
83 .await?
84 .into_iter()
85 .map(|gr| GateView {
86 kind: gr.get("gate_kind"),
87 passed: gr.get::<Option<i64>, _>("passed").map(|v| v != 0),
88 finished_at: gr.get("finished_at"),
89 detail: gr.get("detail"),
90 })
91 .collect()
92 } else {
93 Vec::new()
94 };
95
96 tiers.push(TierView {
97 name,
98 ord: r.get("ord"),
99 provisioned: r.get::<i64, _>("provisioned") != 0,
100 canary: r.get("canary"),
101 current_version,
102 previous_version: r.get("previous_version"),
103 burn_in_started_at: r.get("burn_in_started_at"),
104 nodes,
105 gates,
106 });
107 }
108
109 Ok(Json(StateView { tiers }))
110 }
111
112 #[derive(Deserialize)]
113 struct PromoteBody {
114 version: String,
115 #[serde(default)]
116 hotfix: bool,
117 #[serde(default)]
118 reset_burn_in: bool,
119 }
120
121 async fn promote(
122 State(s): State<AppState>,
123 Path(tier): Path<String>,
124 Json(body): Json<PromoteBody>,
125 ) -> Result<Json<serde_json::Value>> {
126 let idx = s.topo.tiers.iter().position(|t| t.name == tier)
127 .ok_or(crate::error::Error::NotFound)?;
128 if idx == 0 {
129 return Err(crate::error::Error::GateBlocked(
130 "cannot /promote to the first tier; use /rebuild".into(),
131 ));
132 }
133 let target = &s.topo.tiers[idx];
134 let source = &s.topo.tiers[idx - 1];
135
136 // 1. Predecessor must have all of its gates green for this version (with
137 // optional hotfix override that skips burn_in).
138 let pending = unsatisfied_gates(&s.pool, &source.name, &body.version, body.hotfix).await?;
139 if !pending.is_empty() {
140 return Err(crate::error::Error::GateBlocked(format!(
141 "{} gate(s) not satisfied on tier {}: {}",
142 pending.len(),
143 source.name,
144 pending.join(", "),
145 )));
146 }
147
148 // 2. Look up the artifact for this version.
149 let bin: Option<(String,)> = sqlx::query_as(
150 "SELECT artifact_path FROM versions WHERE version = ?",
151 )
152 .bind(&body.version)
153 .fetch_optional(&s.pool)
154 .await
155 .map_err(crate::error::Error::Db)?;
156 let Some((bin,)) = bin else {
157 return Err(crate::error::Error::NotFound);
158 };
159 let bin_path = std::path::PathBuf::from(bin);
160
161 // 3. Deploy to each node. Sequential canary is the only policy
162 // implemented in v0; parallel is a one-line change once we trust the
163 // sequential path.
164 for node in &target.nodes {
165 crate::deploy::deploy_node(node, &body.version, &bin_path)
166 .await
167 .map_err(crate::error::Error::Other)?;
168 let now = chrono::Utc::now().to_rfc3339();
169 sqlx::query(
170 "INSERT INTO deploys (version, tier, node, started_at, finished_at, outcome, hotfix, reset_burn_in)
171 VALUES (?, ?, ?, ?, ?, 'ok', ?, ?)",
172 )
173 .bind(&body.version).bind(&target.name).bind(&node.name)
174 .bind(&now).bind(&now)
175 .bind(body.hotfix as i64).bind(body.reset_burn_in as i64)
176 .execute(&s.pool).await.map_err(crate::error::Error::Db)?;
177 }
178
179 // 4. Advance tier_state. burn_in_started_at is set to now so the target
180 // tier's burn_in gate starts ticking. reset_burn_in on the *source*
181 // tier nulls its clock only when the operator explicitly asked for it.
182 let prev: Option<String> = sqlx::query_scalar(
183 "SELECT current_version FROM tier_state WHERE tier = ?",
184 )
185 .bind(&target.name)
186 .fetch_optional(&s.pool).await.map_err(crate::error::Error::Db)?.flatten();
187 sqlx::query(
188 "UPDATE tier_state SET previous_version = ?, current_version = ?, burn_in_started_at = ?
189 WHERE tier = ?",
190 )
191 .bind(prev)
192 .bind(&body.version)
193 .bind(chrono::Utc::now().to_rfc3339())
194 .bind(&target.name)
195 .execute(&s.pool).await.map_err(crate::error::Error::Db)?;
196
197 if body.reset_burn_in {
198 sqlx::query("UPDATE tier_state SET burn_in_started_at = NULL WHERE tier = ?")
199 .bind(&source.name)
200 .execute(&s.pool).await.map_err(crate::error::Error::Db)?;
201 }
202
203 tracing::info!(
204 version = %body.version, tier = %target.name,
205 hotfix = body.hotfix, reset_burn_in = body.reset_burn_in,
206 "promote complete",
207 );
208
209 Ok(Json(serde_json::json!({
210 "tier": target.name,
211 "version": body.version,
212 "nodes_deployed": target.nodes.iter().map(|n| n.name.clone()).collect::<Vec<_>>(),
213 })))
214 }
215
216 /// Returns the kinds of gates on `tier` that have not (yet) passed for
217 /// `version`. `hotfix` suppresses the burn_in requirement only.
218 async fn unsatisfied_gates(
219 pool: &sqlx::SqlitePool,
220 tier: &str,
221 version: &str,
222 hotfix: bool,
223 ) -> std::result::Result<Vec<String>, crate::error::Error> {
224 // We need the configured gate list for the tier to know what *should*
225 // pass. The route handler has Topology in hand and could pass it in, but
226 // the DB also captures it implicitly via gate_runs rows. Simplest correct
227 // answer: re-read from topology via tier name; the caller has it.
228 // For now we inspect the latest gate_runs.
229 let rows: Vec<(String, Option<i64>)> = sqlx::query_as(
230 "SELECT gate_kind, passed FROM gate_runs g
231 WHERE tier = ?1 AND version = ?2
232 AND id = (SELECT MAX(id) FROM gate_runs
233 WHERE tier = ?1 AND version = ?2 AND gate_kind = g.gate_kind)",
234 )
235 .bind(tier).bind(version)
236 .fetch_all(pool).await.map_err(crate::error::Error::Db)?;
237 let mut bad = Vec::new();
238 for (kind, passed) in rows {
239 if hotfix && kind == "burn_in" {
240 continue;
241 }
242 if passed.unwrap_or(0) == 0 {
243 bad.push(kind);
244 }
245 }
246 Ok(bad)
247 }
248
249 async fn rollback(
250 State(s): State<AppState>,
251 Path(tier): Path<String>,
252 ) -> Result<Json<serde_json::Value>> {
253 let target = s.topo.tiers.iter().find(|t| t.name == tier)
254 .ok_or(crate::error::Error::NotFound)?;
255
256 let row: Option<(Option<String>, Option<String>)> = sqlx::query_as(
257 "SELECT current_version, previous_version FROM tier_state WHERE tier = ?",
258 )
259 .bind(&tier)
260 .fetch_optional(&s.pool).await.map_err(crate::error::Error::Db)?;
261 let (Some(current), Some(previous)) = row.unwrap_or((None, None)) else {
262 return Err(crate::error::Error::GateBlocked(
263 "no previous_version to roll back to".into(),
264 ));
265 };
266
267 let bin: Option<(String,)> = sqlx::query_as(
268 "SELECT artifact_path FROM versions WHERE version = ?",
269 )
270 .bind(&previous)
271 .fetch_optional(&s.pool).await.map_err(crate::error::Error::Db)?;
272 let Some((bin,)) = bin else {
273 return Err(crate::error::Error::GateBlocked(
274 format!("previous version {previous} has no artifact_path; rollback impossible"),
275 ));
276 };
277 let bin_path = std::path::PathBuf::from(bin);
278
279 for node in &target.nodes {
280 crate::deploy::deploy_node(node, &previous, &bin_path)
281 .await
282 .map_err(crate::error::Error::Other)?;
283 }
284
285 sqlx::query(
286 "UPDATE tier_state SET current_version = ?, previous_version = ?, burn_in_started_at = NULL
287 WHERE tier = ?",
288 )
289 .bind(&previous)
290 .bind(&current)
291 .bind(&tier)
292 .execute(&s.pool).await.map_err(crate::error::Error::Db)?;
293
294 tracing::warn!(tier = %tier, from = %current, to = %previous, "rollback complete");
295
296 Ok(Json(serde_json::json!({
297 "tier": tier,
298 "rolled_back_from": current,
299 "now_running": previous,
300 })))
301 }
302
303 #[derive(Deserialize, Default)]
304 struct RebuildBody {
305 /// Specific sha to build. If absent, resolve `topo.repo.branch` from the bare repo.
306 #[serde(default)]
307 sha: Option<String>,
308 }
309
310 async fn rebuild(
311 State(s): State<AppState>,
312 body: Option<Json<RebuildBody>>,
313 ) -> Result<Json<serde_json::Value>> {
314 let body = body.map(|Json(b)| b).unwrap_or_default();
315 let sha = match body.sha {
316 Some(s) => s,
317 None => crate::git::resolve_ref(
318 std::path::Path::new(&s.topo.repo.bare_path),
319 &s.topo.repo.branch,
320 )
321 .await
322 .map_err(crate::error::Error::Other)?,
323 };
324
325 tracing::info!(sha = %sha, "rebuild requested");
326
327 let pool = s.pool.clone();
328 let cfg = s.cfg.clone();
329 let topo = s.topo.clone();
330 let sha_for_task = sha.clone();
331 tokio::spawn(async move {
332 if let Err(e) = crate::build::build_and_run_mm(pool, cfg, topo, sha_for_task.clone()).await {
333 tracing::error!(sha = %sha_for_task, error = %e, "rebuild pipeline failed");
334 }
335 });
336
337 Ok(Json(serde_json::json!({ "accepted": true, "sha": sha })))
338 }
339
340 async fn backup_fetch(State(s): State<AppState>) -> Result<Json<serde_json::Value>> {
341 let fb = crate::backup::fetch(&s.pool, &s.cfg, &s.topo)
342 .await
343 .map_err(crate::error::Error::Other)?;
344 Ok(Json(serde_json::json!({
345 "source": fb.source,
346 "local_path": fb.local_path,
347 "byte_size": fb.byte_size,
348 })))
349 }
350
351 async fn events_ws(ws: WebSocketUpgrade, State(_s): State<AppState>) -> impl IntoResponse {
352 ws.on_upgrade(|_socket| async move {
353 // tail of deploy/gate events for the TUI
354 })
355 }
356