//! Diffusion (#148, docs/diffusion.md): an upstream's changes reach the

//! repositories that pin it as a subrepo.

//!

//! D1 starts with the reverse pin index (§4.1): when a push moves a

//! repository's default bookmark, or the bookmark its work lands on, each

//! Oxygen subrepo pin in the new state is recorded. That answers "who

//! consumes this repository, at which state, and how far behind", which

//! an upstream's page shows as Used by.

use crate::auth::roles;

use crate::db::{self, accounts::Account, repos::RepoRow, Pool};

use crate::repos::RepoName;

use crate::Shared;

use oxy_core::hash::ChangeHash;

use serde::Serialize;

/// How far back landed proposals teach where a repository's work lands.

pub const LEARN_DAYS: i64 = 90;

/// The bookmark a repository's work lands on (§4.3): the most common

/// target of its proposals landed in the last `LEARN_DAYS`, else its

/// default bookmark.

pub async fn target(pool: &Pool, row: &RepoRow) -> String {

    let since = db::now_ms() - LEARN_DAYS * 24 * 3600 * 1000;

    let learned: Option<String> = sqlx::query_scalar(

        "SELECT target FROM proposals WHERE repo_id = ? AND state = 'landed' AND landed_ms >= ?

         GROUP BY target ORDER BY COUNT(*) DESC, MAX(landed_ms) DESC LIMIT 1",

    )

    .bind(row.id)

    .bind(since)

    .fetch_optional(pool)

    .await

    .ok()

    .flatten();

    learned.filter(|t| !t.is_empty()).unwrap_or_else(|| row.default_bookmark.clone())

}

/// The bookmarks whose pins are indexed: the default and the learned

/// target (one when they are the same).

pub async fn followed_bookmarks(pool: &Pool, row: &RepoRow) -> Vec<String> {

    let mut out = vec![row.default_bookmark.clone()];

    let t = target(pool, row).await;

    if !out.contains(&t) {

        out.push(t);

    }

    out

}

/// Record `row`'s pins on `bookmark` at `heads`, replacing what was there.

pub async fn record(app: &Shared, row: &RepoRow, bookmark: &str, heads: &[ChangeHash]) {

    // A failure is logged and tried once more after a moment (oxygenhub#210):

    // an index left behind shows a consumer behind with no bump.

    let repo = format!("{}/{}", row.owner, row.name);

    for attempt in 1..=2 {

        match try_record(app, row, bookmark, heads).await {

            Ok(()) => {

                if attempt > 1 {

                    tracing::info!("diffusion: indexed {repo} {bookmark} on the second try");

                }

                return;

            }

            Err(why) if attempt == 1 => {

                tracing::warn!("diffusion: indexing {repo} {bookmark} failed ({why}); trying again");

                tokio::time::sleep(RETRY_AFTER).await;

            }

            Err(why) => {

                FAILURES.fetch_add(1, std::sync::atomic::Ordering::Relaxed);

                tracing::error!("diffusion: indexing {repo} {bookmark} failed again ({why}); its pins stay as they were");

            }

        }

    }

}

/// How long [`record`] waits before its second try.

const RETRY_AFTER: std::time::Duration = std::time::Duration::from_millis(250);

/// Indexing failures that the retry did not mend, since the hub started.

pub static FAILURES: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);

/// Failures to force, per repository (tests only: [`fail_next_records`]).

static FORCED: std::sync::Mutex<Option<std::collections::HashMap<String, u32>>> = std::sync::Mutex::new(None);

/// Make the next `n` index writes for a repository fail, as a busy

/// database would (for tests of the retry).

#[doc(hidden)]

pub fn fail_next_records(app: &Shared, repo_id: i64, n: u32) {

    FORCED.lock().unwrap_or_else(|p| p.into_inner()).get_or_insert_with(Default::default).insert(key(app, repo_id), n);

}

/// Forced failures not yet used (tests only).

#[doc(hidden)]

pub fn forced_left(app: &Shared, repo_id: i64) -> u32 {

    FORCED.lock().unwrap_or_else(|p| p.into_inner()).as_ref().and_then(|m| m.get(&key(app, repo_id)).copied()).unwrap_or(0)

}

