OXYGENHUB
# 5d90e0e045e2 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
f677d627 Andy McCall ↶ 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(
5d90e0e0 Andy McCall 26 
        "SELECT target FROM proposals WHERE repo_id = $1 AND state = 'landed' AND landed_ms >= $2
f677d627 Andy McCall ↶ 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]) {
5d54cf74 Andy McCall ↶ 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 
    };
f677d627 Andy McCall ↶ 117 
    let now = db::now_ms();
757195ad Andy McCall ↶ 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 
    }
5d54cf74 Andy McCall ↶ 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}"))?;
5d90e0e0 Andy McCall 127 
    sqlx::query("DELETE FROM diffusion_pins WHERE consumer_id = $1 AND bookmark = $2")
f677d627 Andy McCall ↶ 128 
        .bind(row.id)
129 
        .bind(bookmark)
130 
        .execute(&mut *tx)
131 
        .await
5d54cf74 Andy McCall ↶ 132 
        .map_err(|e| format!("delete: {e}"))?;
757195ad Andy McCall ↶ 133 
    for ((path, id, pinned), upstream_row) in pins.iter().zip(&upstreams) {
f677d627 Andy McCall ↶ 134 
        let heads: Vec<String> = pinned.iter().map(|h| h.to_hex()).collect();
5d54cf74 Andy McCall ↶ 135 
        sqlx::query(
5d90e0e0 Andy McCall 136 
            "INSERT INTO diffusion_pins (consumer_id, bookmark, path, upstream, heads, updated_ms, upstream_repo_id) VALUES ($1, $2, $3, $4, $5, $6, $7)",
757195ad Andy McCall ↶ 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
5d54cf74 Andy McCall ↶ 147 
        .map_err(|e| format!("insert {path}: {e}"))?;
f677d627 Andy McCall ↶ 148 
    }
5d90e0e0 Andy McCall 149 
    sqlx::query("INSERT OR REPLACE INTO diffusion_indexed (consumer_id, bookmark, heads, updated_ms) VALUES ($1, $2, $3, $4)")
0e04dfc2 Andy McCall ↶ 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}"))?;
5d54cf74 Andy McCall ↶ 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);
f677d627 Andy McCall ↶ 161 
    }
5d54cf74 Andy McCall ↶ 162 
    Ok(())
f677d627 Andy McCall ↶ 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 
    }
757195ad Andy McCall ↶ 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));
f677d627 Andy McCall ↶ 183 
}
184
757195ad Andy McCall ↶ 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
f677d627 Andy McCall ↶ 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;
0e04dfc2 Andy McCall ↶ 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 };
5d90e0e0 Andy McCall 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();
0e04dfc2 Andy McCall ↶ 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;
f677d627 Andy McCall ↶ 262 
            }
0e04dfc2 Andy McCall ↶ 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;
f677d627 Andy McCall ↶ 316 
        }
317 
    }
0e04dfc2 Andy McCall ↶ 318 
    repairs
f677d627 Andy McCall ↶ 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>,
757195ad Andy McCall ↶ 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,
73ef9f76 Andy McCall ↶ 349 
    /// The pill's two halves: the upstream, then this consumer.
350 
    pub upstream: Half,
351 
    pub downstream: Half,
757195ad Andy McCall ↶ 352 
}
353
eb2a3502 Andy McCall ↶ 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
73ef9f76 Andy McCall ↶ 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
757195ad Andy McCall ↶ 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,
f677d627 Andy McCall ↶ 430 
}
431
757195ad Andy McCall ↶ 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
f677d627 Andy McCall ↶ 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(
5d90e0e0 Andy McCall 477 
        "SELECT consumer_id, bookmark, path, heads FROM diffusion_pins WHERE upstream = $1 ORDER BY consumer_id, bookmark, path",
f677d627 Andy McCall ↶ 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();
757195ad Andy McCall ↶ 485 
    let unverified = unverified(app, row).await;
f677d627 Andy McCall ↶ 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) {
757195ad Andy McCall ↶ 489 
            continue;
490 
        }
491 
        // One row per consumer and subrepo: the bookmark its work lands on.
492 
        if bookmark != target(&app.db, &c).await {
f677d627 Andy McCall ↶ 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<_>>());
757195ad Andy McCall ↶ 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)> =
5d90e0e0 Andy McCall 507 
            sqlx::query_as("SELECT number, status FROM flow_runs WHERE repo_id = $1 AND bookmark = $2 ORDER BY id DESC LIMIT 1")
757195ad Andy McCall ↶ 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,
73ef9f76 Andy McCall ↶ 527 
            upstream: Half { class: "df-grey", word: "", title: String::new() },
