zuka
zuka/src/api/runs.rs

worklyn / zukapublic

Agent-first git hosting. One Rust binary: git over HTTP and SSH, a REST API, MCP, CI, and multi-tenant isolation.

Get a copy: git clone https://zuka.worklyn.com/worklyn/zuka.git
branchbc985d6bbd321be340e3750b3da4734ee1553cbd ▾
main All branches →
zuka/src/api/runs.rs
RSruns.rs5.8 KBDownload
1// CI runs.
2//
3// Cancel is `PATCH ... {"status":"cancelled"}`, not `DELETE`: the run still exists
4// afterwards and is still readable, so deleting it would be the wrong verb for what
5// happens.
6//
7// Logs are two endpoints rather than one that negotiates on run state. Serving SSE
8// while running and plain text once finished makes an identical request's content
9// type depend on a race with the executor, which is not content negotiation.
10
11use crate::account::Identity;
12use crate::ci::run::{Run, Status};
13use crate::error::{Error, Result};
14use crate::git::validate::{self, Name};
15use crate::http::response::{self, Body};
16use crate::http::{request, AppState};
17use hyper::body::Incoming;
18use hyper::{Method, Request, Response, StatusCode};
19use serde::Deserialize;
20use std::sync::Arc;
21
22#[derive(Debug, Deserialize)]
23struct PatchRun {
24 status: Status,
25}
26
27pub async fn route(
28 request: Request<Incoming>,
29 state: Arc<AppState>,
30 identity: Identity,
31 account: &str,
32 repo: &str,
33 tail: &[&str],
34) -> Result<Response<Body>> {
35 let account = validate::name(account)?;
36 let repo = validate::name(repo)?;
37 identity.require_read(&account, &repo)?;
38
39 if state.meta.get_ready_repo(&account, &repo)?.is_none() {
40 return Err(Error::NotFound("repository"));
41 }
42
43 let method = request.method().clone();
44 match (&method, tail) {
45 (&Method::GET, []) => list(&state, &account, &repo, request.uri().query()),
46 (&Method::POST, []) => trigger(request, &state, &identity, &account, &repo).await,
47 (&Method::GET, [id]) => {
48 let run = fetch(&state, &account, &repo, id)?;
49 Ok(response::json(StatusCode::OK, &run))
50 }
51 (&Method::PATCH, [id]) => cancel(request, &state, &identity, &account, &repo, id).await,
52 (&Method::GET, [id, "logs"]) => {
53 let run = fetch(&state, &account, &repo, id)?;
54 let log = state.runs.read_log(&account, &repo, &run.id)?;
55 Ok(response::log(log, run.status.terminal()))
56 }
57 (_, []) | (_, [_]) | (_, [_, "logs"]) => Err(Error::MethodNotAllowed),
58 _ => Err(Error::NotFound("route")),
59 }
60}
61
62fn fetch(state: &AppState, account: &Name, repo: &Name, id: &str) -> Result<Run> {
63 state
64 .runs
65 .get(account, repo, id)?
66 .ok_or(Error::NotFound("run"))
67}
68
69fn list(
70 state: &AppState,
71 account: &Name,
72 repo: &Name,
73 query: Option<&str>,
74) -> Result<Response<Body>> {
75 let wanted = query.and_then(|q| request::query_param(q, "status"));
76 let mut runs = state.runs.list(account, repo)?;
77
78 if let Some(wanted) = wanted {
79 let status: Status = serde_json::from_value(serde_json::Value::String(wanted.clone()))
80 .map_err(|_| Error::invalid("invalid-query", format!("unknown status {wanted:?}")))?;
81 runs.retain(|r| r.status == status);
82 }
83 let truncated = runs.len() > state.config.max_page_limit;
84 runs.truncate(state.config.max_page_limit);
85
86 Ok(response::json(
87 StatusCode::OK,
88 &crate::api::repos::page(runs, truncated),
89 ))
90}
91
92/// Queue a run by hand, for a revision that is already pushed.
93async fn trigger(
94 request: Request<Incoming>,
95 state: &AppState,
96 identity: &Identity,
97 account: &Name,
98 repo: &Name,
99) -> Result<Response<Body>> {
100 identity.require_write(account, repo)?;
101 if !state.config.ci_enabled {
102 return Err(Error::conflict(
103 "ci-disabled",
104 "CI is not enabled on this server",
105 ));
106 }
107
108 #[derive(Deserialize)]
109 struct Trigger {
110 #[serde(rename = "ref")]
111 git_ref: Option<String>,
112 }
113
114 let limit = state.config.max_body_bytes;
115 let body: Trigger = request::read_json(request.into_body(), limit)
116 .await
117 .unwrap_or(Trigger { git_ref: None });
118
119 let git_ref = match body.git_ref {
120 Some(value) => {
121 validate::ref_name(&value)?;
122 value
123 }
124 None => state
125 .meta
126 .get_ready_repo(account, repo)?
127 .map(|r| r.default_branch)
128 .unwrap_or_else(|| "refs/heads/main".into()),
129 };
130
131 let path = state.git.repo_path(account, repo)?;
132 let listed =
133 crate::git::exec::blocking(move || crate::git::discover::refs(&path, None)).await?;
134 let sha = listed["items"]
135 .as_array()
136 .and_then(|refs| {
137 refs.iter()
138 .find(|r| r["name"] == git_ref.as_str())
139 .and_then(|r| r["sha"].as_str())
140 .map(String::from)
141 })
142 .ok_or_else(|| Error::conflict("no-such-ref", format!("{git_ref} does not exist")))?;
143
144 let run = Run::queued(account.as_str(), repo.as_str(), &git_ref, &sha);
145 state.runs.put(account, repo, &run)?;
146
147 let location = format!("/v1/repos/{account}/{repo}/runs/{}", run.id);
148 Ok(response::created(&location, &run))
149}
150
151async fn cancel(
152 request: Request<Incoming>,
153 state: &AppState,
154 identity: &Identity,
155 account: &Name,
156 repo: &Name,
157 id: &str,
158) -> Result<Response<Body>> {
159 identity.require_write(account, repo)?;
160
161 let limit = state.config.max_body_bytes;
162 let body: PatchRun = request::read_json(request.into_body(), limit).await?;
163 if body.status != Status::Cancelled {
164 return Err(Error::invalid(
165 "invalid-run",
166 "the only supported transition is to \"cancelled\"",
167 ));
168 }
169
170 let mut run = fetch(state, account, repo, id)?;
171 if run.status.terminal() {
172 return Err(Error::conflict(
173 "run-finished",
174 format!("this run already {:?}", run.status),
175 ));
176 }
177
178 run.status = Status::Cancelled;
179 run.finished_at = Some(crate::account::token::now_secs());
180 run.detail = Some("cancelled".into());
181 state.runs.put(account, repo, &run)?;
182
183 Ok(response::json(StatusCode::OK, &run))
184}