crates/hub/src/diffusion.rs at draft-ready · andymccall/oxygenhub

◆ draft-ready oxygenhub / crates / hub / src / diffusion.rs
66.3 KiB · Rust · last changed One auto-land and one Diffusion bump at a time across hub processes (closes #275) 15 days ago source blame history raw
//! Diffusion (#148, docs/diffusion.md): an upstream's changes reach the
//! repositories that pin it as a subrepo.
//!
//! D1 starts with the reverse pin index (§4.1): when a push moves a
//! repository's default bookmark, or the bookmark its work lands on, each
//! Oxygen subrepo pin in the new state is recorded. That answers "who
//! consumes this repository, at which state, and how far behind", which
//! an upstream's page shows as Used by.

use crate::auth::roles;
use crate::db::{self, accounts::Account, repos::RepoRow, Pool};
use crate::repos::RepoName;
use crate::Shared;
use oxy_core::hash::ChangeHash;
use serde::Serialize;

/// How far back landed proposals teach where a repository's work lands.
pub const LEARN_DAYS: i64 = 90;

/// The bookmark a repository's work lands on (§4.3): the most common
/// target of its proposals landed in the last `LEARN_DAYS`, else its
/// default bookmark.
pub async fn target(pool: &Pool, row: &RepoRow) -> String {
    let since = db::now_ms() - LEARN_DAYS * 24 * 3600 * 1000;
    let learned: Option<String> = sqlx::query_scalar(
        "SELECT target FROM proposals WHERE repo_id = $1 AND state = 'landed' AND landed_ms >= $2
         GROUP BY target ORDER BY COUNT(*) DESC, MAX(landed_ms) DESC LIMIT 1",
    )
    .bind(row.id)
    .bind(since)
    .fetch_optional(pool)
    .await
    .ok()
    .flatten();
    learned.filter(|t| !t.is_empty()).unwrap_or_else(|| row.default_bookmark.clone())
}

/// The bookmarks whose pins are indexed: the default and the learned
/// target (one when they are the same).
pub async fn followed_bookmarks(pool: &Pool, row: &RepoRow) -> Vec<String> {
    let mut out = vec![row.default_bookmark.clone()];
    let t = target(pool, row).await;
    if !out.contains(&t) {
        out.push(t);
    }
    out
}

/// Record `row`'s pins on `bookmark` at `heads`, replacing what was there.
pub async fn record(app: &Shared, row: &RepoRow, bookmark: &str, heads: &[ChangeHash]) {
    // A failure is logged and tried once more after a moment (oxygenhub#210):
    // an index left behind shows a consumer behind with no bump.
    let repo = format!("{}/{}", row.owner, row.name);
    for attempt in 1..=2 {
        match try_record(app, row, bookmark, heads).await {
            Ok(()) => {
                if attempt > 1 {
                    tracing::info!("diffusion: indexed {repo} {bookmark} on the second try");
                }
                return;
            }
            Err(why) if attempt == 1 => {
                tracing::warn!("diffusion: indexing {repo} {bookmark} failed ({why}); trying again");
                tokio::time::sleep(RETRY_AFTER).await;
            }
            Err(why) => {
                FAILURES.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                tracing::error!("diffusion: indexing {repo} {bookmark} failed again ({why}); its pins stay as they were");
            }
        }
    }
}

/// How long [`record`] waits before its second try.
const RETRY_AFTER: std::time::Duration = std::time::Duration::from_millis(250);

/// Indexing failures that the retry did not mend, since the hub started.
pub static FAILURES: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);

/// Failures to force, per repository (tests only: [`fail_next_records`]).
static FORCED: std::sync::Mutex<Option<std::collections::HashMap<String, u32>>> = std::sync::Mutex::new(None);

/// Make the next `n` index writes for a repository fail, as a busy
/// database would (for tests of the retry).
#[doc(hidden)]
pub fn fail_next_records(app: &Shared, repo_id: i64, n: u32) {
    FORCED.lock().unwrap_or_else(|p| p.into_inner()).get_or_insert_with(Default::default).insert(key(app, repo_id), n);
}

/// Forced failures not yet used (tests only).
#[doc(hidden)]
pub fn forced_left(app: &Shared, repo_id: i64) -> u32 {
    FORCED.lock().unwrap_or_else(|p| p.into_inner()).as_ref().and_then(|m| m.get(&key(app, repo_id)).copied()).unwrap_or(0)
}

fn forced(app: &Shared, repo_id: i64) -> bool {
    let mut forced = FORCED.lock().unwrap_or_else(|p| p.into_inner());
    match forced.as_mut().and_then(|m| m.get_mut(&key(app, repo_id))) {
        Some(n) if *n > 0 => {
            *n -= 1;
            true
        }
        _ => false,
    }
}

/// One try at replacing a bookmark's indexed pins; which step failed and
/// why. A state that cannot be read leaves the index as it was rather than
/// emptying it.
async fn try_record(app: &Shared, row: &RepoRow, bookmark: &str, heads: &[ChangeHash]) -> Result<(), String> {
    let Some(name) = RepoName::new(&row.owner, &row.name) else { return Ok(()) };
    let pins = if heads.is_empty() {
        Vec::new()
    } else {
        crate::flow::pins_in(app, name, heads.to_vec()).await.map_err(|e| format!("reading the pins: {e}"))?
    };
    let now = db::now_ms();
    // The upstream's repository on this hub, when it is one.
    let mut upstreams = Vec::new();
    for (_, id, _) in &pins {
        upstreams.push(repo_by_oxy_id(app, &hex::encode(id)).await.map(|r| r.id));
    }
    if forced(app, row.id) {
        return Err("begin: forced for a test".into());
    }
    let mut tx = app.db.begin().await.map_err(|e| format!("begin: {e}"))?;
    sqlx::query("DELETE FROM diffusion_pins WHERE consumer_id = $1 AND bookmark = $2")
        .bind(row.id)
        .bind(bookmark)
        .execute(&mut *tx)
        .await
        .map_err(|e| format!("delete: {e}"))?;
    for ((path, id, pinned), upstream_row) in pins.iter().zip(&upstreams) {
        let heads: Vec<String> = pinned.iter().map(|h| h.to_hex()).collect();
        sqlx::query(
            "INSERT INTO diffusion_pins (consumer_id, bookmark, path, upstream, heads, updated_ms, upstream_repo_id) VALUES ($1, $2, $3, $4, $5, $6, $7)",
        )
        .bind(row.id)
        .bind(bookmark)
        .bind(path)
        .bind(hex::encode(id))
        .bind(heads.join("+"))
        .bind(now)
        .bind(upstream_row)
        .execute(&mut *tx)
        .await
        .map_err(|e| format!("insert {path}: {e}"))?;
    }
    sqlx::query("INSERT INTO diffusion_indexed (consumer_id, bookmark, heads, updated_ms) VALUES ($1, $2, $3, $4) ON CONFLICT (consumer_id, bookmark) DO UPDATE SET heads = excluded.heads, updated_ms = excluded.updated_ms")
        .bind(row.id)
        .bind(bookmark)
        .bind(heads_key(heads))
        .bind(now)
        .execute(&mut *tx)
        .await
        .map_err(|e| format!("noting the state: {e}"))?;
    tx.commit().await.map_err(|e| format!("commit: {e}"))?;
    for u in upstreams.iter().flatten() {
        mark(app, row.id);
        mark(app, *u);
    }
    Ok(())
}

/// After a push: index the pins of each moved bookmark that is followed.
pub async fn after_push(app: &Shared, row: &RepoRow, moved: &[(String, String, Vec<ChangeHash>)]) {
    let followed = followed_bookmarks(&app.db, row).await;
    for (bookmark, _, heads) in moved {
        if followed.contains(bookmark) {
            record(app, row, bookmark, heads).await;
        }
    }
}

static INVOLVED: std::sync::Mutex<Option<std::collections::HashSet<String>>> = std::sync::Mutex::new(None);

fn key(app: &Shared, repo_id: i64) -> String {
    format!("{}#{repo_id}", app.repos.root().display())
}

fn mark(app: &Shared, repo_id: i64) {
    INVOLVED.lock().unwrap_or_else(|p| p.into_inner()).get_or_insert_with(Default::default).insert(key(app, repo_id));
}

/// Whether a repository takes part in Diffusion, as an upstream or a
/// consumer on this hub: its Diffusion tab shows (#150).
pub fn involved(app: &Shared, repo_id: i64) -> bool {
    INVOLVED.lock().unwrap_or_else(|p| p.into_inner()).as_ref().is_some_and(|s| s.contains(&key(app, repo_id)))
}