fn forced(app: &Shared, repo_id: i64) -> bool {

    let mut forced = FORCED.lock().unwrap_or_else(|p| p.into_inner());

    match forced.as_mut().and_then(|m| m.get_mut(&key(app, repo_id))) {

        Some(n) if *n > 0 => {

            *n -= 1;

            true

        }

        _ => false,

    }

}

/// One try at replacing a bookmark's indexed pins; which step failed and

/// why. A state that cannot be read leaves the index as it was rather than

/// emptying it.

async fn try_record(app: &Shared, row: &RepoRow, bookmark: &str, heads: &[ChangeHash]) -> Result<(), String> {

    let Some(name) = RepoName::new(&row.owner, &row.name) else { return Ok(()) };

    let pins = if heads.is_empty() {

        Vec::new()

    } else {

        crate::flow::pins_in(app, name, heads.to_vec()).await.map_err(|e| format!("reading the pins: {e}"))?

    };

    let now = db::now_ms();

    // The upstream's repository on this hub, when it is one.

    let mut upstreams = Vec::new();

    for (_, id, _) in &pins {

        upstreams.push(repo_by_oxy_id(app, &hex::encode(id)).await.map(|r| r.id));

    }

    if forced(app, row.id) {

        return Err("begin: forced for a test".into());

    }

    let mut tx = app.db.begin().await.map_err(|e| format!("begin: {e}"))?;

    sqlx::query("DELETE FROM diffusion_pins WHERE consumer_id = ? AND bookmark = ?")

        .bind(row.id)

        .bind(bookmark)

        .execute(&mut *tx)

        .await

        .map_err(|e| format!("delete: {e}"))?;

    for ((path, id, pinned), upstream_row) in pins.iter().zip(&upstreams) {

        let heads: Vec<String> = pinned.iter().map(|h| h.to_hex()).collect();

        sqlx::query(

            "INSERT INTO diffusion_pins (consumer_id, bookmark, path, upstream, heads, updated_ms, upstream_repo_id) VALUES (?, ?, ?, ?, ?, ?, ?)",

        )

        .bind(row.id)

        .bind(bookmark)

        .bind(path)

        .bind(hex::encode(id))

        .bind(heads.join("+"))

        .bind(now)

        .bind(upstream_row)

        .execute(&mut *tx)

        .await

        .map_err(|e| format!("insert {path}: {e}"))?;

    }

    tx.commit().await.map_err(|e| format!("commit: {e}"))?;

    for u in upstreams.iter().flatten() {

        mark(app, row.id);

        mark(app, *u);

    }

    Ok(())

}

/// After a push: index the pins of each moved bookmark that is followed.

pub async fn after_push(app: &Shared, row: &RepoRow, moved: &[(String, String, Vec<ChangeHash>)]) {

    let followed = followed_bookmarks(&app.db, row).await;

    for (bookmark, _, heads) in moved {

        if followed.contains(bookmark) {

            record(app, row, bookmark, heads).await;

        }

    }

}

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

fn key(app: &Shared, repo_id: i64) -> String {

    format!("{}#{repo_id}", app.repos.root().display())

}

fn mark(app: &Shared, repo_id: i64) {

    INVOLVED.lock().unwrap_or_else(|p| p.into_inner()).get_or_insert_with(Default::default).insert(key(app, repo_id));

}

/// Whether a repository takes part in Diffusion, as an upstream or a

/// consumer on this hub: its Diffusion tab shows (#150).

pub fn involved(app: &Shared, repo_id: i64) -> bool {

    INVOLVED.lock().unwrap_or_else(|p| p.into_inner()).as_ref().is_some_and(|s| s.contains(&key(app, repo_id)))

}

/// Index every repository's followed bookmarks, for a hub that had no

/// index before (run once at start, in the background).

pub async fn backfill(app: &Shared) {

    let Ok(rows) = db::repos::list(&app.db).await else { return };

    for row in rows {

        let Some(name) = RepoName::new(&row.owner, &row.name) else { continue };

        for bookmark in followed_bookmarks(&app.db, &row).await {

            let b = bookmark.clone();

            let heads = app.repos.read(&name, move |r, _| heads_of(r, &b)).await;

            if let Ok(heads) = heads {

                record(app, &row, &bookmark, &heads).await;

            }

        }

    }

}