528 
            downstream: Half { class: "df-grey", word: "", title: String::new() },
757195ad Andy McCall ↶ 529 
        });
f677d627 Andy McCall ↶ 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?;
73ef9f76 Andy McCall ↶ 546 
    let up_run: Option<String> =
5d90e0e0 Andy McCall 547 
        sqlx::query_scalar("SELECT status FROM flow_runs WHERE repo_id = $1 AND bookmark = $2 ORDER BY id DESC LIMIT 1")
73ef9f76 Andy McCall ↶ 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);
f677d627 Andy McCall ↶ 555 
    for (c, b) in out.iter_mut().zip(behind) {
556 
        c.behind = b;
73ef9f76 Andy McCall ↶ 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));
f677d627 Andy McCall ↶ 561 
    }
562 
    Ok(out)
757195ad Andy McCall ↶ 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> =
5d90e0e0 Andy McCall 569 
        sqlx::query_scalar("SELECT DISTINCT upstream_repo_id FROM diffusion_pins WHERE consumer_id = $1 AND upstream_repo_id IS NOT NULL")
757195ad Andy McCall ↶ 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
f677d627 Andy McCall ↶ 603 
}
e7158350 Andy McCall ↶ 604
1d634f69 Andy McCall ↶ 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.
f01cb865 Andy McCall ↶ 614 
    Moved { bookmark: String, change: ChangeHash, heads: Vec<ChangeHash>, from: Vec<ChangeHash>, state: String },
1d634f69 Andy McCall ↶ 615 
    /// The pin already holds the upstream state.
616 
    UpToDate,
617 
    /// The bump bookmark already moves the pin to that state.
9aaa88b2 Andy McCall ↶ 618 
    Already { from: Vec<ChangeHash>, to: Vec<ChangeHash> },
7a14bdae Andy McCall ↶ 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,
1d634f69 Andy McCall ↶ 624 
    /// The consumer has no Oxygen pin of that upstream at `path` on its
625 
    /// target bookmark.
626 
    NoPin,
627 
}
628
7a14bdae Andy McCall ↶ 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> {
1d634f69 Andy McCall ↶ 649 
    app.repos
650 
        .read(upstream, move |r, cache| {
7a14bdae Andy McCall ↶ 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);
1d634f69 Andy McCall ↶ 663 
            }
7a14bdae Andy McCall ↶ 664 
            let added = b.changes.len() - o.changes.len();
665 
            Ok(if added == 0 { Plan::UpToDate } else { Plan::To(heads, added) })
1d634f69 Andy McCall ↶ 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) };
7a14bdae Andy McCall ↶ 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 
    };
1d634f69 Andy McCall ↶ 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 {
9aaa88b2 Andy McCall ↶ 723 
            return Ok(Bumped::Already { from: old, to: new_heads });
1d634f69 Andy McCall ↶ 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 };
a7e6e8e6 Andy McCall ↶ 732 
    let old = old.clone();
1d634f69 Andy McCall ↶ 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;
f01cb865 Andy McCall ↶ 751 
            let state = update.new_state.to_hex();
1d634f69 Andy McCall ↶ 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 
            }
f01cb865 Andy McCall ↶ 759 
            Ok(Bumped::Moved { bookmark, change: hash, heads, from: old, state })
1d634f69 Andy McCall ↶ 760 
        })
761 
        .await
762 
}
763
a7e6e8e6 Andy McCall ↶ 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
c4e432da Andy McCall ↶ 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() } };
a7e6e8e6 Andy McCall ↶ 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)?);
c4e432da Andy McCall ↶ 813 
            let mut added: Vec<(u64, ChangeHash, String, String, Vec<String>)> = Vec::new();
a7e6e8e6 Andy McCall ↶ 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();
c4e432da Andy McCall ↶ 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));
a7e6e8e6 Andy McCall ↶ 828 
            }
829 
            added.sort_by_key(|a| std::cmp::Reverse(a.0));
c4e432da Andy McCall ↶ 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 
                    )
a7e6e8e6 Andy McCall ↶ 848 
                })
c4e432da Andy McCall ↶ 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 })
a7e6e8e6 Andy McCall ↶ 852 
        })
853 
        .await
c4e432da Andy McCall ↶ 854 
        .unwrap_or(Changelog { lines: Vec::new(), watched: Vec::new() })