/// Index every repository's followed bookmarks, for a hub that had no
/// index before (run once at start, in the background).
pub async fn backfill(app: &Shared) {
    let Ok(rows) = db::repos::list(&app.db).await else { return };
    for row in rows {
        let Some(name) = RepoName::new(&row.owner, &row.name) else { continue };
        for bookmark in followed_bookmarks(&app.db, &row).await {
            let b = bookmark.clone();
            let heads = app.repos.read(&name, move |r, _| heads_of(r, &b)).await;
            if let Ok(heads) = heads {
                record(app, &row, &bookmark, &heads).await;
            }
        }
    }
}

/// Heads as one comparable string: sorted, joined with `+`.
fn heads_key(heads: &[ChangeHash]) -> String {
    let mut h: Vec<String> = heads.iter().map(|h| h.to_hex()).collect();
    h.sort();
    h.join("+")
}

/// How often [`reconciler`] looks.
pub const RECONCILE_EVERY: std::time::Duration = std::time::Duration::from_secs(10 * 60);

/// What one reconciling pass mended.
#[derive(Debug, Default, PartialEq, Eq)]
pub struct Repairs {
    /// Followed bookmarks whose index was behind the bookmark.
    pub reindexed: usize,
    /// Consumers behind an upstream's newest passing state with no open
    /// bump, bumped now.
    pub caught_up: usize,
}

/// Every [`RECONCILE_EVERY`], mend the pin index and the bumps it drives
/// (oxygenhub#210).
pub async fn reconciler(app: Shared) {
    loop {
        tokio::time::sleep(RECONCILE_EVERY).await;
        let r = reconcile(&app).await;
        if r != Repairs::default() {
            tracing::warn!("diffusion: reconciled: {} bookmarks re-indexed, {} consumers caught up", r.reindexed, r.caught_up);
        }
    }
}

/// One pass: re-index each followed bookmark whose indexed state is not
/// the state it is at now; then bump each consumer whose pin is not its
/// upstream's newest passing state and that has no open bump for it.
pub async fn reconcile(app: &Shared) -> Repairs {
    let mut repairs = Repairs::default();
    let Ok(rows) = db::repos::list(&app.db).await else { return repairs };
    for row in &rows {
        let Some(name) = RepoName::new(&row.owner, &row.name) else { continue };
        for bookmark in followed_bookmarks(&app.db, row).await {
            let b = bookmark.clone();
            let Ok(heads) = app.repos.read(&name, move |r, _| heads_of(r, &b)).await else { continue };
            let indexed: Option<String> =
                sqlx::query_scalar("SELECT heads FROM diffusion_indexed WHERE consumer_id = $1 AND bookmark = $2")
                    .bind(row.id)
                    .bind(&bookmark)
                    .fetch_optional(&app.db)
                    .await
                    .ok()
                    .flatten();
            if indexed.as_deref() != Some(heads_key(&heads).as_str()) {
                tracing::info!("diffusion: {}/{} {bookmark}: the index was not at the bookmark; re-indexing", row.owner, row.name);
                record(app, row, &bookmark, &heads).await;
                repairs.reindexed += 1;
            }
        }
    }
    // Consumers behind with no open bump.
    let pins: Vec<(i64, String, String, String)> =
        sqlx::query_as("SELECT consumer_id, path, upstream, heads FROM diffusion_pins WHERE upstream_repo_id IS NOT NULL")
            .fetch_all(&app.db)
            .await
            .unwrap_or_default();
    let mut caught = std::collections::HashSet::new();
    for (consumer_id, path, upstream, pinned) in pins {
        if caught.contains(&consumer_id) {
            continue;
        }
        let Some(up) = repo_by_oxy_id(app, &upstream).await else { continue };
        let Some(green) = last_green(app, &up).await else { continue };
        if heads_key(&green) == heads_key(&pinned.split('+').filter_map(ChangeHash::from_hex).collect::<Vec<_>>()) {
            continue;
        }
        let Ok(Some(consumer)) = db::repos::get_by_id(&app.db, consumer_id).await else { continue };
        let Some(cname) = RepoName::new(&consumer.owner, &consumer.name) else { continue };
        let learned = target(&app.db, &consumer).await;
        let modules = oxymodules(app, &cname, &learned).await;
        if rule(&app.config.diffusion.follow, &consumer, modules.as_deref(), &path).follow == Follow::Off {
            continue;
        }
        // The newest bump at this path: open or a draft is a bump already,
        // and one a person closed is a "no" the reconciler respects (the
        // upstream's next move proposes again as usual).
        let bump_bm = bump_bookmark(&path);
        let newest = || async {
            db::proposals::list(&app.db, consumer_id, &[])
                .await
                .unwrap_or_default()
                .into_iter()
                .filter(|p| p.source_bookmark == bump_bm)
                .max_by_key(|p| p.number)
                .map(|p| p.state)
        };
        let has_bump = || async { matches!(newest().await.as_deref(), Some("open" | "draft")) };
        if matches!(newest().await.as_deref(), Some("open" | "draft" | "closed")) {
            continue;
        }
        caught.insert(consumer_id);
        catch_up(app, &consumer).await;
        if has_bump().await {
            tracing::info!(
                "diffusion: {}/{} {path} was behind {}/{} with no bump; bumped",
                consumer.owner,
                consumer.name,
                up.owner,
                up.name
            );
            repairs.caught_up += 1;
        }
    }
    repairs
}

/// A bookmark's heads as the repository has them now (not a cached
/// list, which a push refreshes a moment later); empty when it has none.
fn heads_of(r: &oxy_core::repo::Repo, bookmark: &str) -> Result<Vec<ChangeHash>, crate::repos::RepoError> {
    Ok(r.bookmarks()?.into_iter().find(|b| b.name == bookmark).map(|b| b.heads).unwrap_or_default())
}

/// One consumer of an upstream, as an upstream's page and API show it.
#[derive(Clone, Debug, Serialize)]
pub struct Consumer {
    pub owner: String,
    pub name: String,
    pub bookmark: String,
    pub path: String,
    /// The upstream heads it pins, hex.
    pub pinned: Vec<String>,
    /// Upstream changes on its default bookmark the pin lacks; `None` when
    /// the hub cannot tell (the pinned changes are not in the upstream).
    pub behind: Option<usize>,
    /// How it follows the upstream (from its setting, or the site's).
    pub follow: Follow,
    /// Its open bump proposal's number, if it has one.
    pub bump: Option<i64>,
    /// The newest run of its bump bookmark: number and status.
    pub run: Option<(i64, String)>,
    /// Its last landed bump: number and when.
    pub landed: Option<(i64, i64)>,
    /// What the views show (#150).
    pub colour: Colour,
    /// The pill's two halves: the upstream, then this consumer.
    pub upstream: Half,
    pub downstream: Half,
}

impl Consumer {
    /// As the API gives it (oxygenhub#153), with its upstream's name.
    pub fn api(&self, upstream: &str) -> hub_api::DiffusionPin {
        let half = |h: &Half| hub_api::DiffusionHalf {
            tone: h.class.trim_start_matches("df-").to_string(),
            word: h.word.to_string(),
            title: h.title.clone(),
        };
        hub_api::DiffusionPin {
            upstream: upstream.to_string(),
            consumer: format!("{}/{}", self.owner, self.name),
            path: self.path.clone(),
            bookmark: self.bookmark.clone(),
            behind: self.behind.map(|n| n as u64),
            follow: self.follow.as_str().to_string(),
            bump: self.bump,
            run: self.run.as_ref().map(|r| r.0),
            run_status: self.run.as_ref().map(|r| r.1.clone()),
            upstream_state: half(&self.upstream),
            consumer_state: half(&self.downstream),
        }
    }
}

/// One half of a Diffusion pill: a colour class and a word (#150).
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
pub struct Half {
    pub class: &'static str,
    pub word: &'static str,
    pub title: String,
}

impl Half {
    /// The upstream's half: was the upstream itself tested?
    pub fn upstream(unverified: bool, run: Option<&str>, slug: &str) -> Half {
        let (class, word, title) = match (unverified, run) {
            (true, _) => ("df-amber", "untested", format!("{slug} has no checks of its own: nothing has tested it")),
            (false, Some("passed")) => ("df-green", "checked", format!("{slug}'s newest run passed")),
            (false, Some("failed" | "error" | "cancelled")) => ("df-red", "failing", format!("{slug}'s newest run failed")),
            (false, _) => ("df-grey", "testing", format!("{slug}'s newest run has not finished")),
        };
        Half { class, word, title }
    }

