forge
ExploreLog in

Tahinli/forge

public
0 watching
★ 0
34.9 KiB
1//! Pages: CI runs (`/{owner}/{repo}/ci` and `/{owner}/{repo}/ci/runs/{number}`). Anyone who can read
2//! the repo sees its runs, their jobs and the config error of a run. Starting a manual run, cancelling,
3//! re-running and approving need Write, and the buttons are shown only to writers. Anyone else gets a
4//! 404 for a private repo and a refusal for a change. Every change notifies the job waker and syncs
5//! the commit status of each job it changed. A run or job page that is not finished stays live: it
6//! opens a `changed` event stream (`.../events`) and re-renders itself on each event, with no reload
7//! and no timer.
8
9use std::collections::HashMap;
10use std::future::Future;
11use std::time::Duration;
12
13use forge_core::ci::{CiJob, CiRun, RunState};
14use forge_core::{Action, Denied, Store, User, authorize};
15use forge_git::browse::{RefInfo, RefKind};
16use futures_util::{Stream, stream};
17use serde::Deserialize;
18use time::OffsetDateTime;
19use time::format_description::well_known::Rfc3339;
20use tokio::sync::watch;
21use topcoat::Result;
22use topcoat::context::{Cx, app_context};
23use topcoat::router::content::Form;
24use topcoat::router::content::sse::{Event, KeepAlive, Sse};
25use topcoat::router::error::{SeeOther, forbidden, not_found, see_other};
26use topcoat::router::request::uri;
27use topcoat::router::response::{IntoResponse, Response};
28use topcoat::router::{HeaderValue, page, path_param, route};
29use topcoat::view::{View, component, view};
30
31use crate::auth::{current_user, require_csrf};
32use crate::i18n::{
33 self, Lang,
34 releases_ci::{self as g, K},
35};
36use crate::pages::refs::load_refs;
37use crate::pages::repo::repo_head;
38use crate::pages::repo_ctx::{RepoCtx, RepoHead, load_repo};
39use crate::state::AppState;
40use crate::views::{alert, enc, now_unix, relative_time, shell, short_oid};
41
42path_param!(owner);
43path_param!(repo);
44path_param!(run_number);
45
46/// Runs per list page.
47const PER_PAGE: u32 = 25;
48/// Highest accepted `?page=` value. Anything else falls back to page 1.
49const MAX_PAGE: u32 = 1000;
50
51/// Notice words (the only text a `?notice=` query can show), their text and whether they are errors.
52const NOTICES: [(&str, K, bool); 8] = [
53 ("started", K::CiNoticeStarted, false),
54 ("cancelled", K::CiNoticeCancelled, false),
55 ("approved", K::CiNoticeApproved, false),
56 ("rerun", K::CiNoticeRerun, false),
57 ("finished", K::CiNoticeFinished, true),
58 ("not-waiting", K::CiNoticeNotWaiting, true),
59 ("no-branch", K::CiNoticeNoBranch, true),
60 ("no-manual", K::CiNoticeNoManual, true),
61];
62
63#[derive(Deserialize)]
64struct CsrfForm {
65 #[serde(default)]
66 csrf: String,
67}
68
69#[derive(Deserialize)]
70struct ManualForm {
71 #[serde(default)]
72 csrf: String,
73 #[serde(default)]
74 branch: String,
75}
76
77/// The viewer, if signed in.
78fn viewer_user(ctx: &RepoCtx) -> Option<&User> {
79 ctx.viewer.as_ref().map(|(user, _)| user)
80}
81
82/// Whether the viewer may change CI runs of this repo. Archived repos refuse.
83fn can_write(ctx: &RepoCtx) -> bool {
84 authorize(
85 viewer_user(ctx),
86 &ctx.repo,
87 ctx.role,
88 ctx.org_role,
89 Action::Write,
90 )
91 .is_ok()
92}
93
94/// Write access, or the refusal that matches why it is missing.
95fn require_write(ctx: &RepoCtx) -> Result<()> {
96 match authorize(
97 viewer_user(ctx),
98 &ctx.repo,
99 ctx.role,
100 ctx.org_role,
101 Action::Write,
102 ) {
103 Ok(()) => Ok(()),
104 Err(Denied::NotFound) => Err(not_found().into()),
105 Err(Denied::Forbidden) => Err(forbidden().into()),
106 }
107}
108
109fn title_of(ctx: &RepoCtx) -> String {
110 format!("{}/{}", ctx.repo.owner_name, ctx.repo.name)
111}
112
113fn ci_href(title: &str) -> String {
114 format!("/{title}/ci")
115}
116
117/// The page of run `number`.
118pub fn run_href(title: &str, number: i64) -> String {
119 format!("/{title}/ci/runs/{number}")
120}
121
122/// The page of one job.
123pub fn job_href(title: &str, job_id: &str) -> String {
124 format!("/{title}/ci/jobs/{}", enc(job_id))
125}
126
127/// The badge image (same origin, default branch) and the README snippet that links it to the CI
128/// runs page. The snippet uses the configured base URL so it works when pasted elsewhere.
129fn badge_info(state: &AppState, title: &str, default_branch: &str) -> BadgeInfo {
130 let base = state.config.base_url.trim_end_matches('/');
131 let image = format!("/{}/badge.svg?branch={}", enc(title), enc(default_branch));
132 let snippet = format!("[![CI]({base}{image})]({base}{})", ci_href(&enc(title)));
133 BadgeInfo { image, snippet }
134}
135
136/// The notice from `?notice=<word>`. Anything else is ignored.
137fn notice_from_query(cx: &Cx, lang: Lang) -> Option<(&'static str, bool)> {
138 let query = uri(cx).query()?;
139 let word = form_urlencoded::parse(query.as_bytes())
140 .find(|(key, _)| key == "notice")
141 .map(|(_, value)| value.into_owned())?;
142 NOTICES
143 .iter()
144 .find(|(name, _, _)| *name == word)
145 .map(|(_, key, is_error)| (g::t(lang, *key), *is_error))
146}
147
148fn notice_redirect(href: &str, word: &str) -> SeeOther {
149 see_other(format!("{href}?notice={word}"))
150}
151
152/// The page number from `?page=`. Anything unreadable or out of range is page 1.
153fn page_from_query(cx: &Cx) -> u32 {
154 uri(cx)
155 .query()
156 .and_then(|query| {
157 form_urlencoded::parse(query.as_bytes())
158 .find(|(key, _)| key == "page")
159 .and_then(|(_, value)| value.parse::<u32>().ok())
160 })
161 .filter(|page| (1..=MAX_PAGE).contains(page))
162 .unwrap_or(1)
163}
164
165/// Unix seconds of an RFC 3339 stamp, if it parses.
166fn unix_of(stamp: &str) -> Option<i64> {
167 OffsetDateTime::parse(stamp, &Rfc3339)
168 .ok()
169 .map(|at| at.unix_timestamp())
170}
171
172/// "3 minutes ago" for a stamp. An unreadable stamp is shown as it is stored.
173fn age_text(lang: Lang, stamp: &str) -> String {
174 match unix_of(stamp) {
175 Some(unix) => relative_time(lang, unix, now_unix()),
176 None => stamp.to_owned(),
177 }
178}
179
180/// "1m 5s", or "" when the job never started. A running job counts up to now.
181fn duration_text(lang: Lang, started: Option<&str>, finished: Option<&str>) -> String {
182 let Some(start) = started.and_then(unix_of) else {
183 return String::new();
184 };
185 let end = finished.and_then(unix_of).unwrap_or_else(now_unix);
186 let secs = (end - start).max(0);
187 if secs >= 60 {
188 g::duration_minutes(lang, secs / 60, secs % 60)
189 } else {
190 g::duration_seconds(lang, secs)
191 }
192}
193
194/// Words for a state name, as the pages show it: `waiting_approval` reads "waiting approval".
195/// A state this build does not know is shown with its underscores as spaces.
196pub fn state_text(lang: Lang, state: &str) -> String {
197 let key = match state {
198 "waiting_approval" => K::StateWaitingApproval,
199 "queued" => K::StateQueued,
200 "running" => K::StateRunning,
201 "success" => K::StateSuccess,
202 "failure" => K::StateFailure,
203 "cancelled" => K::StateCancelled,
204 "error" => K::StateError,
205 "blocked" => K::StateBlocked,
206 "skipped" => K::StateSkipped,
207 other => return other.replace('_', " "),
208 };
209 g::t(lang, key).to_owned()
210}
211
212/// Whether a run has finished. Waiting for approval is not finished.
213pub fn run_finished(state: &RunState) -> bool {
214 matches!(
215 state,
216 RunState::Success | RunState::Failure | RunState::Cancelled | RunState::Error
217 )
218}
219
220/// Branch names of the repo, the default branch first.
221fn branch_choices(refs: &[RefInfo], default_branch: &str) -> Vec<String> {
222 let mut names: Vec<String> = refs
223 .iter()
224 .filter(|r| r.kind == RefKind::Branch)
225 .map(|r| r.name.clone())
226 .collect();
227 names.sort();
228 names.dedup();
229 if let Some(pos) = names.iter().position(|name| name == default_branch) {
230 let default = names.remove(pos);
231 names.insert(0, default);
232 }
233 names
234}
235
236/// The run with `number` in this repo, or 404.
237async fn run_of(state: &AppState, ctx: &RepoCtx, number: &str) -> Result<CiRun> {
238 let Ok(number) = number.parse::<i64>() else {
239 return Err(not_found().into());
240 };
241 match state.store.run_by_number(&ctx.repo.id, number).await? {
242 Some(run) => Ok(run),
243 None => Err(not_found().into()),
244 }
245}
246
247/// Wakes the job queue and syncs the commit status of every job whose state changed since `before`.
248/// A job missing from `before` (a fresh run) counts as changed.
249async fn after_change(state: &AppState, run_id: &str, before: &[CiJob]) -> Result<()> {
250 state.ci_waker.notify_waiters();
251 let Some(run) = state.store.run_by_id(run_id).await? else {
252 return Ok(());
253 };
254 state.ci_live.bump_run(run_id);
255 for job in state.store.jobs_for_run(run_id).await? {
256 let changed = before
257 .iter()
258 .find(|old| old.id == job.id)
259 .is_none_or(|old| old.state != job.state);
260 if changed {
261 if job.state.is_terminal() {
262 state.ci_live.close_job(&job.id);
263 } else {
264 state.ci_live.bump_job(&job.id);
265 }
266 crate::ci::sync_status(state, &run, &job).await;
267 }
268 }
269 Ok(())
270}
271
272/// One run in the list, with its values already shown as text.
273struct RunRow {
274 href: String,
275 number: i64,
276 state: String,
277 state_class: String,
278 trigger: String,
279 git_ref: String,
280 commit: String,
281 actor: String,
282 when: String,
283}
284
285/// The run-list form that starts a manual run.
286struct ManualRun {
287 action: String,
288 csrf: String,
289 branches: Vec<String>,
290}
291
292/// The status badge section: the image path (same origin) and the Markdown for a README.
293struct BadgeInfo {
294 image: String,
295 snippet: String,
296}
297
298/// A job row on the run page.
299struct JobRow {
300 href: String,
301 name: String,
302 state: String,
303 state_class: String,
304 duration: String,
305 runner: String,
306 message: String,
307}
308
309/// The run page's header values.
310struct RunView {
311 number: i64,
312 state: String,
313 state_class: String,
314 trigger: String,
315 git_ref: String,
316 commit: String,
317 actor: String,
318 when: String,
319 config_error: Option<String>,
320 action_base: String,
321 can_approve: bool,
322 can_cancel: bool,
323}
324
325/// Maps each user id in `ids` to a display name. A user that no longer exists shows as "deleted user".
326async fn usernames(
327 state: &AppState,
328 lang: Lang,
329 ids: impl Iterator<Item = &str>,
330) -> Result<HashMap<String, String>> {
331 let mut names = HashMap::new();
332 for id in ids {
333 if names.contains_key(id) {
334 continue;
335 }
336 let name = match state.store.user_by_id(id).await? {
337 Some(user) => user.username,
338 None => g::t(lang, K::DeletedUser).to_owned(),
339 };
340 names.insert(id.to_owned(), name);
341 }
342 Ok(names)
343}
344
345/// A small form that POSTs the CSRF token to `action`, with one button.
346#[component]
347async fn action_button(action: String, csrf: String, label: &'static str) -> Result<impl View> {
348 Ok(view! {
349 <form method="post" action=(action.as_str()) class="inline-form">
350 <input type="hidden" name="csrf" value=(csrf.as_str())>
351 <button type="submit">(label)</button>
352 </form>
353 })
354}
355
356/// The run list body: the manual run form for writers, the runs, and the pager.
357#[component]
358async fn runs_body(
359 head: RepoHead,
360 lang: Lang,
361 notice: Option<(&'static str, bool)>,
362 manual: Option<ManualRun>,
363 badge: BadgeInfo,
364 rows: Vec<RunRow>,
365 newer_href: Option<String>,
366 older_href: Option<String>,
367 page_label: String,
368) -> Result<impl View> {
369 Ok(view! {
370 repo_head(head: head)
371 if let Some((text, is_error)) = notice {
372 if is_error {
373 alert(message: text)
374 } else {
375 <section class="card notice" role="status">(text)</section>
376 }
377 }
378 if let Some(form) = manual {
379 <section class="card">
380 <h2>(g::t(lang, K::RunWorkflow))</h2>
381 <p class="muted small">(g::t(lang, K::RunWorkflowHelp))</p>
382 <form method="post" action=(form.action.as_str()) class="ci-manual-form">
383 <input type="hidden" name="csrf" value=(form.csrf.as_str())>
384 <label>
385 (g::t(lang, K::Branch)) " "
386 <select name="branch">
387 for branch in form.branches.iter() {
388 <option value=(branch.as_str())>(branch.as_str())</option>
389 }
390 </select>
391 </label>
392 <button type="submit">(g::t(lang, K::RunWorkflow))</button>
393 </form>
394 </section>
395 }
396 <section class="card">
397 <details>
398 <summary>(g::t(lang, K::BadgeHeading))</summary>
399 <p class="muted small">(g::t(lang, K::BadgeHelp))</p>
400 <p><img src=(badge.image.as_str()) alt="CI"></p>
401 <label class="small">
402 (g::t(lang, K::BadgeSnippet))
403 <textarea readonly="readonly" rows="2" class="ci-badge-snippet">(badge.snippet.as_str())</textarea>
404 </label>
405 </details>
406 </section>
407 <section class="card">
408 <h2>(g::t(lang, K::Runs))</h2>
409 if rows.is_empty() {
410 <p class="muted">(g::t(lang, K::NoRunsYet))</p>
411 } else {
412 <div class="table-wrap">
413 <table class="table ci-runs">
414 <thead>
415 <tr>
416 <th>(g::t(lang, K::ColRun))</th>
417 <th>(g::t(lang, K::ColState))</th>
418 <th>(g::t(lang, K::ColTrigger))</th>
419 <th>(g::t(lang, K::ColRef))</th>
420 <th>(g::t(lang, K::ColCommit))</th>
421 <th>(g::t(lang, K::ColActor))</th>
422 <th>(g::t(lang, K::ColAge))</th>
423 </tr>
424 </thead>
425 <tbody>
426 for row in rows.iter() {
427 <tr>
428 <td><a href=(row.href.as_str())>"#" (row.number)</a></td>
429 <td><span class=(row.state_class.as_str())>(row.state.as_str())</span></td>
430 <td>(row.trigger.as_str())</td>
431 <td><code>(row.git_ref.as_str())</code></td>
432 <td><code>(row.commit.as_str())</code></td>
433 <td>(row.actor.as_str())</td>
434 <td>(row.when.as_str())</td>
435 </tr>
436 }
437 </tbody>
438 </table>
439 </div>
440 }
441 <nav class="pager">
442 if let Some(href) = newer_href {
443 <a href=(href.as_str())>(g::t(lang, K::Newer))</a>
444 }
445 <span class="muted small">(page_label.as_str())</span>
446 if let Some(href) = older_href {
447 <a href=(href.as_str())>(g::t(lang, K::Older))</a>
448 }
449 </nav>
450 </section>
451 })
452}
453
454/// Lists the repo's runs, newest first, 25 to a page. The manual run form is for writers only.
455#[page("/{owner}/{repo}/ci")]
456async fn runs_list_page(cx: &Cx) -> Result<impl View> {
457 let owner = path_param::<Owner>(cx);
458 let name = path_param::<Repo>(cx);
459 let ctx = load_repo(cx, owner, name).await?;
460 let lang = i18n::lang(cx, viewer_user(&ctx));
461 let state = app_context::<AppState>(cx);
462 let page = page_from_query(cx);
463 let (runs, total) = state.store.list_runs(&ctx.repo.id, page, PER_PAGE).await?;
464 let names = usernames(state, lang, runs.iter().map(|run| run.actor_id.as_str())).await?;
465 let rows: Vec<RunRow> = runs
466 .iter()
467 .map(|run| RunRow {
468 href: run_href(&title_of(&ctx), run.number),
469 number: run.number,
470 state: state_text(lang, run.state.as_str()),
471 state_class: format!("ci-state ci-{}", run.state.as_str()),
472 trigger: run.trigger.clone(),
473 git_ref: run.git_ref.clone(),
474 commit: short_oid(&run.commit_oid).to_owned(),
475 actor: names.get(&run.actor_id).cloned().unwrap_or_default(),
476 when: age_text(lang, &run.created_at),
477 })
478 .collect();
479 let last_page = total.div_ceil(u64::from(PER_PAGE)).max(1);
480 let base = ci_href(&title_of(&ctx));
481 let newer_href = (page > 1).then(|| format!("{base}?page={}", page - 1));
482 let older_href = (u64::from(page) < last_page).then(|| format!("{base}?page={}", page + 1));
483 let page_label = g::page_label(lang, page);
484 let manual = if can_write(&ctx) {
485 let csrf = match current_user(cx).await? {
486 Some((_, session)) => session.csrf_token.clone(),
487 None => String::new(),
488 };
489 let refs = load_refs(ctx.path.clone()).await?;
490 Some(ManualRun {
491 action: base.clone(),
492 csrf,
493 branches: branch_choices(&refs, &ctx.repo.default_branch),
494 })
495 } else {
496 None
497 };
498 let notice = notice_from_query(cx, lang);
499 let badge = badge_info(state, &title_of(&ctx), &ctx.repo.default_branch);
500 let head = ctx.head(state, "ci").await?;
501 let title = format!("{} \u{b7} CI", title_of(&ctx));
502 Ok(view! {
503 shell(title: title.as_str(), viewer: ctx.viewer.as_ref(),
504 runs_body(head: head, lang: lang, notice: notice, manual: manual, badge: badge, rows: rows,
505 newer_href: newer_href, older_href: older_href, page_label: page_label))
506 })
507}
508
509/// Starts a manual run on a branch. Writers only. The branch must be one of the repo's branches.
510#[route(POST "/{owner}/{repo}/ci")]
511async fn manual_run_post(cx: &Cx, Form(form): Form<ManualForm>) -> Result<SeeOther> {
512 let owner = path_param::<Owner>(cx);
513 let name = path_param::<Repo>(cx);
514 let ctx = load_repo(cx, owner, name).await?;
515 let (user, _) = require_csrf(cx, &form.csrf).await?;
516 require_write(&ctx)?;
517 let state = app_context::<AppState>(cx);
518 let title = title_of(&ctx);
519 let branch = form.branch.trim();
520 let refs = load_refs(ctx.path.clone()).await?;
521 if !branch_choices(&refs, &ctx.repo.default_branch)
522 .iter()
523 .any(|choice| choice == branch)
524 {
525 return Ok(notice_redirect(&ci_href(&title), "no-branch"));
526 }
527 let run = match crate::ci::manual_run(state, &ctx.repo, branch, &user.id).await {
528 Ok(run) => run,
529 Err(forge_core::Error::Invalid(_)) => {
530 return Ok(notice_redirect(&ci_href(&title), "no-manual"));
531 }
532 Err(e) => return Err(e.into()),
533 };
534 after_change(state, &run.id, &[]).await?;
535 Ok(see_other(run_href(&title, run.number)))
536}
537
538/// The header values of a run for its page.
539async fn run_view(
540 state: &AppState,
541 ctx: &RepoCtx,
542 run: &CiRun,
543 writer: bool,
544 lang: Lang,
545) -> Result<RunView> {
546 let names = usernames(state, lang, std::iter::once(run.actor_id.as_str())).await?;
547 let title = title_of(ctx);
548 let action_base = run_href(&title, run.number);
549 Ok(RunView {
550 number: run.number,
551 state: state_text(lang, run.state.as_str()),
552 state_class: format!("ci-state ci-{}", run.state.as_str()),
553 trigger: run.trigger.clone(),
554 git_ref: run.git_ref.clone(),
555 commit: run.commit_oid.clone(),
556 actor: names.get(&run.actor_id).cloned().unwrap_or_default(),
557 when: age_text(lang, &run.created_at),
558 config_error: run.config_error.clone(),
559 can_approve: writer && run.state == RunState::WaitingApproval,
560 can_cancel: writer && !run_finished(&run.state),
561 action_base,
562 })
563}
564
565/// The job rows of a run page. Runner names come from the runner list.
566fn job_rows(
567 title: &str,
568 jobs: &[CiJob],
569 runners: &HashMap<String, String>,
570 lang: Lang,
571) -> Vec<JobRow> {
572 jobs.iter()
573 .map(|job| JobRow {
574 href: job_href(title, &job.id),
575 name: job.name.clone(),
576 state: state_text(lang, job.state.as_str()),
577 state_class: format!("ci-state ci-{}", job.state.as_str()),
578 duration: duration_text(lang, job.started_at.as_deref(), job.finished_at.as_deref()),
579 runner: job
580 .runner_id
581 .as_ref()
582 .and_then(|id| runners.get(id).cloned())
583 .unwrap_or_default(),
584 message: job.message.clone().unwrap_or_default(),
585 })
586 .collect()
587}
588
589/// How often an idle event stream sends a `: keepalive` comment, so a reverse proxy does not drop
590/// the connection. This is not a poll: events follow the change signals.
591const STREAM_KEEP_ALIVE: Duration = Duration::from_secs(30);
592
593/// A run's or job's status word and whether it is terminal. `None` when the row is gone.
594pub(crate) type Status = Option<(String, bool)>;
595
596pub(crate) async fn run_status(store: &Store, run_id: &str) -> Result<Status> {
597 Ok(store
598 .run_by_id(run_id)
599 .await?
600 .map(|run| (run.state.as_str().to_owned(), run_finished(&run.state))))
601}
602
603pub(crate) async fn job_status(store: &Store, job_id: &str) -> Result<Status> {
604 Ok(store
605 .job_by_id(job_id)
606 .await?
607 .map(|job| (job.state.as_str().to_owned(), job.state.is_terminal())))
608}
609
610/// One `changed` event carrying the status word.
611pub(crate) fn changed_event(status: String) -> Result<Event> {
612 Ok(Event::new().event("changed").data(status))
613}
614
615/// The `changed` events of a CI page. The first event carries the status now. Each later one
616/// follows a wake of `wake`: with `only_on_status`, only when the status differs from the last one
617/// sent. The stream ends after the terminal status is sent, when the row is gone, or when the signal
618/// closes. Subscribe `wake` before the first read, so a change in between is not lost.
619pub(crate) fn changed_stream<F, Fut>(
620 wake: watch::Receiver<u64>,
621 only_on_status: bool,
622 read: F,
623) -> impl Stream<Item = Result<Event>> + Send + 'static
624where
625 F: Fn() -> Fut + Send + 'static,
626 Fut: Future<Output = Result<Status>> + Send,
627{
628 struct Follow<F> {
629 wake: watch::Receiver<u64>,
630 only_on_status: bool,
631 read: F,
632 /// The status last read, sent or not.
633 last: Option<String>,
634 /// Set once a terminal status is read. Nothing follows it.
635 done: bool,
636 /// False until the first read. Every later read waits for a wake first.
637 primed: bool,
638 }
639 stream::unfold(
640 Follow {
641 wake,
642 only_on_status,
643 read,
644 last: None,
645 done: false,
646 primed: false,
647 },
648 |mut follow| async move {
649 loop {
650 if follow.done {
651 return None;
652 }
653 if follow.primed && follow.wake.changed().await.is_err() {
654 return None;
655 }
656 follow.primed = true;
657 let (status, terminal) = match (follow.read)().await {
658 Ok(Some(found)) => found,
659 Ok(None) => return None,
660 Err(e) => {
661 tracing::error!(error = %e, "ci: live stream could not read the status");
662 return None;
663 }
664 };
665 follow.done = terminal;
666 let changed = follow.last.as_deref() != Some(status.as_str());
667 let send = changed || !follow.only_on_status;
668 follow.last = Some(status.clone());
669 if send {
670 return Some((changed_event(status), follow));
671 }
672 }
673 },
674 )
675}
676
677/// The response of an event stream for a CI page: keep-alive comments while idle, and no buffering
678/// by a reverse proxy.
679pub(crate) fn events_response(
680 cx: &Cx,
681 events: impl Stream<Item = Result<Event>> + Send + 'static,
682) -> Result<Response> {
683 let events = Sse::new(events).keep_alive(
684 KeepAlive::new()
685 .interval(STREAM_KEEP_ALIVE)
686 .text("keepalive"),
687 );
688 let mut res = events.into_response(cx)?;
689 res.headers_mut()
690 .insert("x-accel-buffering", HeaderValue::from_static("no"));
691 Ok(res)
692}
693
694/// The run page body: the header, the buttons for writers, the config error, and the jobs. A run
695/// that is not finished carries `live`, the URL of its event stream.
696#[component]
697async fn run_body(
698 head: RepoHead,
699 lang: Lang,
700 notice: Option<(&'static str, bool)>,
701 run: RunView,
702 jobs: Vec<JobRow>,
703 writer_csrf: Option<String>,
704 live: Option<String>,
705) -> Result<impl View> {
706 Ok(view! {
707 repo_head(head: head)
708 <div data-live=(live.as_deref())>
709 if let Some((text, is_error)) = notice {
710 if is_error {
711 alert(message: text)
712 } else {
713 <section class="card notice" role="status">(text)</section>
714 }
715 }
716 <section class="card">
717 <h2>(g::t(lang, K::RunNumber)) (run.number)</h2>
718 <p class="muted small">
719 <span class=(run.state_class.as_str())>(run.state.as_str())</span>
720 " \u{b7} " (run.trigger.as_str())
721 " \u{b7} " <code>(run.git_ref.as_str())</code>
722 " \u{b7} " <code>(short_oid(&run.commit).to_owned())</code>
723 if !run.actor.is_empty() {
724 " \u{b7} " (run.actor.as_str())
725 }
726 " \u{b7} " (run.when.as_str())
727 </p>
728 if let Some(message) = run.config_error.as_ref() {
729 <pre class="ci-config-error">(message.as_str())</pre>
730 }
731 if let Some(csrf) = writer_csrf {
732 <div class="ci-actions">
733 if run.can_approve {
734 action_button(action: format!("{}/approve", run.action_base), csrf: csrf.clone(), label: g::t(lang, K::ApproveAndRun))
735 }
736 if run.can_cancel {
737 action_button(action: format!("{}/cancel", run.action_base), csrf: csrf.clone(), label: g::t(lang, K::Cancel))
738 }
739 action_button(action: format!("{}/rerun", run.action_base), csrf: csrf, label: g::t(lang, K::Rerun))
740 </div>
741 }
742 </section>
743 <section class="card">
744 <h2>(g::t(lang, K::Jobs))</h2>
745 if jobs.is_empty() {
746 <p class="muted">(g::t(lang, K::NoJobs))</p>
747 } else {
748 <div class="table-wrap">
749 <table class="table ci-jobs">
750 <thead>
751 <tr>
752 <th>(g::t(lang, K::ColJob))</th>
753 <th>(g::t(lang, K::ColState))</th>
754 <th>(g::t(lang, K::ColDuration))</th>
755 <th>(g::t(lang, K::ColRunner))</th>
756 <th>(g::t(lang, K::ColMessage))</th>
757 </tr>
758 </thead>
759 <tbody>
760 for job in jobs.iter() {
761 <tr>
762 <td><a href=(job.href.as_str())>(job.name.as_str())</a></td>
763 <td><span class=(job.state_class.as_str())>(job.state.as_str())</span></td>
764 <td>(job.duration.as_str())</td>
765 <td>(job.runner.as_str())</td>
766 <td>(job.message.as_str())</td>
767 </tr>
768 }
769 </tbody>
770 </table>
771 </div>
772 }
773 </section>
774 </div>
775 })
776}
777
778/// One run: its state, the buttons for writers, the config error and its jobs.
779#[page("/{owner}/{repo}/ci/runs/{run_number}")]
780async fn run_page(cx: &Cx) -> Result<impl View> {
781 let owner = path_param::<Owner>(cx);
782 let name = path_param::<Repo>(cx);
783 let number = path_param::<RunNumber>(cx).to_owned();
784 let ctx = load_repo(cx, owner, name).await?;
785 let lang = i18n::lang(cx, viewer_user(&ctx));
786 let state = app_context::<AppState>(cx);
787 let run = run_of(state, &ctx, &number).await?;
788 let writer = can_write(&ctx);
789 let writer_csrf = if writer {
790 match current_user(cx).await? {
791 Some((_, session)) => Some(session.csrf_token.clone()),
792 None => None,
793 }
794 } else {
795 None
796 };
797 let jobs = state.store.jobs_for_run(&run.id).await?;
798 let runners: HashMap<String, String> = state
799 .store
800 .list_runners()
801 .await?
802 .into_iter()
803 .map(|runner| (runner.id, runner.name))
804 .collect();
805 let title = title_of(&ctx);
806 let rows = job_rows(&title, &jobs, &runners, lang);
807 let live =
808 (!run_finished(&run.state)).then(|| format!("{}/events", run_href(&title, run.number)));
809 let notice = notice_from_query(cx, lang);
810 let header = run_view(state, &ctx, &run, writer, lang).await?;
811 let head = ctx.head(state, "ci").await?;
812 let page_title = format!("{title} \u{b7} CI #{}", run.number);
813 Ok(view! {
814 shell(title: page_title.as_str(), viewer: ctx.viewer.as_ref(),
815 run_body(head: head, lang: lang, notice: notice, run: header, jobs: rows,
816 writer_csrf: writer_csrf, live: live))
817 })
818}
819
820/// The live `changed` stream of a run that is not finished. It is sent on every change to the run
821/// or to one of its jobs, so the page can render its rows again.
822#[route(GET "/{owner}/{repo}/ci/runs/{run_number}/events")]
823async fn run_events(cx: &Cx) -> Result<Response> {
824 let owner = path_param::<Owner>(cx);
825 let name = path_param::<Repo>(cx);
826 let number = path_param::<RunNumber>(cx).to_owned();
827 let ctx = load_repo(cx, owner, name).await?;
828 let state = app_context::<AppState>(cx);
829 let run = run_of(state, &ctx, &number).await?;
830 // Subscribed before the status is read, so a change in between is sent.
831 let wake = state.ci_live.subscribe_run(&run.id);
832 let store = state.store.clone();
833 let run_id = run.id.clone();
834 let events = changed_stream(wake, false, move || {
835 let store = store.clone();
836 let run_id = run_id.clone();
837 async move { run_status(&store, &run_id).await }
838 });
839 events_response(cx, events)
840}
841
842/// Cancels a run that has not finished. Queued and blocked jobs are cancelled at once; running jobs
843/// get a cancel request that their runner sees on its next heartbeat.
844#[route(POST "/{owner}/{repo}/ci/runs/{run_number}/cancel")]
845async fn run_cancel(cx: &Cx, Form(form): Form<CsrfForm>) -> Result<SeeOther> {
846 let owner = path_param::<Owner>(cx);
847 let name = path_param::<Repo>(cx);
848 let number = path_param::<RunNumber>(cx).to_owned();
849 let ctx = load_repo(cx, owner, name).await?;
850 require_csrf(cx, &form.csrf).await?;
851 require_write(&ctx)?;
852 let state = app_context::<AppState>(cx);
853 let run = run_of(state, &ctx, &number).await?;
854 let href = run_href(&title_of(&ctx), run.number);
855 if run_finished(&run.state) {
856 return Ok(notice_redirect(&href, "finished"));
857 }
858 let before = state.store.jobs_for_run(&run.id).await?;
859 state.store.request_cancel(&run.id).await?;
860 after_change(state, &run.id, &before).await?;
861 Ok(notice_redirect(&href, "cancelled"))
862}
863
864/// Approves a run that waits for approval: its jobs are queued. Anything else is refused.
865#[route(POST "/{owner}/{repo}/ci/runs/{run_number}/approve")]
866async fn run_approve(cx: &Cx, Form(form): Form<CsrfForm>) -> Result<SeeOther> {
867 let owner = path_param::<Owner>(cx);
868 let name = path_param::<Repo>(cx);
869 let number = path_param::<RunNumber>(cx).to_owned();
870 let ctx = load_repo(cx, owner, name).await?;
871 require_csrf(cx, &form.csrf).await?;
872 require_write(&ctx)?;
873 let state = app_context::<AppState>(cx);
874 let run = run_of(state, &ctx, &number).await?;
875 let href = run_href(&title_of(&ctx), run.number);
876 if run.state != RunState::WaitingApproval {
877 return Ok(notice_redirect(&href, "not-waiting"));
878 }
879 let before = state.store.jobs_for_run(&run.id).await?;
880 state.store.approve_run(&run.id).await?;
881 after_change(state, &run.id, &before).await?;
882 Ok(notice_redirect(&href, "approved"))
883}
884
885/// Starts a new run with the same commit, ref, trigger and job specs, then opens it.
886#[route(POST "/{owner}/{repo}/ci/runs/{run_number}/rerun")]
887async fn run_rerun(cx: &Cx, Form(form): Form<CsrfForm>) -> Result<SeeOther> {
888 let owner = path_param::<Owner>(cx);
889 let name = path_param::<Repo>(cx);
890 let number = path_param::<RunNumber>(cx).to_owned();
891 let ctx = load_repo(cx, owner, name).await?;
892 let (user, _) = require_csrf(cx, &form.csrf).await?;
893 require_write(&ctx)?;
894 let state = app_context::<AppState>(cx);
895 let run = run_of(state, &ctx, &number).await?;
896 let new_run = state.store.rerun(&run.id, &user.id).await?;
897 after_change(state, &new_run.id, &[]).await?;
898 Ok(notice_redirect(
899 &run_href(&title_of(&ctx), new_run.number),
900 "rerun",
901 ))
902}
903
904#[cfg(test)]
905mod tests {
906 use super::*;
907
908 #[test]
909 fn durations_read_well() {
910 assert_eq!(
911 duration_text(
912 Lang::En,
913 Some("2026-10-01T10:00:00.000000000Z"),
914 Some("2026-10-01T10:01:05.000000000Z")
915 ),
916 "1m 5s"
917 );
918 assert_eq!(
919 duration_text(
920 Lang::En,
921 Some("2026-10-01T10:00:00.000000000Z"),
922 Some("2026-10-01T10:00:09.000000000Z")
923 ),
924 "9s"
925 );
926 assert_eq!(duration_text(Lang::En, None, None), "");
927 }
928
929 #[test]
930 fn state_names_read_as_words() {
931 assert_eq!(state_text(Lang::En, "waiting_approval"), "waiting approval");
932 assert_eq!(state_text(Lang::En, "success"), "success");
933 assert_eq!(state_text(Lang::Tr, "waiting_approval"), "onay bekliyor");
934 assert_eq!(state_text(Lang::Tr, "queued"), "sırada");
935 }
936
937 #[test]
938 fn finished_runs_are_the_terminal_ones() {
939 assert!(run_finished(&RunState::Success));
940 assert!(run_finished(&RunState::Cancelled));
941 assert!(!run_finished(&RunState::WaitingApproval));
942 assert!(!run_finished(&RunState::Running));
943 }
944
945 #[test]
946 fn default_branch_comes_first() {
947 let refs = vec![
948 RefInfo {
949 name: "dev".into(),
950 full_name: "refs/heads/dev".into(),
951 target: "a".into(),
952 kind: RefKind::Branch,
953 },
954 RefInfo {
955 name: "main".into(),
956 full_name: "refs/heads/main".into(),
957 target: "b".into(),
958 kind: RefKind::Branch,
959 },
960 RefInfo {
961 name: "v1".into(),
962 full_name: "refs/tags/v1".into(),
963 target: "c".into(),
964 kind: RefKind::Tag,
965 },
966 ];
967 assert_eq!(branch_choices(&refs, "main"), vec!["main", "dev"]);
968 }
969}