Runs by event: on decides the jobs; tag pushes run tag jobs c889911bec12

- **What starts a job: `on` (oxygenflow#24).** A pipeline's or a job's `on { push "main" "feature/*"; tag "v*"; manual }` decides which events run it: pushing a tag now runs the jobs that name it, on the tagged state; a run has only the jobs its event starts (and what they need), and an event that starts none starts no run. A rerun repeats its run's event. A bookmark sent again unchanged starts no run.

        trigger: r.trigger_label(),

            "schedule" => "started on schedule".to_string(),

            t if t.starts_with("tag:") => format!("tag {} pushed by {}", &t[4..], self.actor),

            t if t.starts_with("rerun:") => format!("re-run of #{} by {}", self.rerun_of().unwrap_or(""), self.actor),

    /// The run a rerun repeats: `N` of `rerun:N` or `rerun:N:EVENT`.

    pub fn rerun_of(&self) -> Option<&str> {

        let rest = self.trigger.strip_prefix("rerun:")?;

        Some(rest.split(':').next().unwrap_or(rest))

    }

    /// The trigger as people read it: `rerun:N`, without the event it repeats.

    pub fn trigger_label(&self) -> String {

        match self.rerun_of() {

            Some(n) => format!("rerun:{n}"),

            None => self.trigger.clone(),

        }

    }

    if heads.is_empty() || !enabled(app, row) {

    // A pushed tag runs the jobs whose `on` names it (oxygenflow#24).

    let started = match bookmark.strip_prefix("tags/") {

        Some(tag) => start_run_with(app, row, bookmark, state, heads, actor, &format!("tag:{tag}"), None).await,

        None => {

            let touched = touched_since_last_run(app, row, bookmark, heads).await;

            start_run_scoped(app, row, bookmark, state, heads, actor, "push", touched.as_deref()).await

        }

    };

    if let Err(e) = started {

    // A rerun repeats its run's event: the jobs it had are the jobs it gets.

    let trigger = format!("rerun:{}:{}", run.number, original_event(run));

    start_run(app, row, &run.bookmark, &run.state, &heads, actor, &trigger)

}

/// The event a run's trigger stands for: `push`, `tag:NAME`, `manual`,

/// `schedule`, and `rerun:N:EVENT` repeating EVENT.

fn event_of<'a>(trigger: &'a str, bookmark: &'a str) -> flow_core::pipeline::Event<'a> {

    use flow_core::pipeline::Event;

    let t = match trigger.strip_prefix("rerun:") {

        Some(rest) => rest.split_once(':').map(|(_, e)| e).unwrap_or("manual"),

        None => trigger,

    };

    if let Some(tag) = t.strip_prefix("tag:") {

        Event::Tag(tag)

    } else if t == "push" {

        Event::Push(bookmark)

    } else if t == "schedule" {

        Event::Schedule

    } else {

        Event::Manual

    }

/// The event to repeat when rerunning `run`.

fn original_event(run: &fdb::Run) -> String {

    let t = run.trigger.as_str();

    match t.strip_prefix("rerun:") {

        Some(rest) => rest.split_once(':').map(|(_, e)| e.to_string()).unwrap_or_else(|| "manual".into()),

        None => t.to_string(),

    }

}

            // The jobs this run's event starts (`on`), out of those its

            // push touched (`scope`); then what they depend on.

            let event = event_of(trigger, bookmark);

            let mut keep: std::collections::HashSet<String> = p

                .order()

                .into_iter()

                .filter(|j| p.on_of(j).matches(event) && in_scope(&j.scope, touched))

                .map(|j| j.name.clone())

                .collect();

            if jobs.is_empty() {

                // Nothing this event starts (or this push touched).

            // A bookmark sent again unchanged starts nothing.

            .filter(|u| u.old_state != u.new_state && report.results.iter().any(|r| r.ok && r.name == u.name))

    assert_eq!((third.trigger_label().as_str(), third.status.as_str(), third.state.as_str()), ("rerun:1", "queued", first[0].state.as_str()));

    assert_eq!(third.trigger, "rerun:1:push", "a rerun keeps the event it repeats");

/// oxygenflow#24: a job's `on` decides which events start it: a pushed

/// tag runs only the jobs that name it; a push to a bookmark the pipeline

/// does not name starts no run; a rerun repeats its run's event.

#[tokio::test(flavor = "multi_thread")]

async fn on_decides_which_events_start_a_job() {

    let h = hub().await;

    db::accounts::create_account(&h.app.db, "andy", "[email protected]", "x", "salt", true).await.unwrap();

    create(&h.app, "andy", "events").await;

    let url = format!("{}/andy/events", h.base);

    let work = tempfile::tempdir().unwrap();

    let c = client(&url, work.path());

    let text = format!(

        "on {{\n    push \"main\"\n    manual\n}}\n{}job \"dist\" {{\n    sandbox \"alpine\"\n    on {{\n        tag \"v*\"\n    }}\n    exec \"true\"\n}}\n",

        pipeline("")

    );

    // `on` must come after schema-version; pipeline() starts with it.

    let text = text.replacen(

        "on {\n    push \"main\"\n    manual\n}\nschema-version 1\n",

        "schema-version 1\non {\n    push \"main\"\n    manual\n}\n",

        1,

    );

    write(work.path(), ".flow/pipeline.kdl", &text);

    write(work.path(), "test.sh", "echo ok\n");

    c.commit("Start", &[]).unwrap();

    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));

    let jobs_of_latest = |app: Shared, n: usize| async move {

        let mut runs = Vec::new();

        for _ in 0..100 {

            runs = runs_of(&app, "andy", "events").await;

            if runs.len() >= n {

                break;

            }

            tokio::time::sleep(std::time::Duration::from_millis(50)).await;

        }

        assert_eq!(runs.len(), n, "runs");

        let mut j: Vec<String> = db::flow::jobs(&app.db, runs[0].id).await.unwrap().into_iter().map(|j| j.name).collect();

        j.sort();

        (runs[0].clone(), j)

    };

    let (_, j) = jobs_of_latest(h.app.clone(), 1).await;

    assert_eq!(j, ["build", "lint", "test"], "a push to main: no dist");

    // A tag: only dist.

    let heads = c.bookmark_heads().unwrap();

    c.tag_create("v1.0.0", &heads, "first").unwrap();

    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));

    let (run, j) = jobs_of_latest(h.app.clone(), 2).await;

    assert_eq!(j, ["dist"], "a tag runs only what names it");

    assert_eq!(run.trigger, "tag:v1.0.0");

    // A rerun repeats the tag's event.

    let row = db::repos::get(&h.app.db, "andy", "events").await.unwrap().unwrap();

    db::flow::finish(

        &h.app.db,

        db::flow::jobs(&h.app.db, run.id).await.unwrap()[0].id,

        db::flow::Outcome { status: "passed", exit_code: Some(0), error: "", receipt: "", receipt_hash: "" },

    )

    .await

    .unwrap();

    let run = db::flow::run_by_id(&h.app.db, run.id).await.unwrap().unwrap();

    oxygenhub::flow::rerun(&h.app, &row, &run, "andy").await.unwrap();

    let (_, j) = jobs_of_latest(h.app.clone(), 3).await;

    assert_eq!(j, ["dist"], "the rerun is the tag's run again");

    // A bookmark the pipeline does not name starts nothing.

    c.bookmark_create("develop", None).unwrap();

    assert!(c.push_bookmark("origin", "develop", false).unwrap().iter().all(|r| r.ok));

    tokio::time::sleep(std::time::Duration::from_millis(300)).await;

    assert_eq!(runs_of(&h.app, "andy", "events").await.len(), 3, "no run for develop");

}