    /// The consumer's half: does it pass with the upstream?
    pub fn downstream(follow: Follow, behind: Option<usize>, bump: Option<i64>, run: Option<&str>, name: &str) -> Half {
        let (class, word, title) = if follow == Follow::Off {
            ("df-off", "not following", format!("{name} does not follow it"))
        } else {
            match (bump, run) {
                (Some(_), Some("passed")) => ("df-green", "passes", format!("{name}'s run on the bump passed")),
                (Some(_), Some("failed" | "error" | "cancelled")) => ("df-red", "fails", format!("{name}'s run on the bump failed")),
                (Some(_), Some(_)) => ("df-grey", "testing", format!("{name}'s run on the bump has not finished")),
                (Some(_), None) => ("df-grey", "waiting", format!("{name}'s bump has no run yet")),
                (None, _) if behind == Some(0) => ("df-green", "up to date", format!("{name} pins the newest state")),
                (None, _) => ("df-grey", "waiting", format!("{name} is behind; its bump has not been made yet")),
            }
        };
        Half { class, word, title }
    }
}

/// A consumer's Diffusion status at a glance (#150, Andy's colours):
/// green when the upstream was checked and the consumer's run on the bump
/// passed; amber when the run passed but the upstream has no checks; red
/// when the run failed; grey while it runs; up to date when there is
/// nothing to bump; off when it does not follow.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
#[serde(rename_all = "kebab-case")]
pub enum Colour {
    Green,
    Amber,
    Red,
    Grey,
    UpToDate,
    Off,
}

impl Colour {
    pub fn class(&self) -> &'static str {
        match self {
            Colour::Green => "df-green",
            Colour::Amber => "df-amber",
            Colour::Red => "df-red",
            Colour::Grey => "df-grey",
            Colour::UpToDate => "df-current",
            Colour::Off => "df-off",
        }
    }

    pub fn label(&self) -> &'static str {
        match self {
            Colour::Green => "green",
            Colour::Amber => "amber: upstream unverified",
            Colour::Red => "red: its run failed",
            Colour::Grey => "testing",
            Colour::UpToDate => "up to date",
            Colour::Off => "not following",
        }
    }

    /// The colour from what is known of a consumer.
    pub fn of(follow: Follow, behind: Option<usize>, bump: Option<i64>, run: Option<&str>, unverified: bool) -> Colour {
        if follow == Follow::Off {
            return Colour::Off;
        }
        match (bump, run) {
            (Some(_), Some("passed")) if unverified => Colour::Amber,
            (Some(_), Some("passed")) => Colour::Green,
            (Some(_), Some("failed" | "error" | "cancelled")) => Colour::Red,
            (Some(_), _) => Colour::Grey,
            (None, _) if behind == Some(0) => Colour::UpToDate,
            (None, _) => Colour::Grey,
        }
    }
}

/// The consumers of `row` that `account` may read, with how far behind
/// the upstream's default bookmark each pin is.
pub async fn consumers(app: &Shared, row: &RepoRow, account: Option<&Account>) -> Result<Vec<Consumer>, crate::repos::RepoError> {
    let Some(name) = RepoName::new(&row.owner, &row.name) else { return Ok(Vec::new()) };
    let id = app.repos.read(&name, |r, _| Ok(r.id()?)).await?;
    let found: Vec<(i64, String, String, String)> = sqlx::query_as(
        "SELECT consumer_id, bookmark, path, heads FROM diffusion_pins WHERE upstream = $1 ORDER BY consumer_id, bookmark, path",
    )
    .bind(hex::encode(id))
    .fetch_all(&app.db)
    .await
    .map_err(|e| crate::repos::RepoError::Io(std::io::Error::other(e)))?;
    let mut out = Vec::new();
    let mut pinned_sets = Vec::new();
    let unverified = unverified(app, row).await;
    for (consumer_id, bookmark, path, heads) in found {
        let Ok(Some(c)) = db::repos::get_by_id(&app.db, consumer_id).await else { continue };
        if !roles::allows(roles::role_for(&app.db, account, &c).await.ok().flatten(), crate::auth::Access::Read) {
            continue;
        }
        // One row per consumer and subrepo: the bookmark its work lands on.
        if bookmark != target(&app.db, &c).await {
            continue;
        }
        let pinned: Vec<String> = heads.split('+').filter(|h| !h.is_empty()).map(String::from).collect();
        pinned_sets.push(pinned.iter().filter_map(|h| ChangeHash::from_hex(h)).collect::<Vec<_>>());
        let bump_bm = bump_bookmark(&path);
        let mine: Vec<db::proposals::ProposalRow> = db::proposals::list(&app.db, c.id, &[])
            .await
            .unwrap_or_default()
            .into_iter()
            .filter(|p| p.source_bookmark == bump_bm)
            .collect();
        let bump = mine.iter().find(|p| p.state == "open" || p.state == "draft").map(|p| p.number);
        let landed = mine.iter().filter(|p| p.state == "landed").max_by_key(|p| p.landed_ms).map(|p| (p.number, p.landed_ms));
        let run: Option<(i64, String)> =
            sqlx::query_as("SELECT number, status FROM flow_runs WHERE repo_id = $1 AND bookmark = $2 ORDER BY id DESC LIMIT 1")
                .bind(c.id)
                .bind(&bump_bm)
                .fetch_optional(&app.db)
                .await
                .ok()
                .flatten();
        let follow = rule(&app.config.diffusion.follow, &c, None, &path).follow;
        out.push(Consumer {
            owner: c.owner,
            name: c.name,
            bookmark,
            path,
            pinned,
            behind: None,
            follow,
            bump,
            run,
            landed,
            colour: Colour::Off,
            upstream: Half { class: "df-grey", word: "", title: String::new() },
            downstream: Half { class: "df-grey", word: "", title: String::new() },
        });
    }
    if out.is_empty() {
        return Ok(out);
    }
    let default = row.default_bookmark.clone();
    let behind = app
        .repos
        .read(&name, move |r, cache| {
            let tip = heads_of(r, &default)?;
            let tip = cache.summary(r, &tip)?;
            Ok(pinned_sets
                .iter()
                .map(|p| cache.summary(r, p).ok().map(|s| tip.changes.iter().filter(|c| !s.changes.contains(*c)).count()))
                .collect::<Vec<_>>())
        })
        .await?;
    let up_run: Option<String> =
        sqlx::query_scalar("SELECT status FROM flow_runs WHERE repo_id = $1 AND bookmark = $2 ORDER BY id DESC LIMIT 1")
            .bind(row.id)
            .bind(&row.default_bookmark)
            .fetch_optional(&app.db)
            .await
            .ok()
            .flatten();
    let slug = format!("{}/{}", row.owner, row.name);
    for (c, b) in out.iter_mut().zip(behind) {
        c.behind = b;
        let run = c.run.as_ref().map(|r| r.1.clone());
        c.colour = Colour::of(c.follow, b, c.bump, run.as_deref(), unverified);
        c.upstream = Half::upstream(unverified, up_run.as_deref(), &slug);
        c.downstream = Half::downstream(c.follow, b, c.bump, run.as_deref(), &format!("{}/{}", c.owner, c.name));
    }
    Ok(out)
}

/// What `consumer` follows, as its Follows panel shows: each upstream on
/// this hub it pins, with its status there.
pub async fn follows(app: &Shared, consumer: &RepoRow, account: Option<&Account>) -> Vec<(RepoRow, Consumer)> {
    let ups: Vec<i64> =
        sqlx::query_scalar("SELECT DISTINCT upstream_repo_id FROM diffusion_pins WHERE consumer_id = $1 AND upstream_repo_id IS NOT NULL")
            .bind(consumer.id)
            .fetch_all(&app.db)
            .await
            .unwrap_or_default();
    let mut out = Vec::new();
    for id in ups {
        let Ok(Some(up)) = db::repos::get_by_id(&app.db, id).await else { continue };
        if !roles::allows(roles::role_for(&app.db, account, &up).await.ok().flatten(), crate::auth::Access::Read) {
            continue;
        }
        for c in consumers(app, &up, account).await.unwrap_or_default() {
            if c.owner == consumer.owner && c.name == consumer.name {
                out.push((up.clone(), c));
            }
        }
    }
    out
}

