OXYGENHUB
# 62cdf34daec1 oxygenhub / crates / hub / src / diffusion.rs
64.5 KiB · Rust · last changed Long hashes and state addresses in change titles are shown by their ends 20 days ago source blame history raw
1 
//! Diffusion (#148, docs/diffusion.md): an upstream's changes reach the
2 
//! repositories that pin it as a subrepo.
3 
//!
4 
//! D1 starts with the reverse pin index (§4.1): when a push moves a
5 
//! repository's default bookmark, or the bookmark its work lands on, each
6 
//! Oxygen subrepo pin in the new state is recorded. That answers "who
7 
//! consumes this repository, at which state, and how far behind", which
8 
//! an upstream's page shows as Used by.
9
10 
use crate::auth::roles;
11 
use crate::db::{self, accounts::Account, repos::RepoRow, Pool};
12 
use crate::repos::RepoName;
13 
use crate::Shared;
14 
use oxy_core::hash::ChangeHash;
15 
use serde::Serialize;
16
17 
/// How far back landed proposals teach where a repository's work lands.
18 
pub const LEARN_DAYS: i64 = 90;
19
20 
/// The bookmark a repository's work lands on (§4.3): the most common
21 
/// target of its proposals landed in the last `LEARN_DAYS`, else its
22 
/// default bookmark.
23 
pub async fn target(pool: &Pool, row: &RepoRow) -> String {
24 
    let since = db::now_ms() - LEARN_DAYS * 24 * 3600 * 1000;
25 
    let learned: Option<String> = sqlx::query_scalar(
26 
        "SELECT target FROM proposals WHERE repo_id = ? AND state = 'landed' AND landed_ms >= ?
27 
         GROUP BY target ORDER BY COUNT(*) DESC, MAX(landed_ms) DESC LIMIT 1",
28 
    )
29 
    .bind(row.id)
30 
    .bind(since)
31 
    .fetch_optional(pool)
32 
    .await
33 
    .ok()
34 
    .flatten();
35 
    learned.filter(|t| !t.is_empty()).unwrap_or_else(|| row.default_bookmark.clone())
36 
}
37
38 
/// The bookmarks whose pins are indexed: the default and the learned
39 
/// target (one when they are the same).
40 
pub async fn followed_bookmarks(pool: &Pool, row: &RepoRow) -> Vec<String> {
41 
    let mut out = vec![row.default_bookmark.clone()];
42 
    let t = target(pool, row).await;
43 
    if !out.contains(&t) {
44 
        out.push(t);
45 
    }
46 
    out
47 
}
48
49 
/// Record `row`'s pins on `bookmark` at `heads`, replacing what was there.
50 
pub async fn record(app: &Shared, row: &RepoRow, bookmark: &str, heads: &[ChangeHash]) {
51 
    // A failure is logged and tried once more after a moment (oxygenhub#210):
52 
    // an index left behind shows a consumer behind with no bump.
53 
    let repo = format!("{}/{}", row.owner, row.name);
54 
    for attempt in 1..=2 {
55 
        match try_record(app, row, bookmark, heads).await {
56 
            Ok(()) => {
57 
                if attempt > 1 {
58 
                    tracing::info!("diffusion: indexed {repo} {bookmark} on the second try");
59 
                }
60 
                return;
61 
            }
62 
            Err(why) if attempt == 1 => {
63 
                tracing::warn!("diffusion: indexing {repo} {bookmark} failed ({why}); trying again");
64 
                tokio::time::sleep(RETRY_AFTER).await;
65 
            }
66 
            Err(why) => {
67 
                FAILURES.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
68 
                tracing::error!("diffusion: indexing {repo} {bookmark} failed again ({why}); its pins stay as they were");
69 
            }
70 
        }
71 
    }
72 
}
73
74 
/// How long [`record`] waits before its second try.
75 
const RETRY_AFTER: std::time::Duration = std::time::Duration::from_millis(250);
76
77 
/// Indexing failures that the retry did not mend, since the hub started.
78 
pub static FAILURES: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
79
80 
/// Failures to force, per repository (tests only: [`fail_next_records`]).
81 
static FORCED: std::sync::Mutex<Option<std::collections::HashMap<String, u32>>> = std::sync::Mutex::new(None);
82
83 
/// Make the next `n` index writes for a repository fail, as a busy
84 
/// database would (for tests of the retry).
85 
#[doc(hidden)]
86 
pub fn fail_next_records(app: &Shared, repo_id: i64, n: u32) {
87 
    FORCED.lock().unwrap_or_else(|p| p.into_inner()).get_or_insert_with(Default::default).insert(key(app, repo_id), n);
88 
}
89
90 
/// Forced failures not yet used (tests only).
91 
#[doc(hidden)]
92 
pub fn forced_left(app: &Shared, repo_id: i64) -> u32 {
93 
    FORCED.lock().unwrap_or_else(|p| p.into_inner()).as_ref().and_then(|m| m.get(&key(app, repo_id)).copied()).unwrap_or(0)
94 
}
95
96 
fn forced(app: &Shared, repo_id: i64) -> bool {
97 
    let mut forced = FORCED.lock().unwrap_or_else(|p| p.into_inner());
98 
    match forced.as_mut().and_then(|m| m.get_mut(&key(app, repo_id))) {
99 
        Some(n) if *n > 0 => {
100 
            *n -= 1;
101 
            true
102 
        }
103 
        _ => false,
104 
    }
105 
}
106
107 
/// One try at replacing a bookmark's indexed pins; which step failed and
108 
/// why. A state that cannot be read leaves the index as it was rather than
109 
/// emptying it.
110 
async fn try_record(app: &Shared, row: &RepoRow, bookmark: &str, heads: &[ChangeHash]) -> Result<(), String> {
111 
    let Some(name) = RepoName::new(&row.owner, &row.name) else { return Ok(()) };
112 
    let pins = if heads.is_empty() {
113 
        Vec::new()
114 
    } else {
115 
        crate::flow::pins_in(app, name, heads.to_vec()).await.map_err(|e| format!("reading the pins: {e}"))?
116 
    };
117 
    let now = db::now_ms();
118 
    // The upstream's repository on this hub, when it is one.
119 
    let mut upstreams = Vec::new();
120 
    for (_, id, _) in &pins {
121 
        upstreams.push(repo_by_oxy_id(app, &hex::encode(id)).await.map(|r| r.id));
122 
    }
123 
    if forced(app, row.id) {
124 
        return Err("begin: forced for a test".into());
125 
    }
126 
    let mut tx = app.db.begin().await.map_err(|e| format!("begin: {e}"))?;
127 
    sqlx::query("DELETE FROM diffusion_pins WHERE consumer_id = ? AND bookmark = ?")
128 
        .bind(row.id)
129 
        .bind(bookmark)
130 
        .execute(&mut *tx)
131 
        .await
132 
        .map_err(|e| format!("delete: {e}"))?;
133 
    for ((path, id, pinned), upstream_row) in pins.iter().zip(&upstreams) {
134 
        let heads: Vec<String> = pinned.iter().map(|h| h.to_hex()).collect();
135 
        sqlx::query(
136 
            "INSERT INTO diffusion_pins (consumer_id, bookmark, path, upstream, heads, updated_ms, upstream_repo_id) VALUES (?, ?, ?, ?, ?, ?, ?)",
137 
        )
138 
        .bind(row.id)
139 
        .bind(bookmark)
140 
        .bind(path)
141 
        .bind(hex::encode(id))
142 
        .bind(heads.join("+"))
143 
        .bind(now)
144 
        .bind(upstream_row)
145 
        .execute(&mut *tx)
146 
        .await
147 
        .map_err(|e| format!("insert {path}: {e}"))?;
148 
    }
149 
    sqlx::query("INSERT OR REPLACE INTO diffusion_indexed (consumer_id, bookmark, heads, updated_ms) VALUES (?, ?, ?, ?)")
150 
        .bind(row.id)
151 
        .bind(bookmark)
152 
        .bind(heads_key(heads))
153 
        .bind(now)
154 
        .execute(&mut *tx)
155 
        .await
156 
        .map_err(|e| format!("noting the state: {e}"))?;
157 
    tx.commit().await.map_err(|e| format!("commit: {e}"))?;
158 
    for u in upstreams.iter().flatten() {
159 
        mark(app, row.id);
160 
        mark(app, *u);
161 
    }
162 
    Ok(())
163 
}
164
165 
/// After a push: index the pins of each moved bookmark that is followed.
166 
pub async fn after_push(app: &Shared, row: &RepoRow, moved: &[(String, String, Vec<ChangeHash>)]) {
167 
    let followed = followed_bookmarks(&app.db, row).await;
168 
    for (bookmark, _, heads) in moved {
169 
        if followed.contains(bookmark) {
170 
            record(app, row, bookmark, heads).await;
171 
        }
172 
    }
173 
}
174
175 
static INVOLVED: std::sync::Mutex<Option<std::collections::HashSet<String>>> = std::sync::Mutex::new(None);
176
177 
fn key(app: &Shared, repo_id: i64) -> String {
178 
    format!("{}#{repo_id}", app.repos.root().display())
179 
}
180
181 
fn mark(app: &Shared, repo_id: i64) {
182 
    INVOLVED.lock().unwrap_or_else(|p| p.into_inner()).get_or_insert_with(Default::default).insert(key(app, repo_id));
183 
}
184
185 
/// Whether a repository takes part in Diffusion, as an upstream or a
186 
/// consumer on this hub: its Diffusion tab shows (#150).
187 
pub fn involved(app: &Shared, repo_id: i64) -> bool {
188 
    INVOLVED.lock().unwrap_or_else(|p| p.into_inner()).as_ref().is_some_and(|s| s.contains(&key(app, repo_id)))
189 
}
190
191 
/// Index every repository's followed bookmarks, for a hub that had no
192 
/// index before (run once at start, in the background).
193 
pub async fn backfill(app: &Shared) {
194 
    let Ok(rows) = db::repos::list(&app.db).await else { return };
195 
    for row in rows {
196 
        let Some(name) = RepoName::new(&row.owner, &row.name) else { continue };
197 
        for bookmark in followed_bookmarks(&app.db, &row).await {
198 
            let b = bookmark.clone();
199 
            let heads = app.repos.read(&name, move |r, _| heads_of(r, &b)).await;
200 
            if let Ok(heads) = heads {
201 
                record(app, &row, &bookmark, &heads).await;
202 
            }
203 
        }
204 
    }
205 
}
206
207 
/// Heads as one comparable string: sorted, joined with `+`.
208 
fn heads_key(heads: &[ChangeHash]) -> String {
209 
    let mut h: Vec<String> = heads.iter().map(|h| h.to_hex()).collect();
210 
    h.sort();
211 
    h.join("+")
212 
}
213
214 
/// How often [`reconciler`] looks.
215 
pub const RECONCILE_EVERY: std::time::Duration = std::time::Duration::from_secs(10 * 60);
216
217 
/// What one reconciling pass mended.
218 
#[derive(Debug, Default, PartialEq, Eq)]
219 
pub struct Repairs {
220 
    /// Followed bookmarks whose index was behind the bookmark.
221 
    pub reindexed: usize,
222 
    /// Consumers behind an upstream's newest passing state with no open
223 
    /// bump, bumped now.
224 
    pub caught_up: usize,
225 
}
226
227 
/// Every [`RECONCILE_EVERY`], mend the pin index and the bumps it drives
228 
/// (oxygenhub#210).
229 
pub async fn reconciler(app: Shared) {
230 
    loop {
231 
        tokio::time::sleep(RECONCILE_EVERY).await;
232 
        let r = reconcile(&app).await;
233 
        if r != Repairs::default() {
234 
            tracing::warn!("diffusion: reconciled: {} bookmarks re-indexed, {} consumers caught up", r.reindexed, r.caught_up);
235 
        }
236 
    }
237 
}
238
239 
/// One pass: re-index each followed bookmark whose indexed state is not
240 
/// the state it is at now; then bump each consumer whose pin is not its
241 
/// upstream's newest passing state and that has no open bump for it.
242 
pub async fn reconcile(app: &Shared) -> Repairs {
243 
    let mut repairs = Repairs::default();
244 
    let Ok(rows) = db::repos::list(&app.db).await else { return repairs };
245 
    for row in &rows {
246 
        let Some(name) = RepoName::new(&row.owner, &row.name) else { continue };
247 
        for bookmark in followed_bookmarks(&app.db, row).await {
248 
            let b = bookmark.clone();
249 
            let Ok(heads) = app.repos.read(&name, move |r, _| heads_of(r, &b)).await else { continue };
250 
            let indexed: Option<String> = sqlx::query_scalar("SELECT heads FROM diffusion_indexed WHERE consumer_id = ? AND bookmark = ?")
251 
                .bind(row.id)
252 
                .bind(&bookmark)
253 
                .fetch_optional(&app.db)
254 
                .await
255 
                .ok()
256 
                .flatten();
257 
            if indexed.as_deref() != Some(heads_key(&heads).as_str()) {
258 
                tracing::info!("diffusion: {}/{} {bookmark}: the index was not at the bookmark; re-indexing", row.owner, row.name);
259 
                record(app, row, &bookmark, &heads).await;
260 
                repairs.reindexed += 1;
261 
            }
262 
        }
263 
    }
264 
    // Consumers behind with no open bump.
265 
    let pins: Vec<(i64, String, String, String)> =
266 
        sqlx::query_as("SELECT consumer_id, path, upstream, heads FROM diffusion_pins WHERE upstream_repo_id IS NOT NULL")
267 
            .fetch_all(&app.db)
268 
            .await
269 
            .unwrap_or_default();
270 
    let mut caught = std::collections::HashSet::new();
271 
    for (consumer_id, path, upstream, pinned) in pins {
272 
        if caught.contains(&consumer_id) {
273 
            continue;
274 
        }
275 
        let Some(up) = repo_by_oxy_id(app, &upstream).await else { continue };
276 
        let Some(green) = last_green(app, &up).await else { continue };
277 
        if heads_key(&green) == heads_key(&pinned.split('+').filter_map(ChangeHash::from_hex).collect::<Vec<_>>()) {
278 
            continue;
279 
        }
280 
        let Ok(Some(consumer)) = db::repos::get_by_id(&app.db, consumer_id).await else { continue };
281 
        let Some(cname) = RepoName::new(&consumer.owner, &consumer.name) else { continue };
282 
        let learned = target(&app.db, &consumer).await;
283 
        let modules = oxymodules(app, &cname, &learned).await;
284 
        if rule(&app.config.diffusion.follow, &consumer, modules.as_deref(), &path).follow == Follow::Off {
285 
            continue;
286 
        }
287 
        // The newest bump at this path: open or a draft is a bump already,
288 
        // and one a person closed is a "no" the reconciler respects (the
289 
        // upstream's next move proposes again as usual).
290 
        let bump_bm = bump_bookmark(&path);
291 
        let newest = || async {
292 
            db::proposals::list(&app.db, consumer_id, &[])
293 
                .await
294 
                .unwrap_or_default()
295 
                .into_iter()
296 
                .filter(|p| p.source_bookmark == bump_bm)
297 
                .max_by_key(|p| p.number)
298 
                .map(|p| p.state)
299 
        };
300 
        let has_bump = || async { matches!(newest().await.as_deref(), Some("open" | "draft")) };
301 
        if matches!(newest().await.as_deref(), Some("open" | "draft" | "closed")) {
302 
            continue;
303 
        }
304 
        caught.insert(consumer_id);
305 
        catch_up(app, &consumer).await;
306 
        if has_bump().await {
307 
            tracing::info!(
308 
                "diffusion: {}/{} {path} was behind {}/{} with no bump; bumped",
309 
                consumer.owner,
310 
                consumer.name,
311 
                up.owner,
312 
                up.name
313 
            );
314 
            repairs.caught_up += 1;
315 
        }
316 
    }
317 
    repairs
318 
}
319
320 
/// A bookmark's heads as the repository has them now (not a cached
321 
/// list, which a push refreshes a moment later); empty when it has none.
322 
fn heads_of(r: &oxy_core::repo::Repo, bookmark: &str) -> Result<Vec<ChangeHash>, crate::repos::RepoError> {
323 
    Ok(r.bookmarks()?.into_iter().find(|b| b.name == bookmark).map(|b| b.heads).unwrap_or_default())
324 
}
325
326 
/// One consumer of an upstream, as an upstream's page and API show it.
327 
#[derive(Clone, Debug, Serialize)]
328 
pub struct Consumer {
329 
    pub owner: String,
330 
    pub name: String,
331 
    pub bookmark: String,
332 
    pub path: String,
333 
    /// The upstream heads it pins, hex.
334 
    pub pinned: Vec<String>,
335 
    /// Upstream changes on its default bookmark the pin lacks; `None` when
336 
    /// the hub cannot tell (the pinned changes are not in the upstream).
337 
    pub behind: Option<usize>,
338 
    /// How it follows the upstream (from its setting, or the site's).
339 
    pub follow: Follow,
340 
    /// Its open bump proposal's number, if it has one.
341 
    pub bump: Option<i64>,
342 
    /// The newest run of its bump bookmark: number and status.
343 
    pub run: Option<(i64, String)>,
344 
    /// Its last landed bump: number and when.
345 
    pub landed: Option<(i64, i64)>,
346 
    /// What the views show (#150).
347 
    pub colour: Colour,
348 
    /// The pill's two halves: the upstream, then this consumer.
349 
    pub upstream: Half,
350 
    pub downstream: Half,
351 
}
352
353 
impl Consumer {
354 
    /// As the API gives it (oxygenhub#153), with its upstream's name.
355 
    pub fn api(&self, upstream: &str) -> hub_api::DiffusionPin {
356 
        let half = |h: &Half| hub_api::DiffusionHalf {
357 
            tone: h.class.trim_start_matches("df-").to_string(),
358 
            word: h.word.to_string(),
359 
            title: h.title.clone(),
360 
        };
361 
        hub_api::DiffusionPin {
362 
            upstream: upstream.to_string(),
363 
            consumer: format!("{}/{}", self.owner, self.name),
364 
            path: self.path.clone(),
365 
            bookmark: self.bookmark.clone(),
366 
            behind: self.behind.map(|n| n as u64),
367 
            follow: self.follow.as_str().to_string(),
368 
            bump: self.bump,
369 
            run: self.run.as_ref().map(|r| r.0),
370 
            run_status: self.run.as_ref().map(|r| r.1.clone()),
371 
            upstream_state: half(&self.upstream),
372 
            consumer_state: half(&self.downstream),
373 
        }
374 
    }
375 
}
376
377 
/// One half of a Diffusion pill: a colour class and a word (#150).
378 
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
379 
pub struct Half {
380 
    pub class: &'static str,
381 
    pub word: &'static str,
382 
    pub title: String,
383 
}
384
385 
impl Half {
386 
    /// The upstream's half: was the upstream itself tested?
387 
    pub fn upstream(unverified: bool, run: Option<&str>, slug: &str) -> Half {
388 
        let (class, word, title) = match (unverified, run) {
389 
            (true, _) => ("df-amber", "untested", format!("{slug} has no checks of its own: nothing has tested it")),
390 
            (false, Some("passed")) => ("df-green", "checked", format!("{slug}'s newest run passed")),
391 
            (false, Some("failed" | "error" | "cancelled")) => ("df-red", "failing", format!("{slug}'s newest run failed")),
392 
            (false, _) => ("df-grey", "testing", format!("{slug}'s newest run has not finished")),
393 
        };
394 
        Half { class, word, title }
395 
    }
396
397 
    /// The consumer's half: does it pass with the upstream?
398 
    pub fn downstream(follow: Follow, behind: Option<usize>, bump: Option<i64>, run: Option<&str>, name: &str) -> Half {
399 
        let (class, word, title) = if follow == Follow::Off {
400 
            ("df-off", "not following", format!("{name} does not follow it"))
401 
        } else {
402 
            match (bump, run) {
403 
                (Some(_), Some("passed")) => ("df-green", "passes", format!("{name}'s run on the bump passed")),
404 
                (Some(_), Some("failed" | "error" | "cancelled")) => ("df-red", "fails", format!("{name}'s run on the bump failed")),
405 
                (Some(_), Some(_)) => ("df-grey", "testing", format!("{name}'s run on the bump has not finished")),
406 
                (Some(_), None) => ("df-grey", "waiting", format!("{name}'s bump has no run yet")),
407 
                (None, _) if behind == Some(0) => ("df-green", "up to date", format!("{name} pins the newest state")),
408 
                (None, _) => ("df-grey", "waiting", format!("{name} is behind; its bump has not been made yet")),
409 
            }
410 
        };
411 
        Half { class, word, title }
412 
    }
413 
}
414
415 
/// A consumer's Diffusion status at a glance (#150, Andy's colours):
416 
/// green when the upstream was checked and the consumer's run on the bump
417 
/// passed; amber when the run passed but the upstream has no checks; red
418 
/// when the run failed; grey while it runs; up to date when there is
419 
/// nothing to bump; off when it does not follow.
420 
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
421 
#[serde(rename_all = "kebab-case")]
422 
pub enum Colour {
423 
    Green,
424 
    Amber,
425 
    Red,
426 
    Grey,
427 
    UpToDate,
428 
    Off,
429 
}
430
431 
impl Colour {
432 
    pub fn class(&self) -> &'static str {
433 
        match self {
434 
            Colour::Green => "df-green",
435 
            Colour::Amber => "df-amber",
436 
            Colour::Red => "df-red",
437 
            Colour::Grey => "df-grey",
438 
            Colour::UpToDate => "df-current",
439 
            Colour::Off => "df-off",
440 
        }
441 
    }
