Runners serve the site, an owner or one repository (refs #41, part 2) 74be97962ed8

-- Runner scopes (oxygenhub#41, part 2): a runner takes jobs from the whole

-- site (both empty), from one owner's repositories (scope_owner), or from

-- one repository (scope_repo_id, with its owner). Registration tokens carry

-- the scope the runner gets.

ALTER TABLE flow_runners ADD COLUMN scope_owner TEXT NOT NULL DEFAULT '';

ALTER TABLE flow_runners ADD COLUMN scope_repo_id INTEGER;

ALTER TABLE flow_runner_registrations ADD COLUMN scope_owner TEXT NOT NULL DEFAULT '';

ALTER TABLE flow_runner_registrations ADD COLUMN scope_repo_id INTEGER;

    claim_within(pool, runtimes, runner, token_hash, Room::default(), &Default::default()).await

/// The first queued job a runner with these tiers, this room and this

/// scope may take, now running under `runner`. A scoped runner never sees

/// another scope's jobs (oxygenhub#41).

pub async fn claim_within(

    pool: &Pool,

    runtimes: &[String],

    runner: &str,

    token_hash: &str,

    room: Room,

    scope: &super::runners::Scope,

) -> sqlx::Result<Option<Job>> {

        "SELECT j.* FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id JOIN repositories rep ON rep.id = r.repo_id \

         WHERE j.status = 'queued' AND r.status IN ('queued', 'running') \

         AND (? = '' OR rep.owner = ?) AND (? IS NULL OR r.repo_id = ?) \

         ORDER BY j.run_id, j.depth, j.id LIMIT 200",

    .bind(&scope.owner)

    .bind(&scope.owner)

    .bind(scope.repo_id)

    .bind(scope.repo_id)

    /// The owner whose repositories it serves; empty for the whole site.

    pub scope_owner: String,

    /// The one repository it serves, when it serves one.

    pub scope_repo_id: Option<i64>,

/// Which repositories' jobs a runner may take (oxygenhub#41).

#[derive(Clone, Debug, Default, PartialEq, Eq)]

pub struct Scope {

    /// Empty: any owner.

    pub owner: String,

    /// `None`: any of the owner's repositories.

    pub repo_id: Option<i64>,

}

impl Scope {

    pub fn is_site(&self) -> bool {

        self.owner.is_empty() && self.repo_id.is_none()

    }

}

    pub fn scope(&self) -> Scope {

        Scope { owner: self.scope_owner.clone(), repo_id: self.scope_repo_id }

    }

/// A one-time registration token, valid for a day, for a runner of `scope`.

pub async fn add_registration(pool: &Pool, token_hash: &str, by: &str, scope: &Scope) -> sqlx::Result<()> {

    sqlx::query(

        "INSERT INTO flow_runner_registrations (token_hash, created_by, created_ms, expires_ms, scope_owner, scope_repo_id) VALUES (?, ?, ?, ?, ?, ?)",

    )

    .bind(token_hash)

    .bind(by)

    .bind(now)

    .bind(now + 24 * 3_600_000)

    .bind(&scope.owner)

    .bind(scope.repo_id)

    .execute(pool)

    .await?;

    Ok(())

}

/// Change the repositories a runner serves.

pub async fn set_scope(pool: &Pool, id: i64, scope: &Scope) -> sqlx::Result<()> {

    sqlx::query("UPDATE flow_runners SET scope_owner = ?, scope_repo_id = ? WHERE id = ?")

        .bind(&scope.owner)

        .bind(scope.repo_id)

        .bind(id)

    let found: Option<(String, String, Option<i64>)> = sqlx::query_as(

        "SELECT created_by, scope_owner, scope_repo_id FROM flow_runner_registrations WHERE token_hash = ? AND used_ms IS NULL AND expires_ms > ?",

    )

    .bind(registration)

    .bind(now)

    .fetch_optional(&mut *tx)

    .await?;

    let Some((who, scope_owner, scope_repo_id)) = found else { return Ok(None) };

        "INSERT INTO flow_runners (name, token_hash, source, key, key_name, created_ms, created_by, scope_owner, scope_repo_id) VALUES (?, ?, 'registered', ?, ?, ?, ?, ?, ?) RETURNING id",

    .bind(scope_owner)

    .bind(scope_repo_id)

    let Some(job) = fdb::claim_within(&app.db, &req.runtimes, &runner, &crate::auth::token_hash(&secret), room, &me.scope()).await? else {

        .route("/admin/runners/{id}/scope", post(runners::set_scope))

    /// The repositories it serves, in words.

    pub scope: String,

/// A scope in words: the whole site, an owner's repositories, or one

/// repository.

async fn scope_words(app: &Shared, s: &runners::Scope) -> String {

    match s.repo_id {

        Some(id) => match db::repos::get_by_id(&app.db, id).await.ok().flatten() {

            Some(r) => format!("{}/{}", r.owner, r.name),

            None => "a removed repository".into(),

        },

        None if s.owner.is_empty() => "the whole site".into(),

        None => format!("{}'s repositories", s.owner),

    }

}

/// A scope as typed: empty for the whole site, an owner, or

/// `owner/repository`.

async fn parse_scope(app: &Shared, text: &str) -> Result<runners::Scope, String> {

    let text = text.trim().trim_matches('/');

    if text.is_empty() {

        return Ok(runners::Scope::default());

    }

    if let Some((owner, name)) = text.split_once('/') {

        return match db::repos::get(&app.db, owner, name).await {

            Ok(Some(r)) => Ok(runners::Scope { owner: r.owner.clone(), repo_id: Some(r.id) }),

            _ => Err(format!("There is no repository {text}.")),

        };

    }

    let user = accounts::account_by_name(&app.db, text).await.ok().flatten().is_some();

    let org = accounts::org_by_name(&app.db, text).await.ok().flatten().is_some();

    if user || org {

        Ok(runners::Scope { owner: text.to_string(), repo_id: None })

    } else {

        Err(format!("There is no user or organisation {text}."))

    }

}

    let mut rows = Vec::with_capacity(list.len());

    for r in list {

        let scope = scope_words(app, &r.scope()).await;

        rows.push(RunnerRow { running: running.get(&r.id).copied().unwrap_or(0), scope, r });

    }

    #[serde(rename = "_csrf")]

    csrf_token: String,

}

#[derive(Deserialize)]

pub struct ScopeForm {

    /// Empty for the whole site, an owner, or `owner/repository`.

    #[serde(default)]

    scope: String,

/// `POST /admin/runners/token`: a one-time registration token, shown once,

/// for a runner of the scope given.

pub async fn new_token(State(app): State<Shared>, viewer: Viewer, Form(f): Form<ScopeForm>) -> Result<Page<RunnersPage>, Error> {

    let scope = match parse_scope(&app, &f.scope).await {

        Ok(s) => s,

        Err(e) => return page(&app, shell, None, None, Some(e)).await,

    };

    runners::add_registration(&app.db, &blake3::hash(token.as_bytes()).to_hex(), &me.username, &scope)

    let words = scope_words(&app, &scope).await;

    accounts::audit(&app.db, Some(&me), "flow.runner.token", "", &format!("registration token created for {words}")).await.ok();

    page(&app, shell, Some(token), Some(format!("The runner it registers will take jobs from {words}.")), None).await

}

/// `POST /admin/runners/{id}/scope`: change the repositories a runner serves.

pub async fn set_scope(

    State(app): State<Shared>,

    viewer: Viewer,

    Path(id): Path<i64>,

    Form(f): Form<ScopeForm>,

) -> Result<Page<RunnersPage>, Error> {

    let shell = Shell::new(&app, "/admin/runners", &viewer);

    let me = require_admin(&viewer, &shell)?;

    csrf(&viewer, &f.csrf_token, &shell)?;

    let r =

        runners::get(&app.db, id).await.map_err(|e| Error::from_db(shell.clone(), e))?.ok_or_else(|| Error::not_found(shell.clone()))?;

    let scope = match parse_scope(&app, &f.scope).await {

        Ok(s) => s,

        Err(e) => return page(&app, shell, None, None, Some(e)).await,

    };

    runners::set_scope(&app.db, id, &scope).await.map_err(|e| Error::from_db(shell.clone(), e))?;

    let words = scope_words(&app, &scope).await;

    accounts::audit(&app.db, Some(&me), "flow.runner.scope", &r.name, &words).await.ok();

    page(&app, shell, None, Some(format!("{} now takes jobs from {words}.", r.name)), None).await

/// Runner scopes (oxygenhub#41): a runner registered for one repository, or

/// one owner, takes only those jobs; a site runner takes any.

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

async fn scoped_runners_take_only_their_scope() {

    use oxy_core::identity::SecretKey;

    let h = hub().await;

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

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

    let register = |scope: db::runners::Scope, name: &'static str| {

        let app = h.app.clone();

        async move {

            let token = format!("flowreg_{name}");

            db::runners::add_registration(&app.db, &blake3::hash(token.as_bytes()).to_hex(), "admin", &scope).await.unwrap();

            let identity = SecretKey::generate().identity(name, "", 1);

            let info = serde_json::json!({ "arch": "x86_64", "os": "linux", "version": "0.1.0", "labels": [], "max_jobs": 1 });

            let reg = serde_json::json!({ "token": token, "name": name, "identity": hex::encode(identity.encode()), "info": info });

            let (st, body) = call(&app, "POST", "/api/v1/flow/runners/register", None, Some(reg.to_string())).await;

            assert_eq!(st, StatusCode::OK, "{body}");

            serde_json::from_str::<serde_json::Value>(&body).unwrap()["credential"].as_str().unwrap().to_string()

        }

    };

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

    let only_demo = register(db::runners::Scope { owner: "andy".into(), repo_id: Some(demo.id) }, "demo-runner").await;

    let carols = register(db::runners::Scope { owner: "carol".into(), repo_id: None }, "carol-runner").await;

    let andys = register(db::runners::Scope { owner: "andy".into(), repo_id: None }, "andy-runner").await;

    let claim_as = |who: String| {

        let app = h.app.clone();

        async move {

            let body = serde_json::json!({ "runner": "x", "runtimes": ["microvm"] }).to_string();

            let (st, body) = call(&app, "POST", "/api/v1/flow/claim", Some(&who), Some(body)).await;

            (st == StatusCode::OK).then(|| serde_json::from_str::<serde_json::Value>(&body).unwrap()["repo"].as_str().unwrap().to_string())

        }

    };

    // Jobs queued in andy/other only.

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

    let c = client(&format!("{}/andy/other", h.base), work.path());

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

    c.snapshot("t").unwrap();

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

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

    assert!(!runs_of(&h.app, "andy", "other").await.is_empty());

    assert_eq!(claim_as(only_demo.clone()).await, None, "nothing of andy/demo is queued");

    assert_eq!(claim_as(carols).await, None, "nothing of carol's");

    assert_eq!(claim_as(andys).await.as_deref(), Some("andy/other"), "the owner's runner takes its repositories' jobs");

    // The site runner (hub.toml) takes any.

    assert_eq!(claim(&h.app).await.unwrap()["repo"], "andy/other");

    assert_eq!(claim_as(only_demo).await, None, "still nothing of its own");

}

  - **Scopes** (oxygenhub#41): a runner serves the whole site, one owner's repositories (a user or an organisation), or one repository. The registration token carries the scope (typed as empty, `owner` or `owner/repository` when it is made) and the runner gets it; an administrator can change it on the Runners page. A claim returns only queued jobs of repositories in the runner's scope, filtered before the queue is cut short, so a scoped runner never receives another scope's jobs and never waits behind them. `hub.toml` runners serve the whole site. Changes are audited (`flow.runner.scope`).

  - **The Runners page** lists each runner: online (seen in the last 90 s), offline, draining or disabled; what it serves; signing key; machine; labels; capacity; jobs running now; and when it was last seen.

  <tr><th>Runner</th><th>Status</th><th>Serves</th><th>Signing key</th><th>Machine</th><th>Labels</th><th>Capacity</th><th>Running</th><th>Last seen</th><th></th></tr>

    <td class="small">{{ row.scope }}

      <form class="tiny" method="post" action="/admin/runners/{{ r.id }}/scope" title="Empty for the whole site, an owner, or owner/repository"><input type="hidden" name="_csrf" value="{{ shell.csrf }}"><input type="text" name="scope" size="14" placeholder="owner or owner/repo"> <button>set</button></form></td>

  <p class="small">The runner serves the whole site, one owner's repositories, or one repository; it never takes another scope's jobs.</p>

  <form method="post" action="/admin/runners/token"><input type="hidden" name="_csrf" value="{{ shell.csrf }}">

    <label for="scope">Serves (empty for the whole site, an owner, or owner/repository)</label>

    <input type="text" id="scope" name="scope" placeholder="andymccall or andymccall/oxygen">

    <button type="submit" class="primary-btn">Create a registration token</button></form>