/// Every upstream on this hub with its consumers, for Admin › Diffusion.
pub async fn everything(app: &Shared, account: Option<&Account>) -> Vec<(RepoRow, Vec<Consumer>, bool)> {
    let ups: Vec<i64> = sqlx::query_scalar("SELECT DISTINCT upstream_repo_id FROM diffusion_pins WHERE upstream_repo_id IS NOT NULL")
        .fetch_all(&app.db)
        .await
        .unwrap_or_default();
    let mut out = Vec::new();
    for id in ups {
        let Ok(Some(up)) = db::repos::get_by_id(&app.db, id).await else { continue };
        let list = consumers(app, &up, account).await.unwrap_or_default();
        let unverified = unverified(app, &up).await;
        out.push((up, list, unverified));
    }
    out
}

/// The bookmark a consumer's bump for the subrepo at `path` lives on.
/// A bump change's message: its title names the upstream state by its
/// first head and head count, so a state of many heads does not make it
/// run on (oxygenhub#216); the whole address is in the body.
pub fn bump_message(path: &str, upstream: &str, heads: &[String]) -> String {
    let first = heads.first().map(String::as_str).unwrap_or("");
    match heads.len() {
        0 | 1 => format!("Diffusion: move {path} to {upstream} {first}"),
        n => format!("Diffusion: move {path} to {upstream} {first} (+{} heads)\n\nPins {upstream} at {}", n - 1, heads.join("+")),
    }
}

pub fn bump_bookmark(path: &str) -> String {
    format!("diffusion/{}", path.trim_matches('/'))
}

/// What a bump did.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum Bumped {
    /// The bump bookmark now holds the target plus one change moving the pin.
    Moved { bookmark: String, change: ChangeHash, heads: Vec<ChangeHash>, from: Vec<ChangeHash>, state: String },
    /// The pin already holds the upstream state.
    UpToDate,
    /// The bump bookmark already moves the pin to that state.
    Already { from: Vec<ChangeHash>, to: Vec<ChangeHash> },
    /// The pin names changes the upstream does not hold: not followed.
    Unknown,
    /// Taking the upstream's state with the pin's own extra changes would
    /// conflict.
    Conflicts,
    /// The consumer has no Oxygen pin of that upstream at `path` on its
    /// target bookmark.
    NoPin,
}

/// Where a consumer's pin `old` moves when the upstream reaches `new`,
/// worked out in the upstream (§4.3).
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum Plan {
    /// To these heads, which add this many changes.
    To(Vec<ChangeHash>, usize),
    /// Nothing to add.
    UpToDate,
    /// The pin names changes the upstream does not hold (rewritten, or
    /// from elsewhere): not followed.
    Unknown,
    /// Taking `new` with the pin's own extra changes would conflict.
    Conflicts,
}

/// The pin `old` moved to the upstream's `new`. When the pin has changes
/// `new` lacks (say, from an upstream proposal not landed yet), they are
/// kept: changes that do not touch each other combine, so the move is to
/// both, never dropping what the consumer already relies on.
pub async fn plan(app: &Shared, upstream: &RepoName, old: Vec<ChangeHash>, new: Vec<ChangeHash>) -> Result<Plan, crate::repos::RepoError> {
    app.repos
        .read(upstream, move |r, cache| {
            let Ok(o) = cache.summary(r, &old) else { return Ok(Plan::Unknown) };
            let n = cache.summary(r, &new)?;
            if o.changes.iter().all(|c| n.changes.contains(c)) {
                let added = n.changes.len() - o.changes.len();
                return Ok(if added == 0 { Plan::UpToDate } else { Plan::To(new, added) });
            }
            let mut both = new.clone();
            both.extend(old.iter().copied());
            let heads = r.load_state(&both)?.heads();
            let b = cache.summary(r, &heads)?;
            if b.conflicted && !o.conflicted {
                return Ok(Plan::Conflicts);
            }
            let added = b.changes.len() - o.changes.len();
            Ok(if added == 0 { Plan::UpToDate } else { Plan::To(heads, added) })
        })
        .await
}

/// Move `consumer`'s pin of `upstream` at `path` to `new_heads`, on a bump
/// bookmark built from `target` (§4.3): one change, signed with the hub's
/// key, on the target's heads. The upstream decides forwardness.
pub async fn bump(
    app: &Shared,
    consumer: &RepoRow,
    target: &str,
    path: &str,
    upstream: &RepoRow,
    new_heads: Vec<ChangeHash>,
) -> Result<Bumped, crate::repos::RepoError> {
    use oxy_core::identity::Domain;
    use oxy_core::subrepo::Pin;
    let (Some(cname), Some(uname)) = (RepoName::new(&consumer.owner, &consumer.name), RepoName::new(&upstream.owner, &upstream.name))
    else {
        return Ok(Bumped::NoPin);
    };
    let upstream_id = app.repos.read(&uname, |r, _| Ok(r.id()?)).await?;
    // The consumer's current pin at `path` on its target, and on its bump
    // bookmark if it has one.
    let (t, p) = (target.to_string(), path.trim_matches('/').to_string());
    let (current, bumped) = app
        .repos
        .read(&cname, move |r, cache| {
            let pin_on = |bookmark: &str| -> Result<Option<Pin>, crate::repos::RepoError> {
                let heads = heads_of(r, bookmark)?;
                if heads.is_empty() {
                    return Ok(None);
                }
                let loaded = cache.load(r, &heads)?;
                let Some(f) = loaded.snapshot.files.get(&p) else { return Ok(None) };
                let oxy_core::state::FileContent::Opaque { blobs } = &f.content else { return Ok(None) };
                let Some(b) = blobs.first() else { return Ok(None) };
                Ok(Pin::decode(&r.read_blob(b)?).ok())
            };
            Ok((pin_on(&t)?, pin_on(&bump_bookmark(&p))?))
        })
        .await?;
    let Some(Pin::Oxy { repo, heads: old, bookmark }) = current else { return Ok(Bumped::NoPin) };
    if repo != upstream_id {
        return Ok(Bumped::NoPin);
    }
    let new_heads = match plan(app, &uname, old.clone(), new_heads.clone()).await? {
        Plan::To(heads, _) => heads,
        Plan::UpToDate => return Ok(Bumped::UpToDate),
        Plan::Unknown => return Ok(Bumped::Unknown),
        Plan::Conflicts => return Ok(Bumped::Conflicts),
    };
    if let Some(Pin::Oxy { heads: b, .. }) = &bumped {
        let (mut b, mut n) = (b.clone(), new_heads.clone());
        b.sort();
        n.sort();
        if b == n {
            return Ok(Bumped::Already { from: old, to: new_heads });
        }
    }
    let (key, secret) =
        crate::keys::load_secret(&app.config.keys_dir()).map_err(|e| crate::repos::RepoError::Io(std::io::Error::other(e.to_string())))?;
    let identity = app.hub_key.identity.clone();
    let upstream_label = format!("{}/{}", upstream.owner, upstream.name);
    let (t, p) = (target.to_string(), path.trim_matches('/').to_string());
    let pin = Pin::Oxy { repo, heads: new_heads.clone(), bookmark };
    let old = old.clone();
    let short: Vec<String> = new_heads.iter().map(|h| h.to_hex()[..12].to_string()).collect();
    let message = bump_message(&p, &upstream_label, &short);
    app.repos
        .write(&cname, move |r, _| {
            let base = heads_of(r, &t)?;
            let change = r.build_pin_change(&base, &p, &pin, &message, key)?;
            let hash = change.hash();
            let sig = secret.sign(Domain::Change, &key, &hash);
            let mut state = r.load_state(&base)?;
            state.add(change.clone()).map_err(|e| crate::repos::RepoError::Oxy(oxy_core::repo::RepoError::Msg(e.to_string())))?;
            let mut candidates = base.clone();
            candidates.push(hash);
            let heads = state.maximal(&candidates);
            let bookmark = bump_bookmark(&p);
            // The bump bookmark is the hub's own and is rebuilt on the
            // target each time, so the bump it replaces may go.
            let mut update = r.host_ref_update(&bookmark, &heads, Some((&key, &secret)))?;
            update.force = true;
            let state = update.new_state.to_hex();
            let report = r.apply_hosted(Some((change, vec![sig])), &[identity], update)?;
            if let Some(refused) = report.results.iter().find(|x| !x.ok) {
                return Err(crate::repos::RepoError::Oxy(oxy_core::repo::RepoError::Msg(format!(
                    "the bump was refused: {}",
                    refused.reason
                ))));
            }
            Ok(Bumped::Moved { bookmark, change: hash, heads, from: old, state })
        })
        .await
}

/// The account Diffusion's proposals are opened as.
pub const ACCOUNT: &str = "diffusion";