/// A bookmark's heads as the repository has them now (not a cached

/// list, which a push refreshes a moment later); empty when it has none.

fn heads_of(r: &oxy_core::repo::Repo, bookmark: &str) -> Result<Vec<ChangeHash>, crate::repos::RepoError> {

    Ok(r.bookmarks()?.into_iter().find(|b| b.name == bookmark).map(|b| b.heads).unwrap_or_default())

}

/// One consumer of an upstream, as an upstream's page and API show it.

#[derive(Clone, Debug, Serialize)]

pub struct Consumer {

    pub owner: String,

    pub name: String,

    pub bookmark: String,

    pub path: String,

    /// The upstream heads it pins, hex.

    pub pinned: Vec<String>,

    /// Upstream changes on its default bookmark the pin lacks; `None` when

    /// the hub cannot tell (the pinned changes are not in the upstream).

    pub behind: Option<usize>,

    /// How it follows the upstream (from its setting, or the site's).

    pub follow: Follow,

    /// Its open bump proposal's number, if it has one.

    pub bump: Option<i64>,

    /// The newest run of its bump bookmark: number and status.

    pub run: Option<(i64, String)>,

    /// Its last landed bump: number and when.

    pub landed: Option<(i64, i64)>,

    /// What the views show (#150).

    pub colour: Colour,

}

/// A consumer's Diffusion status at a glance (#150, Andy's colours):

/// green when the upstream was checked and the consumer's run on the bump

/// passed; amber when the run passed but the upstream has no checks; red

/// when the run failed; grey while it runs; up to date when there is

/// nothing to bump; off when it does not follow.

#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]

#[serde(rename_all = "kebab-case")]

pub enum Colour {

    Green,

    Amber,

    Red,

    Grey,

    UpToDate,

    Off,

}

impl Colour {

    pub fn class(&self) -> &'static str {

        match self {

            Colour::Green => "df-green",

            Colour::Amber => "df-amber",

            Colour::Red => "df-red",

            Colour::Grey => "df-grey",

            Colour::UpToDate => "df-current",

            Colour::Off => "df-off",

        }

    }

    pub fn label(&self) -> &'static str {

        match self {

            Colour::Green => "green",

            Colour::Amber => "amber: upstream unverified",

            Colour::Red => "red: its run failed",

            Colour::Grey => "testing",

            Colour::UpToDate => "up to date",

            Colour::Off => "not following",

        }

    }

    /// The colour from what is known of a consumer.

    pub fn of(follow: Follow, behind: Option<usize>, bump: Option<i64>, run: Option<&str>, unverified: bool) -> Colour {

        if follow == Follow::Off {

            return Colour::Off;

        }

        match (bump, run) {

            (Some(_), Some("passed")) if unverified => Colour::Amber,

            (Some(_), Some("passed")) => Colour::Green,

            (Some(_), Some("failed" | "error" | "cancelled")) => Colour::Red,

            (Some(_), _) => Colour::Grey,

            (None, _) if behind == Some(0) => Colour::UpToDate,

            (None, _) => Colour::Grey,

        }

    }

}

/// The consumers of `row` that `account` may read, with how far behind

/// the upstream's default bookmark each pin is.