a7e6e8e6 Andy McCall ↶ 855 
}
856
9aaa88b2 Andy McCall ↶ 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.
a7e6e8e6 Andy McCall ↶ 873 
pub async fn propose(
874 
    app: &Shared,
875 
    consumer: &RepoRow,
876 
    target: &str,
877 
    path: &str,
878 
    upstream: &RepoRow,
9aaa88b2 Andy McCall ↶ 879 
    (from, to): (Vec<ChangeHash>, Vec<ChangeHash>),
f4d6ef41 Andy McCall ↶ 880 
    (why, watch, lands): (Why, Vec<String>, bool),
a7e6e8e6 Andy McCall ↶ 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?;
c4e432da Andy McCall ↶ 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 
    };
a7e6e8e6 Andy McCall ↶ 897 
    let slug = format!("{}/{}", upstream.owner, upstream.name);
898 
    let short: Vec<String> = to.iter().map(|h| h.to_hex()[..12].to_string()).collect();
1b6536f0 Andy McCall ↶ 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})");
9aaa88b2 Andy McCall ↶ 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 
    };
a7e6e8e6 Andy McCall ↶ 912 
    let body = format!(
f4d6ef41 Andy McCall ↶ 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). {}",
6ae3fa0a Andy McCall ↶ 914 
        crate::web::state_label(&short.join("+")),
f4d6ef41 Andy McCall ↶ 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 
        }
a7e6e8e6 Andy McCall ↶ 921 
    );
922 
    let open = pdb::list(&app.db, consumer.id, &["open", "draft"]).await.ok()?.into_iter().find(|p| p.source_bookmark == bookmark);
9aaa88b2 Andy McCall ↶ 923 
    let (number, id) = match open {
a7e6e8e6 Andy McCall ↶ 924 
        Some(p) => {
9aaa88b2 Andy McCall ↶ 925 
            if p.title != title || p.body != body {
926 
                pdb::edit(&app.db, p.id, &title, &body, target).await.ok()?;
6ae3fa0a Andy McCall ↶ 927 
                pdb::add_event(&app.db, p.id, ACCOUNT, "updated", &crate::web::state_label(&short.join("+"))).await.ok();
9aaa88b2 Andy McCall ↶ 928 
            }
929 
            (p.number, p.id)
a7e6e8e6 Andy McCall ↶ 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();
bff9106e Andy McCall ↶ 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;
a7e6e8e6 Andy McCall ↶ 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 
            }
9aaa88b2 Andy McCall ↶ 972 
            (p.number, p.id)
a7e6e8e6 Andy McCall ↶ 973 
        }
974 
    };
9aaa88b2 Andy McCall ↶ 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 
    }
a7e6e8e6 Andy McCall ↶ 1000 
    Some(number)
1001 
}
1002
f01cb865 Andy McCall ↶ 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
a7e6e8e6 Andy McCall ↶ 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) {
3776b70d Andy McCall ↶ 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>) {
a7e6e8e6 Andy McCall ↶ 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 };
5d90e0e0 Andy McCall 1047 
    let rows: Vec<(i64, String, String)> = sqlx::query_as("SELECT consumer_id, bookmark, path FROM diffusion_pins WHERE upstream = $1")
a7e6e8e6 Andy McCall ↶ 1048 
        .bind(hex::encode(id))
1049 
        .fetch_all(&app.db)
1050 
        .await
1051 
        .unwrap_or_default();
9aaa88b2 Andy McCall ↶ 1052 
    let unverified = unverified(app, upstream).await;
a7e6e8e6 Andy McCall ↶ 1053 
    for (consumer_id, indexed, path) in rows {
3776b70d Andy McCall ↶ 1054 
        if only.is_some_and(|o| o != consumer_id) {
1055 
            continue;
1056 
        }
a7e6e8e6 Andy McCall ↶ 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 
        }
9aaa88b2 Andy McCall ↶ 1066 
        // An upstream with no checks never turns green: its moves count
1067 
        // for green followers too, marked amber (unverified).
a7e6e8e6 Andy McCall ↶ 1068 
        let go = match r.follow {
1069 
            Follow::Off => false,
9aaa88b2 Andy McCall ↶ 1070 
            Follow::Green => green || unverified,
a7e6e8e6 Andy McCall ↶ 1071 
            Follow::Any => !green,
1072 
        };
9aaa88b2 Andy McCall ↶ 1073 
        let why = if green {
1074 
            Why::Green
1075 
        } else if unverified {
1076 
            Why::Unverified
1077 
        } else {
1078 
            Why::Moved
1079 
        };
a7e6e8e6 Andy McCall ↶ 1080 
        if !go {
1081 
            continue;
1082 
        }
1083 
        match bump(app, &consumer, &target, &path, upstream, heads.clone()).await {
f01cb865 Andy McCall ↶ 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));
f4d6ef41 Andy McCall ↶ 1087 
                let n = propose(app, &consumer, &target, &path, upstream, (from, heads.clone()), (why, r.watch.clone(), r.land_auto)).await;