/// The `diffusion` account, made on first use with no way to sign in.
pub async fn account(pool: &Pool) -> Option<Account> {
    if let Ok(Some(a)) = db::accounts::account_by_name(pool, ACCOUNT).await {
        return Some(a);
    }
    db::accounts::create_account(pool, ACCOUNT, "diffusion@localhost", "!", "", false).await.ok()
}

/// Whether `path` matches a `watch` pattern: `*` is any run of characters
/// within one path segment, `**` any run across segments, `?` one
/// character; a pattern without a wildcard also matches everything below
/// it, as a directory would.
pub fn watched(pattern: &str, path: &str) -> bool {
    fn m(p: &[u8], s: &[u8]) -> bool {
        match (p.first(), s.first()) {
            (None, None) => true,
            (Some(b'*'), _) if p.get(1) == Some(&b'*') => (0..=s.len()).any(|i| m(&p[2..], &s[i..])),
            (Some(b'*'), _) => (0..=s.len()).take_while(|&i| i == 0 || s[i - 1] != b'/').any(|i| m(&p[1..], &s[i..])),
            (Some(b'?'), Some(c)) if *c != b'/' => m(&p[1..], &s[1..]),
            (Some(a), Some(b)) if a == b => m(&p[1..], &s[1..]),
            _ => false,
        }
    }
    let p = pattern.trim().trim_matches('/');
    if p.is_empty() {
        return false;
    }
    m(p.as_bytes(), path.as_bytes()) || (!p.contains(['*', '?']) && path.starts_with(&format!("{p}/")))
}

/// What a bump brings: the upstream changes in `new` and not in `old`,
/// newest first, as Markdown lines linking to the upstream's change pages,
/// with those touching a `watch` pattern flagged (and listed again).
pub struct Changelog {
    pub lines: Vec<String>,
    pub watched: Vec<String>,
}

pub async fn changelog(app: &Shared, upstream: &RepoRow, old: Vec<ChangeHash>, new: Vec<ChangeHash>, watch: Vec<String>) -> Changelog {
    let Some(name) = RepoName::new(&upstream.owner, &upstream.name) else { return Changelog { lines: Vec::new(), watched: Vec::new() } };
    let linked: std::collections::HashMap<String, String> =
        db::accounts::identity_accounts(&app.db).await.unwrap_or_default().into_iter().collect();
    let base = format!("/{}/{}", upstream.owner, upstream.name);
    app.repos
        .read(&name, move |r, cache| {
            let (o, n) = (cache.summary(r, &old)?, cache.summary(r, &new)?);
            let mut added: Vec<(u64, ChangeHash, String, String, Vec<String>)> = Vec::new();
            for h in n.changes.iter().filter(|c| !o.changes.contains(*c)) {
                let Ok(c) = r.store().get_change(h) else { continue };
                let key = c.author.to_hex();
                let who = linked.get(&key).cloned().unwrap_or_else(|| key[..12].to_string());
                let first = c.message.lines().next().unwrap_or("").trim().to_string();
                // The watched paths it touches, when the game watches any.
                let hits: Vec<String> = if watch.is_empty() {
                    Vec::new()
                } else {
                    crate::repos::view::change(r, cache, h)
                        .map(|d| d.files.into_iter().map(|f| f.path).filter(|p| watch.iter().any(|w| watched(w, p))).collect())
                        .unwrap_or_default()
                };
                added.push((c.hlc.wall_ms, *h, first, who, hits));
            }
            added.sort_by_key(|a| std::cmp::Reverse(a.0));
            let line = |h: &ChangeHash, msg: &str, who: &str, flag: bool| {
                format!(
                    "- {}[`{}`]({base}/change/{}) {} ({who})",
                    if flag { "⚑ " } else { "" },
                    &h.to_hex()[..12],
                    h.to_hex(),
                    msg.replace('\n', " ")
                )
            };
            let watched = added
                .iter()
                .filter(|a| !a.4.is_empty())
                .map(|(_, h, msg, who, hits)| {
                    format!(
                        "{} — touches {}",
                        line(h, msg, who, true),
                        hits.iter().map(|p| format!("`{p}`")).collect::<Vec<_>>().join(", ")
                    )
                })
                .collect();
            let lines = added.iter().map(|(_, h, msg, who, hits)| line(h, msg, who, !hits.is_empty())).collect();
            Ok(Changelog { lines, watched })
        })
        .await
        .unwrap_or(Changelog { lines: Vec::new(), watched: Vec::new() })
}

/// Why a bump was made, which its proposal says.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Why {
    /// The upstream's state passed its checks.
    Green,
    /// The upstream moved (the consumer follows every move).
    Moved,
    /// The upstream has no checks at all: nothing vouched for it, so the
    /// proposal is marked amber, `upstream unverified`.
    Unverified,
}

/// Open the bump proposal for `path`'s bump bookmark, or bring the open
/// one up to date: into `target`, by the `diffusion` account, labelled
/// `dependencies` (and `upstream unverified` while that is so), its body
/// the changelog from the pin `from` to `to`. Its number.
pub async fn propose(
    app: &Shared,
    consumer: &RepoRow,
    target: &str,
    path: &str,
    upstream: &RepoRow,
    (from, to): (Vec<ChangeHash>, Vec<ChangeHash>),
    (why, watch, lands): (Why, Vec<String>, bool),
) -> Option<i64> {
    use crate::db::proposals as pdb;
    let bookmark = bump_bookmark(path);
    let bookmark = bookmark.as_str();
    let me = account(&app.db).await?;
    let log = changelog(app, upstream, from, to.clone(), watch.clone()).await;
    let lines = log.lines;
    let flagged = if log.watched.is_empty() {
        String::new()
    } else {
        format!(
            "\n\n**Touches what you watch** ({}):\n\n{}",
            watch.iter().map(|w| format!("`{w}`")).collect::<Vec<_>>().join(", "),
            log.watched.join("\n")
        )
    };
    let slug = format!("{}/{}", upstream.owner, upstream.name);
    let short: Vec<String> = to.iter().map(|h| h.to_hex()[..12].to_string()).collect();
    let ub = &upstream.default_bookmark;
    let count = if lines.len() == 1 { "1 change".to_string() } else { format!("{} changes", lines.len()) };
    let title = format!("Diffusion: {path} to {slug} {ub} ({count})");
    let when = match why {
        Why::Green => "at its newest state whose checks passed",
        Why::Moved => "as it is now",
        Why::Unverified => "as it is now",
    };
    let amber = if why == Why::Unverified {
        format!("\n\n> **Upstream unverified.** {slug} has no checks of its own, so nothing has tested this state of it. Only this repository's pipeline tests the move; a later bump from a passing run of {slug} clears this.")
    } else {
        String::new()
    };
    let body = format!(
        "Moves the subrepo `{path}` to [{slug}](/{slug})'s `{ub}`, {when} (`{}`). Only the pin changes; this repository's own checks test it.{amber}{flagged}\n\n**What it brings** ({count}):\n\n{}\n\nOpened and kept up to date by Diffusion (docs/diffusion.md). {}",
        crate::web::state_label(&short.join("+")),
        lines.join("\n"),
        if lands {
            "This repository lands its bumps itself (`land = auto`): it lands once the upstream state it pins has passed its own checks and this repository's run on it passes, unless its required checks, approvals or trust policy say otherwise. An untested upstream always waits for a person."
        } else {
            "It never lands by itself."
        }
    );
    let open = pdb::list(&app.db, consumer.id, &["open", "draft"]).await.ok()?.into_iter().find(|p| p.source_bookmark == bookmark);
    let (number, id) = match open {
        Some(p) => {
            if p.title != title || p.body != body {
                pdb::edit(&app.db, p.id, &title, &body, target).await.ok()?;
                pdb::add_event(&app.db, p.id, ACCOUNT, "updated", &crate::web::state_label(&short.join("+"))).await.ok();
            }
            (p.number, p.id)
        }
        None => {
            let p = pdb::create(
                &app.db,
                pdb::NewProposal {
                    repo_id: consumer.id,
                    title: &title,
                    body: &body,
                    author_id: Some(me.id),
                    author: ACCOUNT,
                    draft: false,
                    target,
                    source_kind: "bookmark",
                    source_repo_id: None,
                    source_bookmark: bookmark,
                    source_changes: "",
                },
            )
            .await
            .ok()?;
            pdb::add_event(&app.db, p.id, ACCOUNT, "opened", "").await.ok();
            db::notify::index(&app.db, "proposal", consumer.id, p.id, p.number, &p.title, &p.body).await.ok();
            // Told as a proposal opened on the web is (oxygenhub#183).
            crate::notify::deliver(
                app,
                crate::notify::Event {
                    kind: "proposal.opened",
                    title: format!("{}/{}#{}: {}", consumer.owner, consumer.name, p.number, p.title),
                    url: format!("/{}/{}/proposal/{}", consumer.owner, consumer.name, p.number),
                    actor: &me,
                    repo_id: consumer.id,
                    thread: Some(("proposal", p.id)),
                    mention_text: "",
                    direct: vec![],
                },
            )
            .await;
            if let Ok(label) =
                db::issues::upsert_label(&app.db, consumer.id, "dependencies", "#0072ff", "A subrepo moves (Diffusion)").await
            {
                db::issues::set_proposal_label(&app.db, p.id, label.id, true).await.ok();
            }
            (p.number, p.id)
        }
    };
    // Amber while the upstream has no checks; cleared by a green bump.
    let unverified = db::issues::label_by_name(&app.db, consumer.id, "upstream unverified").await.ok().flatten();
    match (why == Why::Unverified, unverified) {
        (true, l) => {
            let l = match l {
                Some(l) => Some(l),
                None => db::issues::upsert_label(
                    &app.db,
                    consumer.id,
                    "upstream unverified",
                    "#e3a008",
                    "The upstream has no checks (Diffusion)",
                )
                .await
                .ok(),
            };
            if let Some(l) = l {
                db::issues::set_proposal_label(&app.db, id, l.id, true).await.ok();
            }
        }
        (false, Some(l)) => {
            db::issues::set_proposal_label(&app.db, id, l.id, false).await.ok();
        }
        (false, None) => {}
    }
    Some(number)
}