pub async fn consumers(app: &Shared, row: &RepoRow, account: Option<&Account>) -> Result<Vec<Consumer>, crate::repos::RepoError> {

    let Some(name) = RepoName::new(&row.owner, &row.name) else { return Ok(Vec::new()) };

    let id = app.repos.read(&name, |r, _| Ok(r.id()?)).await?;

    let found: Vec<(i64, String, String, String)> = sqlx::query_as(

        "SELECT consumer_id, bookmark, path, heads FROM diffusion_pins WHERE upstream = ? ORDER BY consumer_id, bookmark, path",

    )

    .bind(hex::encode(id))

    .fetch_all(&app.db)

    .await

    .map_err(|e| crate::repos::RepoError::Io(std::io::Error::other(e)))?;

    let mut out = Vec::new();

    let mut pinned_sets = Vec::new();

    let unverified = unverified(app, row).await;

    for (consumer_id, bookmark, path, heads) in found {

        let Ok(Some(c)) = db::repos::get_by_id(&app.db, consumer_id).await else { continue };

        if !roles::allows(roles::role_for(&app.db, account, &c).await.ok().flatten(), crate::auth::Access::Read) {

            continue;

        }

        // One row per consumer and subrepo: the bookmark its work lands on.

        if bookmark != target(&app.db, &c).await {

            continue;

        }

        let pinned: Vec<String> = heads.split('+').filter(|h| !h.is_empty()).map(String::from).collect();

        pinned_sets.push(pinned.iter().filter_map(|h| ChangeHash::from_hex(h)).collect::<Vec<_>>());

        let bump_bm = bump_bookmark(&path);

        let mine: Vec<db::proposals::ProposalRow> = db::proposals::list(&app.db, c.id, &[])

            .await

            .unwrap_or_default()

            .into_iter()

            .filter(|p| p.source_bookmark == bump_bm)

            .collect();

        let bump = mine.iter().find(|p| p.state == "open" || p.state == "draft").map(|p| p.number);

        let landed = mine.iter().filter(|p| p.state == "landed").max_by_key(|p| p.landed_ms).map(|p| (p.number, p.landed_ms));

        let run: Option<(i64, String)> =

            sqlx::query_as("SELECT number, status FROM flow_runs WHERE repo_id = ? AND bookmark = ? ORDER BY id DESC LIMIT 1")

                .bind(c.id)

                .bind(&bump_bm)

                .fetch_optional(&app.db)

                .await

                .ok()

                .flatten();

        let follow = rule(&app.config.diffusion.follow, &c, None, &path).follow;

        out.push(Consumer {

            owner: c.owner,

            name: c.name,

            bookmark,

            path,

            pinned,

            behind: None,

            follow,

            bump,

            run,

            landed,

            colour: Colour::Off,

        });

    }

    if out.is_empty() {

        return Ok(out);

    }

    let default = row.default_bookmark.clone();

    let behind = app

        .repos

        .read(&name, move |r, cache| {

            let tip = heads_of(r, &default)?;

            let tip = cache.summary(r, &tip)?;

            Ok(pinned_sets

                .iter()

                .map(|p| cache.summary(r, p).ok().map(|s| tip.changes.iter().filter(|c| !s.changes.contains(*c)).count()))

                .collect::<Vec<_>>())

        })

        .await?;

    for (c, b) in out.iter_mut().zip(behind) {

        c.behind = b;

        c.colour = Colour::of(c.follow, b, c.bump, c.run.as_ref().map(|r| r.1.as_str()), unverified);

    }

    Ok(out)

}

/// What `consumer` follows, as its Follows panel shows: each upstream on

/// this hub it pins, with its status there.

pub async fn follows(app: &Shared, consumer: &RepoRow, account: Option<&Account>) -> Vec<(RepoRow, Consumer)> {

    let ups: Vec<i64> =

        sqlx::query_scalar("SELECT DISTINCT upstream_repo_id FROM diffusion_pins WHERE consumer_id = ? AND upstream_repo_id IS NOT NULL")

            .bind(consumer.id)

            .fetch_all(&app.db)

            .await

            .unwrap_or_default();

    let mut out = Vec::new();

    for id in ups {

        let Ok(Some(up)) = db::repos::get_by_id(&app.db, id).await else { continue };

        if !roles::allows(roles::role_for(&app.db, account, &up).await.ok().flatten(), crate::auth::Access::Read) {

            continue;

        }

        for c in consumers(app, &up, account).await.unwrap_or_default() {

            if c.owner == consumer.owner && c.name == consumer.name {

                out.push((up.clone(), c));

            }

        }

    }

    out

}

/// Every upstream on this hub with its consumers, for Admin › Diffusion.

pub async fn everything(app: &Shared, account: Option<&Account>) -> Vec<(RepoRow, Vec<Consumer>, bool)> {

    let ups: Vec<i64> = sqlx::query_scalar("SELECT DISTINCT upstream_repo_id FROM diffusion_pins WHERE upstream_repo_id IS NOT NULL")

        .fetch_all(&app.db)

        .await

        .unwrap_or_default();

    let mut out = Vec::new();

    for id in ups {

        let Ok(Some(up)) = db::repos::get_by_id(&app.db, id).await else { continue };

        let list = consumers(app, &up, account).await.unwrap_or_default();

        let unverified = unverified(app, &up).await;

        out.push((up, list, unverified));

    }

    out

}