7e01a766 Andy McCall ↶ 1088 
                app.events.live.publish(&crate::events::topic::diffusion(consumer.id), "bump");
1089 
                app.events.live.publish(&crate::events::topic::diffusion(upstream.id), "bump");
a7e6e8e6 Andy McCall ↶ 1090 
                tracing::info!(
1091 
                    "diffusion: {}/{} {path} bumped to {}/{} (proposal {n:?})",
1092 
                    consumer.owner,
1093 
                    consumer.name,
1094 
                    upstream.owner,
1095 
                    upstream.name
1096 
                );
9aaa88b2 Andy McCall ↶ 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.
f4d6ef41 Andy McCall ↶ 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 
                }
a7e6e8e6 Andy McCall ↶ 1107 
            }
7a14bdae Andy McCall ↶ 1108 
            Ok(other) => tracing::info!("diffusion: {}/{} {path}: no bump ({other:?})", consumer.owner, consumer.name),
a7e6e8e6 Andy McCall ↶ 1109 
            Err(e) => tracing::warn!("diffusion: bump {}/{} {path}: {e}", consumer.owner, consumer.name),
3776b70d Andy McCall ↶ 1110 
        }
1111 
    }
f4d6ef41 Andy McCall ↶ 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),
9aaa88b2 Andy McCall ↶ 1129 
}
1130
f4d6ef41 Andy McCall ↶ 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(
5d90e0e0 Andy McCall 1155 
        "SELECT upstream_repo_id FROM diffusion_pins WHERE consumer_id = $1 AND path = $2 AND upstream_repo_id IS NOT NULL LIMIT 1",
f4d6ef41 Andy McCall ↶ 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(
5d90e0e0 Andy McCall 1202 
        "SELECT heads FROM flow_runs WHERE repo_id = $1 AND bookmark = $2 AND status = 'passed' ORDER BY id DESC LIMIT 100",
f4d6ef41 Andy McCall ↶ 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)> =
5d90e0e0 Andy McCall 1213 
        sqlx::query_as("SELECT status, heads FROM flow_runs WHERE repo_id = $1 AND bookmark = $2 ORDER BY id DESC LIMIT 1")
f4d6ef41 Andy McCall ↶ 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) {
5d90e0e0 Andy McCall 1259 
    let paths: Vec<String> = sqlx::query_scalar("SELECT DISTINCT path FROM diffusion_pins WHERE consumer_id = $1")
f4d6ef41 Andy McCall ↶ 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
9aaa88b2 Andy McCall ↶ 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 
    }
5d90e0e0 Andy McCall 1276 
    let runs: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM flow_runs WHERE repo_id = $1 AND bookmark = $2")
9aaa88b2 Andy McCall ↶ 1277 
        .bind(upstream.id)
1278 
        .bind(&upstream.default_bookmark)
1279 
        .fetch_one(&app.db)
1280 
        .await
1281 
        .unwrap_or(0);
1282 
    runs == 0
3776b70d Andy McCall ↶ 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>> {
5d90e0e0 Andy McCall 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()?;
3776b70d Andy McCall ↶ 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) {
5d90e0e0 Andy McCall 1315 
    let upstreams: Vec<String> = sqlx::query_scalar("SELECT DISTINCT upstream FROM diffusion_pins WHERE consumer_id = $1")
3776b70d Andy McCall ↶ 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;
a7e6e8e6 Andy McCall ↶ 1329 
        }
1330 
    }
1331 
}
1332
e7158350 Andy McCall ↶ 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,
eb2a3502 Andy McCall ↶ 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",
e7158350 Andy McCall ↶ 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,
f4d6ef41 Andy McCall ↶ 1368 
    /// Land a bump itself once both halves are green (`land = auto`, or
1369 
    /// the repository's setting); otherwise it only proposes.
e7158350 Andy McCall ↶ 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;
5360cea3 Andy McCall ↶ 1385 
    let mut lines = text.lines();
1386 
    while let Some(line) = lines.next() {
e7158350 Andy McCall ↶ 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 };
5360cea3 Andy McCall ↶ 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();
e7158350 Andy McCall ↶ 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 
    };
f4d6ef41 Andy McCall ↶ 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 }
e7158350 Andy McCall ↶ 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 
        }
f4d6ef41 Andy McCall ↶ 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);
c4e432da Andy McCall ↶ 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"));
e7158350 Andy McCall ↶ 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"]);
5360cea3 Andy McCall ↶ 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");
e7158350 Andy McCall ↶ 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 
}