/// Start a bump's pipeline, as a push of it would. Boxed as `Send`, which
/// breaks the cycle between a bump starting runs and a passing run
/// starting bumps that the compiler cannot see through.
fn start_runs(
    app: Shared,
    row: RepoRow,
    bookmark: String,
    state: String,
    heads: Vec<ChangeHash>,
) -> std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send>> {
    Box::pin(async move { crate::flow::on_push(&app, &row, &bookmark, &state, &heads, ACCOUNT).await })
}

/// The text of `.oxymodules` on `bookmark` of `name`, if any.
async fn oxymodules(app: &Shared, name: &RepoName, bookmark: &str) -> Option<String> {
    let b = bookmark.to_string();
    app.repos
        .read(name, move |r, cache| {
            let heads = heads_of(r, &b)?;
            if heads.is_empty() {
                return Ok(None);
            }
            let loaded = cache.load(r, &heads)?;
            Ok(match loaded.snapshot.files.get(".oxymodules").map(|f| &f.content) {
                Some(oxy_core::state::FileContent::Text { bytes, .. }) => Some(String::from_utf8_lossy(bytes).into_owned()),
                _ => None,
            })
        })
        .await
        .ok()
        .flatten()
}

/// The upstream `upstream` reached `heads` on its default bookmark: by a
/// push (`green` false) or a passing run of that state (`green` true).
/// Bump and propose in every consumer that follows it that way (§4.3).
pub async fn propagate(app: &Shared, upstream: &RepoRow, heads: Vec<ChangeHash>, green: bool) {
    propagate_to(app, upstream, heads, green, None).await
}

/// `propagate`, to one consumer only when `only` names it.
async fn propagate_to(app: &Shared, upstream: &RepoRow, heads: Vec<ChangeHash>, green: bool, only: Option<i64>) {
    let Some(uname) = RepoName::new(&upstream.owner, &upstream.name) else { return };
    let Ok(id) = app.repos.read(&uname, |r, _| Ok(r.id()?)).await else { return };
    let rows: Vec<(i64, String, String)> = sqlx::query_as("SELECT consumer_id, bookmark, path FROM diffusion_pins WHERE upstream = $1")
        .bind(hex::encode(id))
        .fetch_all(&app.db)
        .await
        .unwrap_or_default();
    let unverified = unverified(app, upstream).await;
    for (consumer_id, indexed, path) in rows {
        if only.is_some_and(|o| o != consumer_id) {
            continue;
        }
        let Ok(Some(consumer)) = db::repos::get_by_id(&app.db, consumer_id).await else { continue };
        let Some(cname) = RepoName::new(&consumer.owner, &consumer.name) else { continue };
        let learned = target(&app.db, &consumer).await;
        let modules = oxymodules(app, &cname, &learned).await;
        let r = rule(&app.config.diffusion.follow, &consumer, modules.as_deref(), &path);
        let target = r.target.clone().unwrap_or(learned);
        if indexed != target {
            continue;
        }
        // An upstream with no checks never turns green: its moves count
        // for green followers too, marked amber (unverified).
        let go = match r.follow {
            Follow::Off => false,
            Follow::Green => green || unverified,
            Follow::Any => !green,
        };
        let why = if green {
            Why::Green
        } else if unverified {
            Why::Unverified
        } else {
            Why::Moved
        };
        if !go {
            continue;
        }
        // One bump or landing of this pin at a time across processes (#275).
        let Ok(_pin) = crate::leader::exclusive(&app.db, &format!("diffusion:{}:{path}", consumer.id)).await else { continue };
        match bump(app, &consumer, &target, &path, upstream, heads.clone()).await {
            Ok(Bumped::Moved { from, bookmark, heads: bumped, state, .. }) => {
                // Its pipeline runs as for a push, so the proposal is tested.
                tokio::spawn(start_runs(app.clone(), consumer.clone(), bookmark, state, bumped));
                let n = propose(app, &consumer, &target, &path, upstream, (from, heads.clone()), (why, r.watch.clone(), r.land_auto)).await;
                app.bus.live(&crate::events::topic::diffusion(consumer.id), "bump");
                app.bus.live(&crate::events::topic::diffusion(upstream.id), "bump");
                tracing::info!(
                    "diffusion: {}/{} {path} bumped to {}/{} (proposal {n:?})",
                    consumer.owner,
                    consumer.name,
                    upstream.owner,
                    upstream.name
                );
            }
            Ok(Bumped::Already { from, to }) => {
                // Already moved: its proposal's text and labels are brought
                // up to date all the same.
                propose(app, &consumer, &target, &path, upstream, (from, to), (why, r.watch.clone(), r.land_auto)).await;
                // The upstream may just have turned green under a bump its
                // consumer already passed.
                if green && r.land_auto {
                    tokio::spawn(try_land(app.clone(), consumer.clone(), path.clone()));
                }
            }
            Ok(other) => tracing::info!("diffusion: {}/{} {path}: no bump ({other:?})", consumer.owner, consumer.name),
            Err(e) => tracing::warn!("diffusion: bump {}/{} {path}: {e}", consumer.owner, consumer.name),
        }
    }
}

/// Why a bump was not landed by itself (docs/diffusion.md §4.2).
#[derive(Debug, PartialEq, Eq)]
pub enum Held {
    /// No open bump at that path.
    NoBump,
    /// The consumer did not choose `land = auto`.
    Manual,
    /// The upstream has no checks: amber never lands by itself.
    Unverified,
    /// The upstream state the bump pins has no passing run.
    UpstreamUnchecked,
    /// The consumer's run on the bump as it is now has not passed.
    NotPassed,
    /// The landing was refused (required checks, approvals, the trust policy).
    Refused(String),
}

