//! 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 = ? AND state = 'landed' AND landed_ms >= ?
         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]) {
    let Some(name) = RepoName::new(&row.owner, &row.name) else { return };
    let pins = if heads.is_empty() { Vec::new() } else { crate::flow::pins_in(app, name, heads.to_vec()).await };
    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));
    }
    let Ok(mut tx) = app.db.begin().await else { return };
    let mut ok = sqlx::query("DELETE FROM diffusion_pins WHERE consumer_id = ? AND bookmark = ?")
        .bind(row.id)
        .bind(bookmark)
        .execute(&mut *tx)
        .await
        .is_ok();
    for ((path, id, pinned), upstream_row) in pins.iter().zip(&upstreams) {
        let heads: Vec<String> = pinned.iter().map(|h| h.to_hex()).collect();
        ok &= sqlx::query(
            "INSERT INTO diffusion_pins (consumer_id, bookmark, path, upstream, heads, updated_ms, upstream_repo_id) VALUES (?, ?, ?, ?, ?, ?, ?)",
        )
        .bind(row.id)
        .bind(bookmark)
        .bind(path)
        .bind(hex::encode(id))
        .bind(heads.join("+"))
        .bind(now)
        .bind(upstream_row)
        .execute(&mut *tx)
        .await
        .is_ok();
        if let Some(u) = upstream_row {
            mark(app, row.id);
            mark(app, *u);
        }
    }
    if ok {
        tx.commit().await.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;
            }
        }
    }
}

/// 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,
}

/// 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 = ? 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 = ? AND bookmark = ? 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,
        });
    }
    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?;
    for (c, b) in out.iter_mut().zip(behind) {
        c.behind = b;
        c.colour = Colour::of(c.follow, b, c.bump, c.run.as_ref().map(|r| r.1.as_str()), unverified);
    }
    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 = ? 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.
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 = format!("Diffusion: move {p} to {upstream_label} {}", short.join("+"));
    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): (Why, Vec<String>),
) -> 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). It never lands by itself.",
        short.join("+"),
        lines.join("\n")
    );
    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", &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();
            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 = ?")
        .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;
        }
        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())).await;
                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())).await;
            }
            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),
        }
    }
}

/// 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 = ? AND bookmark = ?")
        .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 = ? AND bookmark = ? 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 = ?")
        .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,
        }
    }
}

/// 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 green bump itself (D5); D1 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;
    for line in text.lines() {
        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 };
        let v = v.trim();
        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"),
        },
    };
    Rule { follow, land_auto: land.unwrap_or(false), 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 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"]);
    }

    #[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);
    }
}