442
443 
    pub fn label(&self) -> &'static str {
444 
        match self {
445 
            Colour::Green => "green",
446 
            Colour::Amber => "amber: upstream unverified",
447 
            Colour::Red => "red: its run failed",
448 
            Colour::Grey => "testing",
449 
            Colour::UpToDate => "up to date",
450 
            Colour::Off => "not following",
451 
        }
452 
    }
453
454 
    /// The colour from what is known of a consumer.
455 
    pub fn of(follow: Follow, behind: Option<usize>, bump: Option<i64>, run: Option<&str>, unverified: bool) -> Colour {
456 
        if follow == Follow::Off {
457 
            return Colour::Off;
458 
        }
459 
        match (bump, run) {
460 
            (Some(_), Some("passed")) if unverified => Colour::Amber,
461 
            (Some(_), Some("passed")) => Colour::Green,
462 
            (Some(_), Some("failed" | "error" | "cancelled")) => Colour::Red,
463 
            (Some(_), _) => Colour::Grey,
464 
            (None, _) if behind == Some(0) => Colour::UpToDate,
465 
            (None, _) => Colour::Grey,
466 
        }
467 
    }
468 
}
469
470 
/// The consumers of `row` that `account` may read, with how far behind
471 
/// the upstream's default bookmark each pin is.
472 
pub async fn consumers(app: &Shared, row: &RepoRow, account: Option<&Account>) -> Result<Vec<Consumer>, crate::repos::RepoError> {
473 
    let Some(name) = RepoName::new(&row.owner, &row.name) else { return Ok(Vec::new()) };
474 
    let id = app.repos.read(&name, |r, _| Ok(r.id()?)).await?;
475 
    let found: Vec<(i64, String, String, String)> = sqlx::query_as(
476 
        "SELECT consumer_id, bookmark, path, heads FROM diffusion_pins WHERE upstream = ? ORDER BY consumer_id, bookmark, path",
477 
    )
478 
    .bind(hex::encode(id))
479 
    .fetch_all(&app.db)
480 
    .await
481 
    .map_err(|e| crate::repos::RepoError::Io(std::io::Error::other(e)))?;
482 
    let mut out = Vec::new();
483 
    let mut pinned_sets = Vec::new();
484 
    let unverified = unverified(app, row).await;
485 
    for (consumer_id, bookmark, path, heads) in found {
486 
        let Ok(Some(c)) = db::repos::get_by_id(&app.db, consumer_id).await else { continue };
487 
        if !roles::allows(roles::role_for(&app.db, account, &c).await.ok().flatten(), crate::auth::Access::Read) {
488 
            continue;
489 
        }
490 
        // One row per consumer and subrepo: the bookmark its work lands on.
491 
        if bookmark != target(&app.db, &c).await {
492 
            continue;
493 
        }
494 
        let pinned: Vec<String> = heads.split('+').filter(|h| !h.is_empty()).map(String::from).collect();
495 
        pinned_sets.push(pinned.iter().filter_map(|h| ChangeHash::from_hex(h)).collect::<Vec<_>>());
496 
        let bump_bm = bump_bookmark(&path);
497 
        let mine: Vec<db::proposals::ProposalRow> = db::proposals::list(&app.db, c.id, &[])
498 
            .await
499 
            .unwrap_or_default()
500 
            .into_iter()
501 
            .filter(|p| p.source_bookmark == bump_bm)
502 
            .collect();
503 
        let bump = mine.iter().find(|p| p.state == "open" || p.state == "draft").map(|p| p.number);
504 
        let landed = mine.iter().filter(|p| p.state == "landed").max_by_key(|p| p.landed_ms).map(|p| (p.number, p.landed_ms));
505 
        let run: Option<(i64, String)> =
506 
            sqlx::query_as("SELECT number, status FROM flow_runs WHERE repo_id = ? AND bookmark = ? ORDER BY id DESC LIMIT 1")
507 
                .bind(c.id)
508 
                .bind(&bump_bm)
509 
                .fetch_optional(&app.db)
510 
                .await
511 
                .ok()
512 
                .flatten();
513 
        let follow = rule(&app.config.diffusion.follow, &c, None, &path).follow;
514 
        out.push(Consumer {
515 
            owner: c.owner,
516 
            name: c.name,
517 
            bookmark,
518 
            path,
519 
            pinned,
520 
            behind: None,
521 
            follow,
522 
            bump,
523 
            run,
524 
            landed,
525 
            colour: Colour::Off,
526 
            upstream: Half { class: "df-grey", word: "", title: String::new() },
527 
            downstream: Half { class: "df-grey", word: "", title: String::new() },
528 
        });
529 
    }
530 
    if out.is_empty() {
531 
        return Ok(out);
532 
    }
533 
    let default = row.default_bookmark.clone();
534 
    let behind = app
535 
        .repos
536 
        .read(&name, move |r, cache| {
537 
            let tip = heads_of(r, &default)?;
538 
            let tip = cache.summary(r, &tip)?;
539 
            Ok(pinned_sets
540 
                .iter()
541 
                .map(|p| cache.summary(r, p).ok().map(|s| tip.changes.iter().filter(|c| !s.changes.contains(*c)).count()))
542 
                .collect::<Vec<_>>())
543 
        })
544 
        .await?;
545 
    let up_run: Option<String> =
546 
        sqlx::query_scalar("SELECT status FROM flow_runs WHERE repo_id = ? AND bookmark = ? ORDER BY id DESC LIMIT 1")
547 
            .bind(row.id)
548 
            .bind(&row.default_bookmark)
549 
            .fetch_optional(&app.db)
550 
            .await
551 
            .ok()
552 
            .flatten();
553 
    let slug = format!("{}/{}", row.owner, row.name);
554 
    for (c, b) in out.iter_mut().zip(behind) {
555 
        c.behind = b;
556 
        let run = c.run.as_ref().map(|r| r.1.clone());
557 
        c.colour = Colour::of(c.follow, b, c.bump, run.as_deref(), unverified);
558 
        c.upstream = Half::upstream(unverified, up_run.as_deref(), &slug);
559 
        c.downstream = Half::downstream(c.follow, b, c.bump, run.as_deref(), &format!("{}/{}", c.owner, c.name));
560 
    }
561 
    Ok(out)
562 
}
563
564 
/// What `consumer` follows, as its Follows panel shows: each upstream on
565 
/// this hub it pins, with its status there.
566 
pub async fn follows(app: &Shared, consumer: &RepoRow, account: Option<&Account>) -> Vec<(RepoRow, Consumer)> {
567 
    let ups: Vec<i64> =
568 
        sqlx::query_scalar("SELECT DISTINCT upstream_repo_id FROM diffusion_pins WHERE consumer_id = ? AND upstream_repo_id IS NOT NULL")
569 
            .bind(consumer.id)
570 
            .fetch_all(&app.db)
571 
            .await
572 
            .unwrap_or_default();
573 
    let mut out = Vec::new();
574 
    for id in ups {
575 
        let Ok(Some(up)) = db::repos::get_by_id(&app.db, id).await else { continue };
576 
        if !roles::allows(roles::role_for(&app.db, account, &up).await.ok().flatten(), crate::auth::Access::Read) {
577 
            continue;
578 
        }
579 
        for c in consumers(app, &up, account).await.unwrap_or_default() {
580 
            if c.owner == consumer.owner && c.name == consumer.name {
581 
                out.push((up.clone(), c));
582 
            }
583 
        }
584 
    }
585 
    out
586 
}
587
588 
/// Every upstream on this hub with its consumers, for Admin › Diffusion.
589 
pub async fn everything(app: &Shared, account: Option<&Account>) -> Vec<(RepoRow, Vec<Consumer>, bool)> {
590 
    let ups: Vec<i64> = sqlx::query_scalar("SELECT DISTINCT upstream_repo_id FROM diffusion_pins WHERE upstream_repo_id IS NOT NULL")
591 
        .fetch_all(&app.db)
592 
        .await
593 
        .unwrap_or_default();
594 
    let mut out = Vec::new();
595 
    for id in ups {
596 
        let Ok(Some(up)) = db::repos::get_by_id(&app.db, id).await else { continue };
597 
        let list = consumers(app, &up, account).await.unwrap_or_default();
598 
        let unverified = unverified(app, &up).await;
599 
        out.push((up, list, unverified));
600 
    }
601 
    out
602 
}
603
604 
/// The bookmark a consumer's bump for the subrepo at `path` lives on.
605 
/// A bump change's message: its title names the upstream state by its
606 
/// first head and head count, so a state of many heads does not make it
607 
/// run on (oxygenhub#216); the whole address is in the body.
608 
pub fn bump_message(path: &str, upstream: &str, heads: &[String]) -> String {
609 
    let first = heads.first().map(String::as_str).unwrap_or("");
610 
    match heads.len() {
611 
        0 | 1 => format!("Diffusion: move {path} to {upstream} {first}"),
612 
        n => format!("Diffusion: move {path} to {upstream} {first} (+{} heads)\n\nPins {upstream} at {}", n - 1, heads.join("+")),
613 
    }
614 
}
615
616 
pub fn bump_bookmark(path: &str) -> String {
617 
    format!("diffusion/{}", path.trim_matches('/'))
618 
}
619
620 
/// What a bump did.
621 
#[derive(Clone, Debug, PartialEq, Eq)]
622 
pub enum Bumped {
623 
    /// The bump bookmark now holds the target plus one change moving the pin.
624 
    Moved { bookmark: String, change: ChangeHash, heads: Vec<ChangeHash>, from: Vec<ChangeHash>, state: String },
625 
    /// The pin already holds the upstream state.
626 
    UpToDate,
627 
    /// The bump bookmark already moves the pin to that state.
628 
    Already { from: Vec<ChangeHash>, to: Vec<ChangeHash> },
629 
    /// The pin names changes the upstream does not hold: not followed.
630 
    Unknown,
631 
    /// Taking the upstream's state with the pin's own extra changes would
632 
    /// conflict.
633 
    Conflicts,
634 
    /// The consumer has no Oxygen pin of that upstream at `path` on its
635 
    /// target bookmark.
636 
    NoPin,
637 
}
638
639 
/// Where a consumer's pin `old` moves when the upstream reaches `new`,
640 
/// worked out in the upstream (§4.3).
641 
#[derive(Clone, Debug, PartialEq, Eq)]
642 
pub enum Plan {
643 
    /// To these heads, which add this many changes.
644 
    To(Vec<ChangeHash>, usize),
645 
    /// Nothing to add.
646 
    UpToDate,
647 
    /// The pin names changes the upstream does not hold (rewritten, or
648 
    /// from elsewhere): not followed.
649 
    Unknown,
650 
    /// Taking `new` with the pin's own extra changes would conflict.
651 
    Conflicts,
652 
}
653
654 
/// The pin `old` moved to the upstream's `new`. When the pin has changes
655 
/// `new` lacks (say, from an upstream proposal not landed yet), they are
656 
/// kept: changes that do not touch each other combine, so the move is to
657 
/// both, never dropping what the consumer already relies on.
658 
pub async fn plan(app: &Shared, upstream: &RepoName, old: Vec<ChangeHash>, new: Vec<ChangeHash>) -> Result<Plan, crate::repos::RepoError> {
659 
    app.repos
660 
        .read(upstream, move |r, cache| {
661 
            let Ok(o) = cache.summary(r, &old) else { return Ok(Plan::Unknown) };
662 
            let n = cache.summary(r, &new)?;
663 
            if o.changes.iter().all(|c| n.changes.contains(c)) {
664 
                let added = n.changes.len() - o.changes.len();
665 
                return Ok(if added == 0 { Plan::UpToDate } else { Plan::To(new, added) });
666 
            }
667 
            let mut both = new.clone();
668 
            both.extend(old.iter().copied());
669 
            let heads = r.load_state(&both)?.heads();
670 
            let b = cache.summary(r, &heads)?;
671 
            if b.conflicted && !o.conflicted {
672 
                return Ok(Plan::Conflicts);
673 
            }
674 
            let added = b.changes.len() - o.changes.len();
675 
            Ok(if added == 0 { Plan::UpToDate } else { Plan::To(heads, added) })
676 
        })
677 
        .await
678 
}
679
680 
/// Move `consumer`'s pin of `upstream` at `path` to `new_heads`, on a bump
681 
/// bookmark built from `target` (§4.3): one change, signed with the hub's
682 
/// key, on the target's heads. The upstream decides forwardness.
683 
pub async fn bump(
684 
    app: &Shared,
685 
    consumer: &RepoRow,
686 
    target: &str,
687 
    path: &str,
688 
    upstream: &RepoRow,
689 
    new_heads: Vec<ChangeHash>,
690 
) -> Result<Bumped, crate::repos::RepoError> {
691 
    use oxy_core::identity::Domain;
692 
    use oxy_core::subrepo::Pin;
693 
    let (Some(cname), Some(uname)) = (RepoName::new(&consumer.owner, &consumer.name), RepoName::new(&upstream.owner, &upstream.name))
694 
    else {
695 
        return Ok(Bumped::NoPin);
696 
    };
697 
    let upstream_id = app.repos.read(&uname, |r, _| Ok(r.id()?)).await?;
698 
    // The consumer's current pin at `path` on its target, and on its bump
699 
    // bookmark if it has one.
700 
    let (t, p) = (target.to_string(), path.trim_matches('/').to_string());
701 
    let (current, bumped) = app
702 
        .repos
703 
        .read(&cname, move |r, cache| {
704 
            let pin_on = |bookmark: &str| -> Result<Option<Pin>, crate::repos::RepoError> {
705 
                let heads = heads_of(r, bookmark)?;
706 
                if heads.is_empty() {
707 
                    return Ok(None);
708 
                }
709 
                let loaded = cache.load(r, &heads)?;
710 
                let Some(f) = loaded.snapshot.files.get(&p) else { return Ok(None) };
711 
                let oxy_core::state::FileContent::Opaque { blobs } = &f.content else { return Ok(None) };
712 
                let Some(b) = blobs.first() else { return Ok(None) };
713 
                Ok(Pin::decode(&r.read_blob(b)?).ok())
714 
            };
715 
            Ok((pin_on(&t)?, pin_on(&bump_bookmark(&p))?))
716 
        })
717 
        .await?;
718 
    let Some(Pin::Oxy { repo, heads: old, bookmark }) = current else { return Ok(Bumped::NoPin) };
719 
    if repo != upstream_id {
720 
        return Ok(Bumped::NoPin);
721 
    }
722 
    let new_heads = match plan(app, &uname, old.clone(), new_heads.clone()).await? {
723 
        Plan::To(heads, _) => heads,
724 
        Plan::UpToDate => return Ok(Bumped::UpToDate),
725 
        Plan::Unknown => return Ok(Bumped::Unknown),
726 
        Plan::Conflicts => return Ok(Bumped::Conflicts),
727 
    };
728 
    if let Some(Pin::Oxy { heads: b, .. }) = &bumped {
729 
        let (mut b, mut n) = (b.clone(), new_heads.clone());
730 
        b.sort();
731 
        n.sort();
732 
        if b == n {
733 
            return Ok(Bumped::Already { from: old, to: new_heads });
734 
        }
735 
    }
736 
    let (key, secret) =
737 
        crate::keys::load_secret(&app.config.keys_dir()).map_err(|e| crate::repos::RepoError::Io(std::io::Error::other(e.to_string())))?;
738 
    let identity = app.hub_key.identity.clone();
739 
    let upstream_label = format!("{}/{}", upstream.owner, upstream.name);
740 
    let (t, p) = (target.to_string(), path.trim_matches('/').to_string());
741 
    let pin = Pin::Oxy { repo, heads: new_heads.clone(), bookmark };
742 
    let old = old.clone();
743 
    let short: Vec<String> = new_heads.iter().map(|h| h.to_hex()[..12].to_string()).collect();
744 
    let message = bump_message(&p, &upstream_label, &short);
745 
    app.repos
746 
        .write(&cname, move |r, _| {
747 
            let base = heads_of(r, &t)?;
748 
            let change = r.build_pin_change(&base, &p, &pin, &message, key)?;
749 
            let hash = change.hash();
750 
            let sig = secret.sign(Domain::Change, &key, &hash);
751 
            let mut state = r.load_state(&base)?;
752 
            state.add(change.clone()).map_err(|e| crate::repos::RepoError::Oxy(oxy_core::repo::RepoError::Msg(e.to_string())))?;
753 
            let mut candidates = base.clone();
754 
            candidates.push(hash);
755 
            let heads = state.maximal(&candidates);
756 
            let bookmark = bump_bookmark(&p);
757 
            // The bump bookmark is the hub's own and is rebuilt on the
758 
            // target each time, so the bump it replaces may go.
759 
            let mut update = r.host_ref_update(&bookmark, &heads, Some((&key, &secret)))?;
760 
            update.force = true;
761 
            let state = update.new_state.to_hex();
762 
            let report = r.apply_hosted(Some((change, vec![sig])), &[identity], update)?;
763 
            if let Some(refused) = report.results.iter().find(|x| !x.ok) {
764 
                return Err(crate::repos::RepoError::Oxy(oxy_core::repo::RepoError::Msg(format!(
765 
                    "the bump was refused: {}",
766 
                    refused.reason
767 
                ))));
768 
            }
769 
            Ok(Bumped::Moved { bookmark, change: hash, heads, from: old, state })
770 
        })
771 
        .await
772 
}
773
774 
/// The account Diffusion's proposals are opened as.
775 
pub const ACCOUNT: &str = "diffusion";
776
777 
/// The `diffusion` account, made on first use with no way to sign in.
778 
pub async fn account(pool: &Pool) -> Option<Account> {
779 
    if let Ok(Some(a)) = db::accounts::account_by_name(pool, ACCOUNT).await {
780 
        return Some(a);
781 
    }
782 
    db::accounts::create_account(pool, ACCOUNT, "diffusion@localhost", "!", "", false).await.ok()
783 
}
784
785 
/// Whether `path` matches a `watch` pattern: `*` is any run of characters
786 
/// within one path segment, `**` any run across segments, `?` one
787 
/// character; a pattern without a wildcard also matches everything below
788 
/// it, as a directory would.
789 
pub fn watched(pattern: &str, path: &str) -> bool {
790 
    fn m(p: &[u8], s: &[u8]) -> bool {
791 
        match (p.first(), s.first()) {
792 
            (None, None) => true,
793 
            (Some(b'*'), _) if p.get(1) == Some(&b'*') => (0..=s.len()).any(|i| m(&p[2..], &s[i..])),
794 
            (Some(b'*'), _) => (0..=s.len()).take_while(|&i| i == 0 || s[i - 1] != b'/').any(|i| m(&p[1..], &s[i..])),
795 
            (Some(b'?'), Some(c)) if *c != b'/' => m(&p[1..], &s[1..]),
796 
            (Some(a), Some(b)) if a == b => m(&p[1..], &s[1..]),
797 
            _ => false,
798 
        }
799 
    }
800 
    let p = pattern.trim().trim_matches('/');
801 
    if p.is_empty() {
802 
        return false;
803 
    }
804 
    m(p.as_bytes(), path.as_bytes()) || (!p.contains(['*', '?']) && path.starts_with(&format!("{p}/")))
805 
}
806
807 
/// What a bump brings: the upstream changes in `new` and not in `old`,
808 
/// newest first, as Markdown lines linking to the upstream's change pages,
809 
/// with those touching a `watch` pattern flagged (and listed again).
810 
pub struct Changelog {
811 
    pub lines: Vec<String>,
812 
    pub watched: Vec<String>,
813 
}
814
815 
pub async fn changelog(app: &Shared, upstream: &RepoRow, old: Vec<ChangeHash>, new: Vec<ChangeHash>, watch: Vec<String>) -> Changelog {
816 
    let Some(name) = RepoName::new(&upstream.owner, &upstream.name) else { return Changelog { lines: Vec::new(), watched: Vec::new() } };
817 
    let linked: std::collections::HashMap<String, String> =
818 
        db::accounts::identity_accounts(&app.db).await.unwrap_or_default().into_iter().collect();
819 
    let base = format!("/{}/{}", upstream.owner, upstream.name);
820 
    app.repos
821 
        .read(&name, move |r, cache| {
822 
            let (o, n) = (cache.summary(r, &old)?, cache.summary(r, &new)?);
823 
            let mut added: Vec<(u64, ChangeHash, String, String, Vec<String>)> = Vec::new();
824 
            for h in n.changes.iter().filter(|c| !o.changes.contains(*c)) {
825 
                let Ok(c) = r.store().get_change(h) else { continue };
826 
                let key = c.author.to_hex();
827 
                let who = linked.get(&key).cloned().unwrap_or_else(|| key[..12].to_string());
828 
                let first = c.message.lines().next().unwrap_or("").trim().to_string();
829 
                // The watched paths it touches, when the game watches any.
830 
                let hits: Vec<String> = if watch.is_empty() {
831 
                    Vec::new()
832 
                } else {
833 
                    crate::repos::view::change(r, cache, h)
834 
                        .map(|d| d.files.into_iter().map(|f| f.path).filter(|p| watch.iter().any(|w| watched(w, p))).collect())
835 
                        .unwrap_or_default()
836 
                };
837 
                added.push((c.hlc.wall_ms, *h, first, who, hits));
838 
            }
839 
            added.sort_by_key(|a| std::cmp::Reverse(a.0));
840 
            let line = |h: &ChangeHash, msg: &str, who: &str, flag: bool| {
841 
                format!(
842 
                    "- {}[`{}`]({base}/change/{}) {} ({who})",
843 
                    if flag { "⚑ " } else { "" },
844 
                    &h.to_hex()[..12],
845 
                    h.to_hex(),
846 
                    msg.replace('\n', " ")
847 
                )
848 
            };
849 
            let watched = added
850 
                .iter()
851 
                .filter(|a| !a.4.is_empty())
852 
                .map(|(_, h, msg, who, hits)| {
853 
                    format!(
854 
                        "{} — touches {}",
855 
                        line(h, msg, who, true),
856 
                        hits.iter().map(|p| format!("`{p}`")).collect::<Vec<_>>().join(", ")
857 
                    )
858 
                })
859 
                .collect();
860 
            let lines = added.iter().map(|(_, h, msg, who, hits)| line(h, msg, who, !hits.is_empty())).collect();
861 
            Ok(Changelog { lines, watched })
862 
        })
863 
        .await
864 
        .unwrap_or(Changelog { lines: Vec::new(), watched: Vec::new() })
865 
}
866
867 
/// Why a bump was made, which its proposal says.
868 
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
869 
pub enum Why {
870 
    /// The upstream's state passed its checks.
871 
    Green,
872 
    /// The upstream moved (the consumer follows every move).
873 
    Moved,
874 
    /// The upstream has no checks at all: nothing vouched for it, so the
875 
    /// proposal is marked amber, `upstream unverified`.
876 
    Unverified,
877 
}
878
879 
/// Open the bump proposal for `path`'s bump bookmark, or bring the open
880 
/// one up to date: into `target`, by the `diffusion` account, labelled
881 
/// `dependencies` (and `upstream unverified` while that is so), its body
882 
/// the changelog from the pin `from` to `to`. Its number.
883 
pub async fn propose(
884 
    app: &Shared,
885 
    consumer: &RepoRow,
886 
    target: &str,
887 
    path: &str,
888 
    upstream: &RepoRow,
889 
    (from, to): (Vec<ChangeHash>, Vec<ChangeHash>),
890 
    (why, watch, lands): (Why, Vec<String>, bool),
891 
) -> Option<i64> {
892 
    use crate::db::proposals as pdb;
893 
    let bookmark = bump_bookmark(path);
894 
    let bookmark = bookmark.as_str();
895 
    let me = account(&app.db).await?;
896 
    let log = changelog(app, upstream, from, to.clone(), watch.clone()).await;
897 
    let lines = log.lines;
898 
    let flagged = if log.watched.is_empty() {
899 
        String::new()
900 
    } else {
901 
        format!(
902 
            "\n\n**Touches what you watch** ({}):\n\n{}",
903 
            watch.iter().map(|w| format!("`{w}`")).collect::<Vec<_>>().join(", "),
904 
            log.watched.join("\n")
905 
        )
906 
    };
907 
    let slug = format!("{}/{}", upstream.owner, upstream.name);
908 
    let short: Vec<String> = to.iter().map(|h| h.to_hex()[..12].to_string()).collect();
909 
    let ub = &upstream.default_bookmark;
910 
    let count = if lines.len() == 1 { "1 change".to_string() } else { format!("{} changes", lines.len()) };
911 
    let title = format!("Diffusion: {path} to {slug} {ub} ({count})");
912 
    let when = match why {
913 
        Why::Green => "at its newest state whose checks passed",
914 
        Why::Moved => "as it is now",
915 
        Why::Unverified => "as it is now",
916 
    };
917 
    let amber = if why == Why::Unverified {
918 
        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.")
919 
    } else {
920 
        String::new()
921 
    };
922 
    let body = format!(
923 
        "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). {}",
924 
        crate::web::state_label(&short.join("+")),
925 
        lines.join("\n"),
926 
        if lands {
927 
            "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."
928 
        } else {
929 
            "It never lands by itself."
930 
        }
931 
    );
932 
    let open = pdb::list(&app.db, consumer.id, &["open", "draft"]).await.ok()?.into_iter().find(|p| p.source_bookmark == bookmark);
933 
    let (number, id) = match open {
934 
        Some(p) => {
935 
            if p.title != title || p.body != body {
936 
                pdb::edit(&app.db, p.id, &title, &body, target).await.ok()?;
937 
                pdb::add_event(&app.db, p.id, ACCOUNT, "updated", &crate::web::state_label(&short.join("+"))).await.ok();
938 
            }
939 
            (p.number, p.id)
940 
        }
941 
        None => {
942 
            let p = pdb::create(
943 
                &app.db,
944 
                pdb::NewProposal {
945 
                    repo_id: consumer.id,
946 
                    title: &title,
947 
                    body: &body,
948 
                    author_id: Some(me.id),
949 
                    author: ACCOUNT,
950 
                    draft: false,
951 
                    target,
952 
                    source_kind: "bookmark",
953 
                    source_repo_id: None,
954 
                    source_bookmark: bookmark,
955 
                    source_changes: "",
956 
                },
957 
            )
958 
            .await
959 
            .ok()?;
960 
            pdb::add_event(&app.db, p.id, ACCOUNT, "opened", "").await.ok();
961 
            db::notify::index(&app.db, "proposal", consumer.id, p.id, p.number, &p.title, &p.body).await.ok();
962 
            // Told as a proposal opened on the web is (oxygenhub#183).
963 
            crate::notify::deliver(
964 
                app,
965 
                crate::notify::Event {
966 
                    kind: "proposal.opened",
967 
                    title: format!("{}/{}#{}: {}", consumer.owner, consumer.name, p.number, p.title),
968 
                    url: format!("/{}/{}/proposal/{}", consumer.owner, consumer.name, p.number),
969 
                    actor: &me,
970 
                    repo_id: consumer.id,
971 
                    thread: Some(("proposal", p.id)),
972 
                    mention_text: "",
973 
                    direct: vec![],
974 
                },
975 
            )
976 
            .await;
977 
            if let Ok(label) =
978 
                db::issues::upsert_label(&app.db, consumer.id, "dependencies", "#0072ff", "A subrepo moves (Diffusion)").await
979 
            {
980 
                db::issues::set_proposal_label(&app.db, p.id, label.id, true).await.ok();
981 
            }
982 
            (p.number, p.id)
983 
        }
984 
    };
985 
    // Amber while the upstream has no checks; cleared by a green bump.
986 
    let unverified = db::issues::label_by_name(&app.db, consumer.id, "upstream unverified").await.ok().flatten();
987 
    match (why == Why::Unverified, unverified) {
988 
        (true, l) => {
989 
            let l = match l {
990 
                Some(l) => Some(l),
991 
                None => db::issues::upsert_label(
992 
                    &app.db,
993 
                    consumer.id,
994 
                    "upstream unverified",
995 
                    "#e3a008",
996 
                    "The upstream has no checks (Diffusion)",
997 
                )
998 
                .await
999 
                .ok(),
1000 
            };
1001 
            if let Some(l) = l {
1002 
                db::issues::set_proposal_label(&app.db, id, l.id, true).await.ok();
1003 
            }
1004 
        }
1005 
        (false, Some(l)) => {
1006 
            db::issues::set_proposal_label(&app.db, id, l.id, false).await.ok();
1007 
        }
1008 
        (false, None) => {}
1009 
    }
1010 
    Some(number)
1011 
}
1012
1013 
/// Start a bump's pipeline, as a push of it would. Boxed as `Send`, which
1014 
/// breaks the cycle between a bump starting runs and a passing run
1015 
/// starting bumps that the compiler cannot see through.
1016 
fn start_runs(
1017 
    app: Shared,
1018 
    row: RepoRow,
1019 
    bookmark: String,
1020 
    state: String,
1021 
    heads: Vec<ChangeHash>,
1022 
) -> std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send>> {
1023 
    Box::pin(async move { crate::flow::on_push(&app, &row, &bookmark, &state, &heads, ACCOUNT).await })
1024 
}
1025
1026 
/// The text of `.oxymodules` on `bookmark` of `name`, if any.
1027 
async fn oxymodules(app: &Shared, name: &RepoName, bookmark: &str) -> Option<String> {
1028 
    let b = bookmark.to_string();
1029 
    app.repos
1030 
        .read(name, move |r, cache| {
1031 
            let heads = heads_of(r, &b)?;
1032 
            if heads.is_empty() {
1033 
                return Ok(None);
1034 
            }
1035 
            let loaded = cache.load(r, &heads)?;
1036 
            Ok(match loaded.snapshot.files.get(".oxymodules").map(|f| &f.content) {
1037 
                Some(oxy_core::state::FileContent::Text { bytes, .. }) => Some(String::from_utf8_lossy(bytes).into_owned()),
1038 
                _ => None,
1039 
            })
1040 
        })
1041 
        .await
1042 
        .ok()
1043 
        .flatten()
1044 
}
1045
1046 
/// The upstream `upstream` reached `heads` on its default bookmark: by a
1047 
/// push (`green` false) or a passing run of that state (`green` true).
1048 
/// Bump and propose in every consumer that follows it that way (§4.3).
1049 
pub async fn propagate(app: &Shared, upstream: &RepoRow, heads: Vec<ChangeHash>, green: bool) {
1050 
    propagate_to(app, upstream, heads, green, None).await
1051 
}
1052
1053 
/// `propagate`, to one consumer only when `only` names it.
1054 
async fn propagate_to(app: &Shared, upstream: &RepoRow, heads: Vec<ChangeHash>, green: bool, only: Option<i64>) {
1055 
    let Some(uname) = RepoName::new(&upstream.owner, &upstream.name) else { return };
1056 
    let Ok(id) = app.repos.read(&uname, |r, _| Ok(r.id()?)).await else { return };
1057 
    let rows: Vec<(i64, String, String)> = sqlx::query_as("SELECT consumer_id, bookmark, path FROM diffusion_pins WHERE upstream = ?")
1058 
        .bind(hex::encode(id))
1059 
        .fetch_all(&app.db)
1060 
        .await
1061 
        .unwrap_or_default();
1062 
    let unverified = unverified(app, upstream).await;
1063 
    for (consumer_id, indexed, path) in rows {
1064 
        if only.is_some_and(|o| o != consumer_id) {
1065 
            continue;
1066 
        }
1067 
        let Ok(Some(consumer)) = db::repos::get_by_id(&app.db, consumer_id).await else { continue };
1068 
        let Some(cname) = RepoName::new(&consumer.owner, &consumer.name) else { continue };
1069 
        let learned = target(&app.db, &consumer).await;
1070 
        let modules = oxymodules(app, &cname, &learned).await;
1071 
        let r = rule(&app.config.diffusion.follow, &consumer, modules.as_deref(), &path);
1072 
        let target = r.target.clone().unwrap_or(learned);
1073 
        if indexed != target {
1074 
            continue;
1075 
        }
1076 
        // An upstream with no checks never turns green: its moves count
1077 
        // for green followers too, marked amber (unverified).
1078 
        let go = match r.follow {
1079 
            Follow::Off => false,
1080 
            Follow::Green => green || unverified,
1081 
            Follow::Any => !green,
1082 
        };
1083 
        let why = if green {
1084 
            Why::Green
1085 
        } else if unverified {
1086 
            Why::Unverified
1087 
        } else {
1088 
            Why::Moved
1089 
        };
1090 
        if !go {
1091 
            continue;
1092 
        }
1093 
        match bump(app, &consumer, &target, &path, upstream, heads.clone()).await {
1094 
            Ok(Bumped::Moved { from, bookmark, heads: bumped, state, .. }) => {
1095 
                // Its pipeline runs as for a push, so the proposal is tested.
1096 
                tokio::spawn(start_runs(app.clone(), consumer.clone(), bookmark, state, bumped));
1097 
                let n = propose(app, &consumer, &target, &path, upstream, (from, heads.clone()), (why, r.watch.clone(), r.land_auto)).await;
1098 
                tracing::info!(
1099 
                    "diffusion: {}/{} {path} bumped to {}/{} (proposal {n:?})",
1100 
                    consumer.owner,
1101 
                    consumer.name,
1102 
                    upstream.owner,
1103 
                    upstream.name
1104 
                );
1105 
            }
1106 
            Ok(Bumped::Already { from, to }) => {
1107 
                // Already moved: its proposal's text and labels are brought
1108 
                // up to date all the same.
1109 
                propose(app, &consumer, &target, &path, upstream, (from, to), (why, r.watch.clone(), r.land_auto)).await;
1110 
                // The upstream may just have turned green under a bump its
1111 
                // consumer already passed.
1112 
                if green && r.land_auto {
1113 
                    tokio::spawn(try_land(app.clone(), consumer.clone(), path.clone()));
1114 
                }
1115 
            }
1116 
            Ok(other) => tracing::info!("diffusion: {}/{} {path}: no bump ({other:?})", consumer.owner, consumer.name),
1117 
            Err(e) => tracing::warn!("diffusion: bump {}/{} {path}: {e}", consumer.owner, consumer.name),
1118 
        }
1119 
    }
1120 
}
1121
1122 
/// Why a bump was not landed by itself (docs/diffusion.md §4.2).
1123 
#[derive(Debug, PartialEq, Eq)]
1124 
pub enum Held {
1125 
    /// No open bump at that path.
1126 
    NoBump,
1127 
    /// The consumer did not choose `land = auto`.
1128 
    Manual,
1129 
    /// The upstream has no checks: amber never lands by itself.
1130 
    Unverified,
1131 
    /// The upstream state the bump pins has no passing run.
1132 
    UpstreamUnchecked,
1133 
    /// The consumer's run on the bump as it is now has not passed.
1134 
    NotPassed,
1135 
    /// The landing was refused (required checks, approvals, the trust policy).
1136 
    Refused(String),
1137 
}
1138
1139 
/// Land the open bump at `path` in `consumer` itself, as the `diffusion`
1140 
/// account, when the consumer chose `land = auto` (its `.oxymodules` entry,
1141 
/// else Settings › Diffusion) and both halves of its pill are green: the
1142 
/// upstream state it pins passed its own run (never amber: an upstream
1143 
/// with no checks always waits for a person), and the consumer's run on
1144 
/// the bump as it is now passed. It lands through the web's own path
1145 
/// (`land_as`), signed by the hub's key, so required checks, approvals and
1146 
/// the trust policy decide as for anyone; a refusal is noted once on the
1147 
/// proposal's timeline. The proposal's number when it landed.
1148 
pub async fn auto_land(app: &Shared, consumer: &RepoRow, path: &str) -> Result<i64, Held> {
1149 
    let path = path.trim_matches('/');
1150 
    let bm = bump_bookmark(path);
1151 
    let Some(cname) = RepoName::new(&consumer.owner, &consumer.name) else { return Err(Held::NoBump) };
1152 
    let bump = db::proposals::list(&app.db, consumer.id, &["open"])
1153 
        .await
1154 
        .ok()
1155 
        .and_then(|ps| ps.into_iter().find(|p| p.source_bookmark == bm && p.author == ACCOUNT))
1156 
        .ok_or(Held::NoBump)?;
1157 
    let learned = target(&app.db, consumer).await;
1158 
    let modules = oxymodules(app, &cname, &learned).await;
1159 
    if !rule(&app.config.diffusion.follow, consumer, modules.as_deref(), path).land_auto {
1160 
        return Err(Held::Manual);
1161 
    }
1162 
    let upstream_id: Option<i64> = sqlx::query_scalar(
1163 
        "SELECT upstream_repo_id FROM diffusion_pins WHERE consumer_id = ? AND path = ? AND upstream_repo_id IS NOT NULL LIMIT 1",
1164 
    )
1165 
    .bind(consumer.id)
1166 
    .bind(path)
1167 
    .fetch_optional(&app.db)
1168 
    .await
1169 
    .ok()
1170 
    .flatten();
1171 
    let Some(upstream) = (match upstream_id {
1172 
        Some(id) => db::repos::get_by_id(&app.db, id).await.ok().flatten(),
1173 
        None => None,
1174 
    }) else {
1175 
        return Err(Held::UpstreamUnchecked);
1176 
    };
1177 
    if unverified(app, &upstream).await {
1178 
        return Err(Held::Unverified);
1179 
    }
1180 
    // What the bump pins, and the bump bookmark's heads now.
1181 
    let (p, b) = (path.to_string(), bm.clone());
1182 
    let (pinned, now) = app
1183 
        .repos
1184 
        .read(&cname, move |r, cache| {
1185 
            let heads = heads_of(r, &b)?;
1186 
            if heads.is_empty() {
1187 
                return Ok((None, heads));
1188 
            }
1189 
            let loaded = cache.load(r, &heads)?;
1190 
            let pin = match loaded.snapshot.files.get(&p).map(|f| &f.content) {
1191 
                Some(oxy_core::state::FileContent::Opaque { blobs }) => match blobs.first() {
1192 
                    Some(blob) => oxy_core::subrepo::Pin::decode(&r.read_blob(blob)?).ok(),
1193 
                    None => None,
1194 
                },
1195 
                _ => None,
1196 
            };
1197 
            Ok((pin, heads))
1198 
        })
1199 
        .await
1200 
        .map_err(|e| Held::Refused(e.to_string()))?;
1201 
    let Some(oxy_core::subrepo::Pin::Oxy { heads: pinned, .. }) = pinned else { return Err(Held::UpstreamUnchecked) };
1202 
    let same = |text: &str, want: &[ChangeHash]| {
1203 
        let mut a: Vec<String> = text.split('+').filter(|h| !h.is_empty()).map(str::to_string).collect();
1204 
        let mut b: Vec<String> = want.iter().map(|h| h.to_hex()).collect();
1205 
        a.sort();
1206 
        b.sort();
1207 
        a == b
1208 
    };
1209 
    let upstream_runs: Vec<String> = sqlx::query_scalar(
1210 
        "SELECT heads FROM flow_runs WHERE repo_id = ? AND bookmark = ? AND status = 'passed' ORDER BY id DESC LIMIT 100",
1211 
    )
1212 
    .bind(upstream.id)
1213 
    .bind(&upstream.default_bookmark)
1214 
    .fetch_all(&app.db)
1215 
    .await
1216 
    .unwrap_or_default();
1217 
    if !upstream_runs.iter().any(|h| same(h, &pinned)) {
1218 
        return Err(Held::UpstreamUnchecked);
1219 
    }
1220 
    let run: Option<(String, String)> =
1221 
        sqlx::query_as("SELECT status, heads FROM flow_runs WHERE repo_id = ? AND bookmark = ? ORDER BY id DESC LIMIT 1")
1222 
            .bind(consumer.id)
1223 
            .bind(&bm)
1224 
            .fetch_optional(&app.db)
1225 
            .await
1226 
            .ok()
1227 
            .flatten();
1228 
    if !run.is_some_and(|(status, heads)| status == "passed" && same(&heads, &now)) {
1229 
        return Err(Held::NotPassed);
1230 
    }
1231 
    let me = account(&app.db).await.ok_or_else(|| Held::Refused("the diffusion account is missing".into()))?;
1232 
    let via = "Diffusion, both halves green";
1233 
    match crate::web::proposals::land_as(app, &cname, consumer, &me, bump.number, via, true).await {
1234 
        Ok(_) => {
1235 
            tracing::info!("diffusion: {}/{} landed #{} itself (both halves green)", consumer.owner, consumer.name, bump.number);
1236 
            Ok(bump.number)
1237 
        }
1238 
        Err(crate::web::proposals::Landing::NotFound) => Err(Held::NoBump),
1239 
        Err(crate::web::proposals::Landing::Refused(why)) => {
1240 
            // Said once, so the timeline explains why a green bump waits.
1241 
            let said = db::proposals::events(&app.db, bump.id)
1242 
                .await
1243 
                .unwrap_or_default()
1244 
                .iter()
1245 
                .any(|e| e.event == "not landed by itself" && e.detail == why);
1246 
            if !said {
1247 
                db::proposals::add_event(&app.db, bump.id, ACCOUNT, "not landed by itself", &why).await.ok();
1248 
            }
1249 
            Err(Held::Refused(why))
1250 
        }
1251 
    }
1252 
}
1253
1254 
/// [`auto_land`] in the background, logged. Boxed as `Send`: a landing
1255 
/// moves the target as a push would, which may bump and land again.
1256 
pub fn try_land(app: Shared, consumer: RepoRow, path: String) -> std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send>> {
1257 
    Box::pin(async move {
1258 
        match auto_land(&app, &consumer, &path).await {
1259 
            Ok(_) | Err(Held::NoBump | Held::Manual) => {}
1260 
            Err(why) => tracing::info!("diffusion: {}/{} {path} waits for a person: {why:?}", consumer.owner, consumer.name),
1261 
        }
1262 
    })
1263 
}
1264
1265 
/// Try every open bump of `consumer` (its setting turned to `auto`).
1266 
pub async fn land_ready(app: &Shared, consumer: &RepoRow) {
1267 
    let paths: Vec<String> = sqlx::query_scalar("SELECT DISTINCT path FROM diffusion_pins WHERE consumer_id = ?")
1268 
        .bind(consumer.id)
1269 
        .fetch_all(&app.db)
1270 
        .await
1271 
        .unwrap_or_default();
1272 
    for path in paths {
1273 
        tokio::spawn(try_land(app.clone(), consumer.clone(), path));
1274 
    }
1275 
}
1276
1277 
/// Whether `upstream` has no checks of its own: Flow is off for it, or
1278 
/// its default bookmark has never had a run. Its moves are then
1279 
/// "unverified" (amber), never green.
1280 
pub async fn unverified(app: &Shared, upstream: &RepoRow) -> bool {
1281 
    if !crate::flow::enabled(app, upstream) {
1282 
        return true;
1283 
    }
1284 
    let runs: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM flow_runs WHERE repo_id = ? AND bookmark = ?")
1285 
        .bind(upstream.id)
1286 
        .bind(&upstream.default_bookmark)
1287 
        .fetch_one(&app.db)
1288 
        .await
1289 
        .unwrap_or(0);
1290 
    runs == 0
1291 
}
1292
1293 
/// The repository on this hub with Oxygen id `id` (hex).
1294 
async fn repo_by_oxy_id(app: &Shared, id: &str) -> Option<RepoRow> {
1295 
    for row in db::repos::list(&app.db).await.ok()? {
1296 
        let Some(name) = RepoName::new(&row.owner, &row.name) else { continue };
1297 
        if app.repos.read(&name, |r, _| Ok(r.id()?)).await.ok().map(hex::encode).as_deref() == Some(id) {
1298 
            return Some(row);
1299 
        }
1300 
    }
1301 
    None
1302 
}
1303
1304 
/// The newest state of `upstream`'s default bookmark whose run passed.
1305 
async fn last_green(app: &Shared, upstream: &RepoRow) -> Option<Vec<ChangeHash>> {
1306 
    let heads: String =
1307 
        sqlx::query_scalar("SELECT heads FROM flow_runs WHERE repo_id = ? AND bookmark = ? AND status = 'passed' ORDER BY id DESC LIMIT 1")
1308 
            .bind(upstream.id)
1309 
            .bind(&upstream.default_bookmark)
1310 
            .fetch_optional(&app.db)
1311 
            .await
1312 
            .ok()
1313 
            .flatten()?;
1314 
    let h: Vec<ChangeHash> = heads.split('+').filter_map(ChangeHash::from_hex).collect();
1315 
    (!h.is_empty()).then_some(h)
1316 
}
1317
1318 
/// Bring `consumer` up to date with the upstreams it pins, at once: when
1319 
/// it turns Diffusion on, it is bumped to each upstream's newest state
1320 
/// that passed (or simply its newest, when it follows every move).
1321 
pub async fn catch_up(app: &Shared, consumer: &RepoRow) {
1322 
    let upstreams: Vec<String> = sqlx::query_scalar("SELECT DISTINCT upstream FROM diffusion_pins WHERE consumer_id = ?")
1323 
        .bind(consumer.id)
1324 
        .fetch_all(&app.db)
1325 
        .await
1326 
        .unwrap_or_default();
1327 
    for id in upstreams {
1328 
        let Some(upstream) = repo_by_oxy_id(app, &id).await else { continue };
1329 
        if let Some(h) = last_green(app, &upstream).await {
1330 
            propagate_to(app, &upstream, h, true, Some(consumer.id)).await;
1331 
        }
1332 
        let Some(name) = RepoName::new(&upstream.owner, &upstream.name) else { continue };
1333 
        let b = upstream.default_bookmark.clone();
1334 
        if let Ok(h) = app.repos.read(&name, move |r, _| heads_of(r, &b)).await {
1335 
            propagate_to(app, &upstream, h, false, Some(consumer.id)).await;
1336 
        }
1337 
    }
1338 
}
1339
1340 
/// When a consumer takes an upstream's moves (§4.2).
1341 
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
1342 
#[serde(rename_all = "lowercase")]
1343 
pub enum Follow {
1344 
    Off,
1345 
    /// When the upstream state's checks pass.
1346 
    Green,
1347 
    /// On every move.
1348 
    Any,
1349 
}
1350
1351 
impl Follow {
1352 
    pub fn parse(s: &str) -> Option<Follow> {
1353 
        match s.trim() {
1354 
            "off" => Some(Follow::Off),
1355 
            "green" => Some(Follow::Green),
1356 
            "any" => Some(Follow::Any),
1357 
            _ => None,
1358 
        }
1359 
    }
1360
1361 
    pub fn as_str(self) -> &'static str {
1362 
        match self {
1363 
            Follow::Off => "off",
1364 
            Follow::Green => "green",
1365 
            Follow::Any => "any",
1366 
        }
1367 
    }