/// Land the open bump at `path` in `consumer` itself, as the `diffusion`
/// account, when the consumer chose `land = auto` (its `.oxymodules` entry,
/// else Settings › Diffusion) and both halves of its pill are green: the
/// upstream state it pins passed its own run (never amber: an upstream
/// with no checks always waits for a person), and the consumer's run on
/// the bump as it is now passed. It lands through the web's own path
/// (`land_as`), signed by the hub's key, so required checks, approvals and
/// the trust policy decide as for anyone; a refusal is noted once on the
/// proposal's timeline. The proposal's number when it landed.
pub async fn auto_land(app: &Shared, consumer: &RepoRow, path: &str) -> Result<i64, Held> {
    let path = path.trim_matches('/');
    let bm = bump_bookmark(path);
    let Some(cname) = RepoName::new(&consumer.owner, &consumer.name) else { return Err(Held::NoBump) };
    let bump = db::proposals::list(&app.db, consumer.id, &["open"])
        .await
        .ok()
        .and_then(|ps| ps.into_iter().find(|p| p.source_bookmark == bm && p.author == ACCOUNT))
        .ok_or(Held::NoBump)?;
    let learned = target(&app.db, consumer).await;
    let modules = oxymodules(app, &cname, &learned).await;
    if !rule(&app.config.diffusion.follow, consumer, modules.as_deref(), path).land_auto {
        return Err(Held::Manual);
    }
    let upstream_id: Option<i64> = sqlx::query_scalar(
        "SELECT upstream_repo_id FROM diffusion_pins WHERE consumer_id = $1 AND path = $2 AND upstream_repo_id IS NOT NULL LIMIT 1",
    )
    .bind(consumer.id)
    .bind(path)
    .fetch_optional(&app.db)
    .await
    .ok()
    .flatten();
    let Some(upstream) = (match upstream_id {
        Some(id) => db::repos::get_by_id(&app.db, id).await.ok().flatten(),
        None => None,
    }) else {
        return Err(Held::UpstreamUnchecked);
    };
    if unverified(app, &upstream).await {
        return Err(Held::Unverified);
    }
    // What the bump pins, and the bump bookmark's heads now.
    let (p, b) = (path.to_string(), bm.clone());
    let (pinned, now) = app
        .repos
        .read(&cname, move |r, cache| {
            let heads = heads_of(r, &b)?;
            if heads.is_empty() {
                return Ok((None, heads));
            }
            let loaded = cache.load(r, &heads)?;
            let pin = match loaded.snapshot.files.get(&p).map(|f| &f.content) {
                Some(oxy_core::state::FileContent::Opaque { blobs }) => match blobs.first() {
                    Some(blob) => oxy_core::subrepo::Pin::decode(&r.read_blob(blob)?).ok(),
                    None => None,
                },
                _ => None,
            };
            Ok((pin, heads))
        })
        .await
        .map_err(|e| Held::Refused(e.to_string()))?;
    let Some(oxy_core::subrepo::Pin::Oxy { heads: pinned, .. }) = pinned else { return Err(Held::UpstreamUnchecked) };
    let same = |text: &str, want: &[ChangeHash]| {
        let mut a: Vec<String> = text.split('+').filter(|h| !h.is_empty()).map(str::to_string).collect();
        let mut b: Vec<String> = want.iter().map(|h| h.to_hex()).collect();
        a.sort();
        b.sort();
        a == b
    };
    let upstream_runs: Vec<String> = sqlx::query_scalar(
        "SELECT heads FROM flow_runs WHERE repo_id = $1 AND bookmark = $2 AND status = 'passed' ORDER BY id DESC LIMIT 100",
    )
    .bind(upstream.id)
    .bind(&upstream.default_bookmark)
    .fetch_all(&app.db)
    .await
    .unwrap_or_default();
    if !upstream_runs.iter().any(|h| same(h, &pinned)) {
        return Err(Held::UpstreamUnchecked);
    }
    let run: Option<(String, String)> =
        sqlx::query_as("SELECT status, heads FROM flow_runs WHERE repo_id = $1 AND bookmark = $2 ORDER BY id DESC LIMIT 1")
            .bind(consumer.id)
            .bind(&bm)
            .fetch_optional(&app.db)
            .await
            .ok()
            .flatten();
    if !run.is_some_and(|(status, heads)| status == "passed" && same(&heads, &now)) {
        return Err(Held::NotPassed);
    }
    let me = account(&app.db).await.ok_or_else(|| Held::Refused("the diffusion account is missing".into()))?;
    let via = "Diffusion, both halves green";
    match crate::web::proposals::land_as(app, &cname, consumer, &me, bump.number, via, true, None).await {
        Ok(_) => {
            tracing::info!("diffusion: {}/{} landed #{} itself (both halves green)", consumer.owner, consumer.name, bump.number);
            Ok(bump.number)
        }
        Err(crate::web::proposals::Landing::NotFound) => Err(Held::NoBump),
        Err(crate::web::proposals::Landing::Refused(why)) => {
            // Said once, so the timeline explains why a green bump waits.
            let said = db::proposals::events(&app.db, bump.id)
                .await
                .unwrap_or_default()
                .iter()
                .any(|e| e.event == "not landed by itself" && e.detail == why);
            if !said {
                db::proposals::add_event(&app.db, bump.id, ACCOUNT, "not landed by itself", &why).await.ok();
            }
            Err(Held::Refused(why))
        }
    }
}

/// [`auto_land`] in the background, logged. Boxed as `Send`: a landing
/// moves the target as a push would, which may bump and land again.
pub fn try_land(app: Shared, consumer: RepoRow, path: String) -> std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send>> {
    Box::pin(async move {
        // One landing of this pin at a time across processes (#275).
        let Ok(_pin) = crate::leader::exclusive(&app.db, &format!("diffusion:{}:{path}", consumer.id)).await else { return };
        match auto_land(&app, &consumer, &path).await {
            Ok(_) | Err(Held::NoBump | Held::Manual) => {}
            Err(why) => tracing::info!("diffusion: {}/{} {path} waits for a person: {why:?}", consumer.owner, consumer.name),
        }
    })
}

/// Try every open bump of `consumer` (its setting turned to `auto`).
pub async fn land_ready(app: &Shared, consumer: &RepoRow) {
    let paths: Vec<String> = sqlx::query_scalar("SELECT DISTINCT path FROM diffusion_pins WHERE consumer_id = $1")
        .bind(consumer.id)
        .fetch_all(&app.db)
        .await
        .unwrap_or_default();
    for path in paths {
        tokio::spawn(try_land(app.clone(), consumer.clone(), path));
    }
}

/// Whether `upstream` has no checks of its own: Flow is off for it, or
/// its default bookmark has never had a run. Its moves are then
/// "unverified" (amber), never green.
pub async fn unverified(app: &Shared, upstream: &RepoRow) -> bool {
    if !crate::flow::enabled(app, upstream) {
        return true;
    }
    let runs: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM flow_runs WHERE repo_id = $1 AND bookmark = $2")
        .bind(upstream.id)
        .bind(&upstream.default_bookmark)
        .fetch_one(&app.db)
        .await
        .unwrap_or(0);
    runs == 0
}

/// The repository on this hub with Oxygen id `id` (hex).
async fn repo_by_oxy_id(app: &Shared, id: &str) -> Option<RepoRow> {
    for row in db::repos::list(&app.db).await.ok()? {
        let Some(name) = RepoName::new(&row.owner, &row.name) else { continue };
        if app.repos.read(&name, |r, _| Ok(r.id()?)).await.ok().map(hex::encode).as_deref() == Some(id) {
            return Some(row);
        }
    }
    None
}

/// The newest state of `upstream`'s default bookmark whose run passed.
async fn last_green(app: &Shared, upstream: &RepoRow) -> Option<Vec<ChangeHash>> {
    let heads: String = sqlx::query_scalar(
        "SELECT heads FROM flow_runs WHERE repo_id = $1 AND bookmark = $2 AND status = 'passed' ORDER BY id DESC LIMIT 1",
    )
    .bind(upstream.id)
    .bind(&upstream.default_bookmark)
    .fetch_optional(&app.db)
    .await
    .ok()
    .flatten()?;
    let h: Vec<ChangeHash> = heads.split('+').filter_map(ChangeHash::from_hex).collect();
    (!h.is_empty()).then_some(h)
}

/// Bring `consumer` up to date with the upstreams it pins, at once: when
/// it turns Diffusion on, it is bumped to each upstream's newest state
/// that passed (or simply its newest, when it follows every move).
pub async fn catch_up(app: &Shared, consumer: &RepoRow) {
    let upstreams: Vec<String> = sqlx::query_scalar("SELECT DISTINCT upstream FROM diffusion_pins WHERE consumer_id = $1")
        .bind(consumer.id)
        .fetch_all(&app.db)
        .await
        .unwrap_or_default();
    for id in upstreams {
        let Some(upstream) = repo_by_oxy_id(app, &id).await else { continue };
        if let Some(h) = last_green(app, &upstream).await {
            propagate_to(app, &upstream, h, true, Some(consumer.id)).await;
        }
        let Some(name) = RepoName::new(&upstream.owner, &upstream.name) else { continue };
        let b = upstream.default_bookmark.clone();
        if let Ok(h) = app.repos.read(&name, move |r, _| heads_of(r, &b)).await {
            propagate_to(app, &upstream, h, false, Some(consumer.id)).await;
        }
    }
}

/// When a consumer takes an upstream's moves (§4.2).
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
#[serde(rename_all = "lowercase")]
pub enum Follow {
    Off,
    /// When the upstream state's checks pass.
    Green,
    /// On every move.
    Any,
}

impl Follow {
    pub fn parse(s: &str) -> Option<Follow> {
        match s.trim() {
            "off" => Some(Follow::Off),
            "green" => Some(Follow::Green),
            "any" => Some(Follow::Any),
            _ => None,
        }
    }

