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