1368 
}
1369
1370 
/// How one consumer follows one subrepo, from its `.oxymodules` entry, its
1371 
/// repository setting and the site's default, in that order.
1372 
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
1373 
pub struct Rule {
1374 
    pub follow: Follow,
1375 
    /// Land a bump itself once both halves are green (`land = auto`, or
1376 
    /// the repository's setting); otherwise it only proposes.
1377 
    pub land_auto: bool,
1378 
    /// The bookmark to bump, when `.oxymodules` names one (else learned).
1379 
    pub target: Option<String>,
1380 
    /// Paths the consumer watches: upstream changes touching them are
1381 
    /// flagged in a bump's changelog.
1382 
    pub watch: Vec<String>,
1383 
    /// Where `follow` came from: `.oxymodules`, `setting` or `site`.
1384 
    pub from: &'static str,
1385 
}
1386
1387 
/// The Diffusion keys of `path`'s entry in `.oxymodules` text: `follow`,
1388 
/// `land`, `target` and `watch` (comma-separated, or a TOML list).
1389 
pub fn module_keys(text: &str, path: &str) -> (Option<Follow>, Option<bool>, Option<String>, Vec<String>) {
1390 
    let (mut follow, mut land, mut target, mut watch) = (None, None, None, Vec::new());
1391 
    let mut here = false;
1392 
    for line in text.lines() {
1393 
        let line = line.trim();
1394 
        if line.is_empty() || line.starts_with('#') {
1395 
            continue;
1396 
        }
1397 
        if let Some(inner) = line.strip_prefix('[').and_then(|l| l.strip_suffix(']')) {
1398 
            here = inner.trim().trim_matches('"').trim_end_matches('/') == path.trim_end_matches('/');
1399 
            continue;
1400 
        }
1401 
        let (true, Some((k, v))) = (here, line.split_once('=')) else { continue };
1402 
        let v = v.trim();
1403 
        let unquoted = v.trim_matches('"').trim();
1404 
        match k.trim() {
1405 
            "follow" => follow = Follow::parse(unquoted),
1406 
            "land" => land = Some(unquoted == "auto"),
1407 
            "target" if !unquoted.is_empty() => target = Some(unquoted.to_string()),
1408 
            "watch" => {
1409 
                watch = v
1410 
                    .trim_start_matches('[')
1411 
                    .trim_end_matches(']')
1412 
                    .split(',')
1413 
                    .map(|w| w.trim().trim_matches('"').trim().to_string())
1414 
                    .filter(|w| !w.is_empty())
1415 
                    .collect()
1416 
            }
1417 
            _ => {}
1418 
        }
1419 
    }
1420 
    (follow, land, target, watch)
1421 
}
1422
1423 
/// How `row` follows its subrepo at `path`, given its `.oxymodules` text.
1424 
pub fn rule(site_default: &str, row: &RepoRow, oxymodules: Option<&str>, path: &str) -> Rule {
1425 
    let (follow, land, target, watch) = oxymodules.map(|t| module_keys(t, path)).unwrap_or_default();
1426 
    let (follow, from) = match follow {
1427 
        Some(f) => (f, ".oxymodules"),
1428 
        None => match Follow::parse(&row.diffusion) {
1429 
            Some(f) => (f, "setting"),
1430 
            None => (Follow::parse(site_default).unwrap_or(Follow::Off), "site"),
1431 
        },
1432 
    };
1433 
    // Landing by itself: the entry's word, else the setting; never by default.
1434 
    let land_auto = land.unwrap_or(row.diffusion_land == "auto");
1435 
    Rule { follow, land_auto, target, watch, from }
1436 
}
1437
1438 
#[cfg(test)]
1439 
mod tests {
1440 
    use super::*;
1441
1442 
    const MODULES: &str = r#"
1443 
[rstar-api]
1444 
url = https://oxygenhub.io/doomsdayone/rstar-api
1445 
follow = green
1446 
land = "propose"
1447 
target = develop
1448 
watch = ["src/x16/irq*", "src/x16/frame*"]
1449
1450 
[other]
1451 
url = https://example/other
1452 
follow = any
1453 
"#;
1454
1455 
    fn row(diffusion: &str) -> RepoRow {
1456 
        RepoRow {
1457 
            id: 1,
1458 
            owner: "o".into(),
1459 
            name: "n".into(),
1460 
            description: String::new(),
1461 
            default_bookmark: "main".into(),
1462 
            visibility: "public".into(),
1463 
            created_ms: 0,
1464 
            updated_ms: 0,
1465 
            git_mirror: false,
1466 
            git_mirror_state: String::new(),
1467 
            forked_from: None,
1468 
            topics: String::new(),
1469 
            flow: String::new(),
1470 
            diffusion: diffusion.into(),
1471 
            ..Default::default()
1472 
        }
1473 
    }
1474
1475 
    #[test]
1476 
    fn landing_by_itself_is_chosen_per_repository_and_per_entry() {
1477 
        let mut r = row("green");
1478 
        assert!(!rule("off", &r, None, "rstar-api").land_auto, "never by default");
1479 
        r.diffusion_land = "auto".into();
1480 
        assert!(rule("off", &r, None, "rstar-api").land_auto, "the setting");
1481 
        assert!(!rule("off", &r, Some(MODULES), "rstar-api").land_auto, "the entry's `land = propose` wins");
1482 
        assert!(rule("off", &r, Some(MODULES), "other").land_auto, "an entry without `land` takes the setting");
1483 
        r.diffusion_land = String::new();
1484 
        let auto = MODULES.replace(r#"land = "propose""#, "land = auto");
1485 
        assert!(rule("off", &r, Some(&auto), "rstar-api").land_auto);
1486 
    }
1487
1488 
    #[test]
1489 
    fn watch_patterns_match_paths() {
1490 
        assert!(watched("src/x16/irq*", "src/x16/irq.s"));
1491 
        assert!(watched("src/x16/irq*", "src/x16/irq_wait.c"));
1492 
        assert!(!watched("src/x16/irq*", "src/x16/sub/irq.s"), "one segment");
1493 
        assert!(watched("src/**/frame*.c", "src/x16/floor/frame_wait.c"));
1494 
        assert!(watched("src/x16", "src/x16/bank.s"), "a directory");
1495 
        assert!(!watched("src/x16", "src/x160/bank.s"));
1496 
        assert!(watched("include/r?m.h", "include/rim.h"));
1497 
        assert!(!watched("", "anything"));
1498 
    }
1499
1500 
    #[test]
1501 
    fn oxymodules_keys_are_read_per_entry() {
1502 
        let (f, land, target, watch) = module_keys(MODULES, "rstar-api");
1503 
        assert_eq!((f, land, target.as_deref()), (Some(Follow::Green), Some(false), Some("develop")));
1504 
        assert_eq!(watch, vec!["src/x16/irq*", "src/x16/frame*"]);
1505 
        assert_eq!(module_keys(MODULES, "other").0, Some(Follow::Any));
1506 
        assert_eq!(module_keys(MODULES, "missing"), (None, None, None, vec![]));
1507 
        assert_eq!(module_keys("[a]\nwatch = x, y\n", "a").3, vec!["x", "y"]);
1508 
    }
1509
1510 
    #[test]
1511 
    fn oxymodules_wins_then_the_setting_then_the_site() {
1512 
        let r = rule("off", &row("any"), Some(MODULES), "rstar-api");
1513 
        assert_eq!((r.follow, r.from), (Follow::Green, ".oxymodules"));
1514 
        let r = rule("off", &row("any"), Some("[rstar-api]\nurl = x\n"), "rstar-api");
1515 
        assert_eq!((r.follow, r.from), (Follow::Any, "setting"));
1516 
        let r = rule("green", &row(""), None, "rstar-api");
1517 
        assert_eq!((r.follow, r.from), (Follow::Green, "site"));
1518 
        assert_eq!(rule("nonsense", &row(""), None, "x").follow, Follow::Off);
1519 
    }
1520 
}
1521
1522 
#[cfg(test)]
1523 
mod message_tests {
1524 
    use super::bump_message;
1525
1526 
    #[test]
1527 
    fn a_bump_of_many_heads_has_a_short_title_and_the_whole_address_below() {
1528 
        let heads: Vec<String> = (0..20).map(|i| format!("{i:02x}a6c9d4746a")).collect();
1529 
        let m = bump_message("rstar-api", "doomsdayone/rstar-api", &heads);
1530 
        let (title, body) = m.split_once("\n\n").unwrap();
1531 
        assert_eq!(title, "Diffusion: move rstar-api to doomsdayone/rstar-api 00a6c9d4746a (+19 heads)");
1532 
        assert_eq!(body, format!("Pins doomsdayone/rstar-api at {}", heads.join("+")));
1533 
        assert_eq!(bump_message("api", "andy/api", &heads[..1]), "Diffusion: move api to andy/api 00a6c9d4746a");
1534 
    }
1535 
}