//! 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}"))?;

    }

    sqlx::query("INSERT OR REPLACE INTO diffusion_indexed (consumer_id, bookmark, heads, updated_ms) VALUES (?, ?, ?, ?)")

        .bind(row.id)

        .bind(bookmark)

        .bind(heads_key(heads))

        .bind(now)

        .execute(&mut *tx)

        .await

        .map_err(|e| format!("noting the state: {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;

            }

        }

    }

}

/// Heads as one comparable string: sorted, joined with `+`.

fn heads_key(heads: &[ChangeHash]) -> String {

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

    h.sort();

    h.join("+")

}

/// How often [`reconciler`] looks.

pub const RECONCILE_EVERY: std::time::Duration = std::time::Duration::from_secs(10 * 60);

/// What one reconciling pass mended.

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

pub struct Repairs {

    /// Followed bookmarks whose index was behind the bookmark.

    pub reindexed: usize,

    /// Consumers behind an upstream's newest passing state with no open

    /// bump, bumped now.

    pub caught_up: usize,

}

/// Every [`RECONCILE_EVERY`], mend the pin index and the bumps it drives

/// (oxygenhub#210).

pub async fn reconciler(app: Shared) {

    loop {

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

        let r = reconcile(&app).await;

        if r != Repairs::default() {

            tracing::warn!("diffusion: reconciled: {} bookmarks re-indexed, {} consumers caught up", r.reindexed, r.caught_up);

        }

    }

}

/// One pass: re-index each followed bookmark whose indexed state is not

/// the state it is at now; then bump each consumer whose pin is not its

/// upstream's newest passing state and that has no open bump for it.

pub async fn reconcile(app: &Shared) -> Repairs {

    let mut repairs = Repairs::default();

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

    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 Ok(heads) = app.repos.read(&name, move |r, _| heads_of(r, &b)).await else { continue };

            let indexed: Option<String> = sqlx::query_scalar("SELECT heads FROM diffusion_indexed WHERE consumer_id = ? AND bookmark = ?")

                .bind(row.id)

                .bind(&bookmark)

                .fetch_optional(&app.db)

                .await

                .ok()

                .flatten();

            if indexed.as_deref() != Some(heads_key(&heads).as_str()) {

                tracing::info!("diffusion: {}/{} {bookmark}: the index was not at the bookmark; re-indexing", row.owner, row.name);

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

                repairs.reindexed += 1;

            }

        }

    }

    // Consumers behind with no open bump.

    let pins: Vec<(i64, String, String, String)> =

        sqlx::query_as("SELECT consumer_id, path, upstream, heads FROM diffusion_pins WHERE upstream_repo_id IS NOT NULL")

            .fetch_all(&app.db)

            .await

            .unwrap_or_default();

    let mut caught = std::collections::HashSet::new();

    for (consumer_id, path, upstream, pinned) in pins {

        if caught.contains(&consumer_id) {

            continue;

        }

        let Some(up) = repo_by_oxy_id(app, &upstream).await else { continue };

        let Some(green) = last_green(app, &up).await else { continue };

        if heads_key(&green) == heads_key(&pinned.split('+').filter_map(ChangeHash::from_hex).collect::<Vec<_>>()) {

            continue;

        }

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

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

        let learned = target(&app.db, &consumer).await;

        let modules = oxymodules(app, &cname, &learned).await;

        if rule(&app.config.diffusion.follow, &consumer, modules.as_deref(), &path).follow == Follow::Off {

            continue;

        }

        // The newest bump at this path: open or a draft is a bump already,

        // and one a person closed is a "no" the reconciler respects (the

        // upstream's next move proposes again as usual).

        let bump_bm = bump_bookmark(&path);

        let newest = || async {

            db::proposals::list(&app.db, consumer_id, &[])

                .await

                .unwrap_or_default()

                .into_iter()

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

                .max_by_key(|p| p.number)

                .map(|p| p.state)

        };

        let has_bump = || async { matches!(newest().await.as_deref(), Some("open" | "draft")) };

        if matches!(newest().await.as_deref(), Some("open" | "draft" | "closed")) {

            continue;

        }

        caught.insert(consumer_id);

        catch_up(app, &consumer).await;

        if has_bump().await {

            tracing::info!(

                "diffusion: {}/{} {path} was behind {}/{} with no bump; bumped",

                consumer.owner,

                consumer.name,

                up.owner,

                up.name

            );

            repairs.caught_up += 1;

        }

    }

    repairs

}

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

    /// The pill's two halves: the upstream, then this consumer.

    pub upstream: Half,

    pub downstream: Half,

}

impl Consumer {

    /// As the API gives it (oxygenhub#153), with its upstream's name.

    pub fn api(&self, upstream: &str) -> hub_api::DiffusionPin {

        let half = |h: &Half| hub_api::DiffusionHalf {

            tone: h.class.trim_start_matches("df-").to_string(),

            word: h.word.to_string(),

            title: h.title.clone(),

        };

        hub_api::DiffusionPin {

            upstream: upstream.to_string(),

            consumer: format!("{}/{}", self.owner, self.name),

            path: self.path.clone(),

            bookmark: self.bookmark.clone(),

            behind: self.behind.map(|n| n as u64),

            follow: self.follow.as_str().to_string(),

            bump: self.bump,

            run: self.run.as_ref().map(|r| r.0),

            run_status: self.run.as_ref().map(|r| r.1.clone()),

            upstream_state: half(&self.upstream),

            consumer_state: half(&self.downstream),

        }

    }

}

/// One half of a Diffusion pill: a colour class and a word (#150).

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

pub struct Half {

    pub class: &'static str,

    pub word: &'static str,

    pub title: String,

}

impl Half {

    /// The upstream's half: was the upstream itself tested?

    pub fn upstream(unverified: bool, run: Option<&str>, slug: &str) -> Half {

        let (class, word, title) = match (unverified, run) {

            (true, _) => ("df-amber", "untested", format!("{slug} has no checks of its own: nothing has tested it")),

            (false, Some("passed")) => ("df-green", "checked", format!("{slug}'s newest run passed")),

            (false, Some("failed" | "error" | "cancelled")) => ("df-red", "failing", format!("{slug}'s newest run failed")),

            (false, _) => ("df-grey", "testing", format!("{slug}'s newest run has not finished")),

        };

        Half { class, word, title }

    }

    /// The consumer's half: does it pass with the upstream?

    pub fn downstream(follow: Follow, behind: Option<usize>, bump: Option<i64>, run: Option<&str>, name: &str) -> Half {

        let (class, word, title) = if follow == Follow::Off {

            ("df-off", "not following", format!("{name} does not follow it"))

        } else {

            match (bump, run) {

                (Some(_), Some("passed")) => ("df-green", "passes", format!("{name}'s run on the bump passed")),

                (Some(_), Some("failed" | "error" | "cancelled")) => ("df-red", "fails", format!("{name}'s run on the bump failed")),

                (Some(_), Some(_)) => ("df-grey", "testing", format!("{name}'s run on the bump has not finished")),

                (Some(_), None) => ("df-grey", "waiting", format!("{name}'s bump has no run yet")),

                (None, _) if behind == Some(0) => ("df-green", "up to date", format!("{name} pins the newest state")),

                (None, _) => ("df-grey", "waiting", format!("{name} is behind; its bump has not been made yet")),

            }

        };

        Half { class, word, title }

    }

}

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