/// The bookmark a consumer's bump for the subrepo at `path` lives on.

pub fn bump_bookmark(path: &str) -> String {

    format!("diffusion/{}", path.trim_matches('/'))

}

/// What a bump did.

#[derive(Clone, Debug, PartialEq, Eq)]

pub enum Bumped {

    /// The bump bookmark now holds the target plus one change moving the pin.

    Moved { bookmark: String, change: ChangeHash, heads: Vec<ChangeHash>, from: Vec<ChangeHash>, state: String },

    /// The pin already holds the upstream state.

    UpToDate,

    /// The bump bookmark already moves the pin to that state.

    Already { from: Vec<ChangeHash>, to: Vec<ChangeHash> },

    /// The pin names changes the upstream does not hold: not followed.

    Unknown,

    /// Taking the upstream's state with the pin's own extra changes would

    /// conflict.

    Conflicts,

    /// The consumer has no Oxygen pin of that upstream at `path` on its

    /// target bookmark.

    NoPin,

}

/// Where a consumer's pin `old` moves when the upstream reaches `new`,

/// worked out in the upstream (§4.3).

#[derive(Clone, Debug, PartialEq, Eq)]

pub enum Plan {

    /// To these heads, which add this many changes.

    To(Vec<ChangeHash>, usize),

    /// Nothing to add.

    UpToDate,

    /// The pin names changes the upstream does not hold (rewritten, or

    /// from elsewhere): not followed.

    Unknown,

    /// Taking `new` with the pin's own extra changes would conflict.

    Conflicts,

}

/// The pin `old` moved to the upstream's `new`. When the pin has changes

/// `new` lacks (say, from an upstream proposal not landed yet), they are

/// kept: changes that do not touch each other combine, so the move is to

/// both, never dropping what the consumer already relies on.

pub async fn plan(app: &Shared, upstream: &RepoName, old: Vec<ChangeHash>, new: Vec<ChangeHash>) -> Result<Plan, crate::repos::RepoError> {

    app.repos

        .read(upstream, move |r, cache| {

            let Ok(o) = cache.summary(r, &old) else { return Ok(Plan::Unknown) };

            let n = cache.summary(r, &new)?;

            if o.changes.iter().all(|c| n.changes.contains(c)) {

                let added = n.changes.len() - o.changes.len();

                return Ok(if added == 0 { Plan::UpToDate } else { Plan::To(new, added) });

            }

            let mut both = new.clone();

            both.extend(old.iter().copied());

            let heads = r.load_state(&both)?.heads();

            let b = cache.summary(r, &heads)?;

            if b.conflicted && !o.conflicted {

                return Ok(Plan::Conflicts);

            }

            let added = b.changes.len() - o.changes.len();

            Ok(if added == 0 { Plan::UpToDate } else { Plan::To(heads, added) })

        })

        .await

}

/// Move `consumer`'s pin of `upstream` at `path` to `new_heads`, on a bump

/// bookmark built from `target` (§4.3): one change, signed with the hub's

/// key, on the target's heads. The upstream decides forwardness.

pub async fn bump(

    app: &Shared,

    consumer: &RepoRow,

    target: &str,

    path: &str,

    upstream: &RepoRow,

    new_heads: Vec<ChangeHash>,

) -> Result<Bumped, crate::repos::RepoError> {

    use oxy_core::identity::Domain;

    use oxy_core::subrepo::Pin;

    let (Some(cname), Some(uname)) = (RepoName::new(&consumer.owner, &consumer.name), RepoName::new(&upstream.owner, &upstream.name))

    else {

        return Ok(Bumped::NoPin);

    };

    let upstream_id = app.repos.read(&uname, |r, _| Ok(r.id()?)).await?;

    // The consumer's current pin at `path` on its target, and on its bump

    // bookmark if it has one.

    let (t, p) = (target.to_string(), path.trim_matches('/').to_string());

    let (current, bumped) = app

        .repos

        .read(&cname, move |r, cache| {

            let pin_on = |bookmark: &str| -> Result<Option<Pin>, crate::repos::RepoError> {