    pub fn as_str(self) -> &'static str {
        match self {
            Follow::Off => "off",
            Follow::Green => "green",
            Follow::Any => "any",
        }
    }
}

/// How one consumer follows one subrepo, from its `.oxymodules` entry, its
/// repository setting and the site's default, in that order.
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
pub struct Rule {
    pub follow: Follow,
    /// Land a bump itself once both halves are green (`land = auto`, or
    /// the repository's setting); otherwise it only proposes.
    pub land_auto: bool,
    /// The bookmark to bump, when `.oxymodules` names one (else learned).
    pub target: Option<String>,
    /// Paths the consumer watches: upstream changes touching them are
    /// flagged in a bump's changelog.
    pub watch: Vec<String>,
    /// Where `follow` came from: `.oxymodules`, `setting` or `site`.
    pub from: &'static str,
}

/// The Diffusion keys of `path`'s entry in `.oxymodules` text: `follow`,
/// `land`, `target` and `watch` (comma-separated, or a TOML list).
pub fn module_keys(text: &str, path: &str) -> (Option<Follow>, Option<bool>, Option<String>, Vec<String>) {
    let (mut follow, mut land, mut target, mut watch) = (None, None, None, Vec::new());
    let mut here = false;
    let mut lines = text.lines();
    while let Some(line) = lines.next() {
        let line = line.trim();
        if line.is_empty() || line.starts_with('#') {
            continue;
        }
        if let Some(inner) = line.strip_prefix('[').and_then(|l| l.strip_suffix(']')) {
            here = inner.trim().trim_matches('"').trim_end_matches('/') == path.trim_end_matches('/');
            continue;
        }
        let (true, Some((k, v))) = (here, line.split_once('=')) else { continue };
        // A list may run over several lines (oxygenhub#200), as TOML allows:
        // read on to its closing `]`.
        let mut v = v.trim().to_string();
        if v.starts_with('[') && !v.contains(']') {
            for more in lines.by_ref() {
                let more = more.trim();
                if more.starts_with('#') {
                    continue;
                }
                v.push_str(more);
                if more.contains(']') {
                    break;
                }
            }
        }
        let v = v.as_str();
        let unquoted = v.trim_matches('"').trim();
        match k.trim() {
            "follow" => follow = Follow::parse(unquoted),
            "land" => land = Some(unquoted == "auto"),
            "target" if !unquoted.is_empty() => target = Some(unquoted.to_string()),
            "watch" => {
                watch = v
                    .trim_start_matches('[')
                    .trim_end_matches(']')
                    .split(',')
                    .map(|w| w.trim().trim_matches('"').trim().to_string())
                    .filter(|w| !w.is_empty())
                    .collect()
            }
            _ => {}
        }
    }
    (follow, land, target, watch)
}

/// How `row` follows its subrepo at `path`, given its `.oxymodules` text.
pub fn rule(site_default: &str, row: &RepoRow, oxymodules: Option<&str>, path: &str) -> Rule {
    let (follow, land, target, watch) = oxymodules.map(|t| module_keys(t, path)).unwrap_or_default();
    let (follow, from) = match follow {
        Some(f) => (f, ".oxymodules"),
        None => match Follow::parse(&row.diffusion) {
            Some(f) => (f, "setting"),
            None => (Follow::parse(site_default).unwrap_or(Follow::Off), "site"),
        },
    };
    // Landing by itself: the entry's word, else the setting; never by default.
    let land_auto = land.unwrap_or(row.diffusion_land == "auto");
    Rule { follow, land_auto, target, watch, from }
}

#[cfg(test)]
mod tests {
    use super::*;

    const MODULES: &str = r#"
[rstar-api]
url = https://oxygenhub.io/doomsdayone/rstar-api
follow = green
land = "propose"
target = develop
watch = ["src/x16/irq*", "src/x16/frame*"]

[other]
url = https://example/other
follow = any
"#;

    fn row(diffusion: &str) -> RepoRow {
        RepoRow {
            id: 1,
            owner: "o".into(),
            name: "n".into(),
            description: String::new(),
            default_bookmark: "main".into(),
            visibility: "public".into(),
            created_ms: 0,
            updated_ms: 0,
            git_mirror: false,
            git_mirror_state: String::new(),
            forked_from: None,
            topics: String::new(),
            flow: String::new(),
            diffusion: diffusion.into(),
            ..Default::default()
        }
    }

    #[test]
    fn landing_by_itself_is_chosen_per_repository_and_per_entry() {
        let mut r = row("green");
        assert!(!rule("off", &r, None, "rstar-api").land_auto, "never by default");
        r.diffusion_land = "auto".into();
        assert!(rule("off", &r, None, "rstar-api").land_auto, "the setting");
        assert!(!rule("off", &r, Some(MODULES), "rstar-api").land_auto, "the entry's `land = propose` wins");
        assert!(rule("off", &r, Some(MODULES), "other").land_auto, "an entry without `land` takes the setting");
        r.diffusion_land = String::new();
        let auto = MODULES.replace(r#"land = "propose""#, "land = auto");
        assert!(rule("off", &r, Some(&auto), "rstar-api").land_auto);
    }

    #[test]
    fn watch_patterns_match_paths() {
        assert!(watched("src/x16/irq*", "src/x16/irq.s"));
        assert!(watched("src/x16/irq*", "src/x16/irq_wait.c"));
        assert!(!watched("src/x16/irq*", "src/x16/sub/irq.s"), "one segment");
        assert!(watched("src/**/frame*.c", "src/x16/floor/frame_wait.c"));
        assert!(watched("src/x16", "src/x16/bank.s"), "a directory");
        assert!(!watched("src/x16", "src/x160/bank.s"));
        assert!(watched("include/r?m.h", "include/rim.h"));
        assert!(!watched("", "anything"));
    }

    #[test]
    fn oxymodules_keys_are_read_per_entry() {
        let (f, land, target, watch) = module_keys(MODULES, "rstar-api");
        assert_eq!((f, land, target.as_deref()), (Some(Follow::Green), Some(false), Some("develop")));
        assert_eq!(watch, vec!["src/x16/irq*", "src/x16/frame*"]);
        assert_eq!(module_keys(MODULES, "other").0, Some(Follow::Any));
        assert_eq!(module_keys(MODULES, "missing"), (None, None, None, vec![]));
        assert_eq!(module_keys("[a]\nwatch = x, y\n", "a").3, vec!["x", "y"]);
        // #200: a list over several lines, as in trailblazers, comments
        // and the keys after it included.
        let multi = "[rstar-api]\nurl = https://hub/x\n\nwatch = [\"src/x16/rbm*\",\n         \"src/x16/rdm.c\",\n  # a note\n         \"src/mega65/rsm.c\"]\nfollow = \"green\"\n";
        let (f, _, _, watch) = module_keys(multi, "rstar-api");
        assert_eq!(watch, vec!["src/x16/rbm*", "src/x16/rdm.c", "src/mega65/rsm.c"]);
        assert_eq!(f, Some(Follow::Green), "the key after the list is still read");
    }

    #[test]
    fn oxymodules_wins_then_the_setting_then_the_site() {
        let r = rule("off", &row("any"), Some(MODULES), "rstar-api");
        assert_eq!((r.follow, r.from), (Follow::Green, ".oxymodules"));
        let r = rule("off", &row("any"), Some("[rstar-api]\nurl = x\n"), "rstar-api");
        assert_eq!((r.follow, r.from), (Follow::Any, "setting"));
        let r = rule("green", &row(""), None, "rstar-api");
        assert_eq!((r.follow, r.from), (Follow::Green, "site"));
        assert_eq!(rule("nonsense", &row(""), None, "x").follow, Follow::Off);
    }
}

#[cfg(test)]
mod message_tests {
    use super::bump_message;

    #[test]
    fn a_bump_of_many_heads_has_a_short_title_and_the_whole_address_below() {
        let heads: Vec<String> = (0..20).map(|i| format!("{i:02x}a6c9d4746a")).collect();
        let m = bump_message("rstar-api", "doomsdayone/rstar-api", &heads);
        let (title, body) = m.split_once("\n\n").unwrap();
        assert_eq!(title, "Diffusion: move rstar-api to doomsdayone/rstar-api 00a6c9d4746a (+19 heads)");
        assert_eq!(body, format!("Pins doomsdayone/rstar-api at {}", heads.join("+")));
        assert_eq!(bump_message("api", "andy/api", &heads[..1]), "Diffusion: move api to andy/api 00a6c9d4746a");
    }
}
`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet
Keyboard
