//! 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]) {

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

    let pins = if heads.is_empty() { Vec::new() } else { crate::flow::pins_in(app, name, heads.to_vec()).await };

    let now = db::now_ms();

    let Ok(mut tx) = app.db.begin().await else { return };

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

        .bind(row.id)

        .bind(bookmark)

        .execute(&mut *tx)

        .await

        .is_ok();

    for (path, id, pinned) in &pins {

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

        ok &=

            sqlx::query("INSERT INTO diffusion_pins (consumer_id, bookmark, path, upstream, heads, updated_ms) VALUES (?, ?, ?, ?, ?, ?)")

                .bind(row.id)

                .bind(bookmark)

                .bind(path)

                .bind(hex::encode(id))

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

                .bind(now)

                .execute(&mut *tx)

                .await

                .is_ok();

    }

    if ok {

        tx.commit().await.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;

        }

    }

}

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

}

/// 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();

    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;

        }

        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<_>>());

        out.push(Consumer { owner: c.owner, name: c.name, bookmark, path, pinned, behind: None });

    }

    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;

    }

    Ok(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,

    /// The upstream state does not hold everything the pin does (upstream

    /// moved backwards or was rewritten): consumers only move forward.

    Backwards,

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

    /// target bookmark.

    NoPin,

}

/// How the upstream's state `new` relates to a consumer's pin `old`, read

/// in the upstream: `Some(n)` new changes when `new` holds all of `old`

/// and more, `Some(0)` when it adds nothing, `None` when it lacks some.

pub async fn ahead_of(

    app: &Shared,

    upstream: &RepoName,

    old: Vec<ChangeHash>,

    new: Vec<ChangeHash>,

) -> Result<Option<usize>, crate::repos::RepoError> {

    app.repos

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

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

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

                return Ok(None);

            }

            Ok(Some(n.changes.len() - o.changes.len()))

        })

        .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> {

                let heads = heads_of(r, bookmark)?;

                if heads.is_empty() {

                    return Ok(None);

                }

                let loaded = cache.load(r, &heads)?;

                let Some(f) = loaded.snapshot.files.get(&p) else { return Ok(None) };

                let oxy_core::state::FileContent::Opaque { blobs } = &f.content else { return Ok(None) };

                let Some(b) = blobs.first() else { return Ok(None) };

                Ok(Pin::decode(&r.read_blob(b)?).ok())

            };

            Ok((pin_on(&t)?, pin_on(&bump_bookmark(&p))?))

        })

        .await?;

    let Some(Pin::Oxy { repo, heads: old, bookmark }) = current else { return Ok(Bumped::NoPin) };

    if let Some(Pin::Oxy { heads: b, .. }) = &bumped {

        let (mut b, mut n) = (b.clone(), new_heads.clone());

        b.sort();

        n.sort();

        if b == n {

            return Ok(Bumped::Already);

        }

    }

    if repo != upstream_id {

        return Ok(Bumped::NoPin);

    }

    match ahead_of(app, &uname, old.clone(), new_heads.clone()).await? {

        None => return Ok(Bumped::Backwards),

        Some(0) => return Ok(Bumped::UpToDate),

        Some(_) => {}

    }

    let (key, secret) =

        crate::keys::load_secret(&app.config.keys_dir()).map_err(|e| crate::repos::RepoError::Io(std::io::Error::other(e.to_string())))?;

    let identity = app.hub_key.identity.clone();

    let upstream_label = format!("{}/{}", upstream.owner, upstream.name);

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

    let pin = Pin::Oxy { repo, heads: new_heads.clone(), bookmark };

    let old = old.clone();

    let short: Vec<String> = new_heads.iter().map(|h| h.to_hex()[..12].to_string()).collect();

    let message = format!("Diffusion: move {p} to {upstream_label} {}", short.join("+"));

    app.repos

        .write(&cname, move |r, _| {

            let base = heads_of(r, &t)?;

            let change = r.build_pin_change(&base, &p, &pin, &message, key)?;

            let hash = change.hash();

            let sig = secret.sign(Domain::Change, &key, &hash);

            let mut state = r.load_state(&base)?;

            state.add(change.clone()).map_err(|e| crate::repos::RepoError::Oxy(oxy_core::repo::RepoError::Msg(e.to_string())))?;

            let mut candidates = base.clone();

            candidates.push(hash);

            let heads = state.maximal(&candidates);

            let bookmark = bump_bookmark(&p);

            // The bump bookmark is the hub's own and is rebuilt on the

            // target each time, so the bump it replaces may go.

            let mut update = r.host_ref_update(&bookmark, &heads, Some((&key, &secret)))?;

            update.force = true;

            let state = update.new_state.to_hex();

            let report = r.apply_hosted(Some((change, vec![sig])), &[identity], update)?;

            if let Some(refused) = report.results.iter().find(|x| !x.ok) {

                return Err(crate::repos::RepoError::Oxy(oxy_core::repo::RepoError::Msg(format!(

                    "the bump was refused: {}",

                    refused.reason

                ))));

            }

            Ok(Bumped::Moved { bookmark, change: hash, heads, from: old, state })

        })

        .await

}

/// The account Diffusion's proposals are opened as.

pub const ACCOUNT: &str = "diffusion";

/// The `diffusion` account, made on first use with no way to sign in.

pub async fn account(pool: &Pool) -> Option<Account> {

    if let Ok(Some(a)) = db::accounts::account_by_name(pool, ACCOUNT).await {

        return Some(a);

    }

    db::accounts::create_account(pool, ACCOUNT, "diffusion@localhost", "!", "", false).await.ok()

}

/// The upstream changes a bump brings: in `new` and not in `old`, newest

/// first, as Markdown lines linking to the upstream's change pages.

pub async fn changelog(app: &Shared, upstream: &RepoRow, old: Vec<ChangeHash>, new: Vec<ChangeHash>) -> Vec<String> {

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

    let linked: std::collections::HashMap<String, String> =

        db::accounts::identity_accounts(&app.db).await.unwrap_or_default().into_iter().collect();

    let base = format!("/{}/{}", upstream.owner, upstream.name);

    app.repos

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

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

            let mut added: Vec<(u64, ChangeHash, String, String)> = Vec::new();

            for h in n.changes.iter().filter(|c| !o.changes.contains(*c)) {

                let Ok(c) = r.store().get_change(h) else { continue };

                let key = c.author.to_hex();

                let who = linked.get(&key).cloned().unwrap_or_else(|| key[..12].to_string());

                let first = c.message.lines().next().unwrap_or("").trim().to_string();

                added.push((c.hlc.wall_ms, *h, first, who));

            }

            added.sort_by_key(|a| std::cmp::Reverse(a.0));

            Ok(added

                .into_iter()

                .map(|(_, h, msg, who)| {

                    format!("- [`{}`]({base}/change/{}) {} ({who})", &h.to_hex()[..12], h.to_hex(), msg.replace('\n', " "))

                })

                .collect())

        })

        .await

        .unwrap_or_default()

}

/// Open the bump proposal for `bookmark`, or bring the open one up to

/// date: from `bookmark` into `target`, by the `diffusion` account,

/// labelled `dependencies`, its body the changelog. Its number.

pub async fn propose(

    app: &Shared,

    consumer: &RepoRow,

    target: &str,

    path: &str,

    upstream: &RepoRow,

    from: Vec<ChangeHash>,

    to: Vec<ChangeHash>,

) -> Option<i64> {

    use crate::db::proposals as pdb;

    let bookmark = bump_bookmark(path);

    let bookmark = bookmark.as_str();

    let me = account(&app.db).await?;

    let lines = changelog(app, upstream, from, to.clone()).await;

    let slug = format!("{}/{}", upstream.owner, upstream.name);

    let short: Vec<String> = to.iter().map(|h| h.to_hex()[..12].to_string()).collect();

    let title = format!(

        "Diffusion: {path} to {slug} {} ({})",

        short.join("+"),

        if lines.len() == 1 { "1 change".into() } else { format!("{} changes", lines.len()) }

    );

    let body = format!(

        "Moves the subrepo `{path}` to the newest state of [{slug}](/{slug}) that {}. Only the pin changes; this repository's own checks test it.\n\n**What it brings** ({} change{}):\n\n{}\n\nOpened and kept up to date by Diffusion (docs/diffusion.md). It never lands by itself.",

        if Follow::parse(&consumer.diffusion) == Some(Follow::Any) { "moved" } else { "passed its checks" },

        lines.len(),

        if lines.len() == 1 { "" } else { "s" },

        lines.join("\n")

    );

    let open = pdb::list(&app.db, consumer.id, &["open", "draft"]).await.ok()?.into_iter().find(|p| p.source_bookmark == bookmark);

    let number = match open {

        Some(p) => {

            pdb::edit(&app.db, p.id, &title, &body, target).await.ok()?;

            pdb::add_event(&app.db, p.id, ACCOUNT, "updated", &short.join("+")).await.ok();

            p.number

        }

        None => {

            let p = pdb::create(

                &app.db,

                pdb::NewProposal {

                    repo_id: consumer.id,

                    title: &title,

                    body: &body,

                    author_id: Some(me.id),

                    author: ACCOUNT,

                    draft: false,

                    target,

                    source_kind: "bookmark",

                    source_repo_id: None,

                    source_bookmark: bookmark,

                    source_changes: "",

                },

            )

            .await

            .ok()?;

            pdb::add_event(&app.db, p.id, ACCOUNT, "opened", "").await.ok();

            db::notify::index(&app.db, "proposal", consumer.id, p.id, p.number, &p.title, &p.body).await.ok();

            if let Ok(label) =

                db::issues::upsert_label(&app.db, consumer.id, "dependencies", "#0072ff", "A subrepo moves (Diffusion)").await

            {

                db::issues::set_proposal_label(&app.db, p.id, label.id, true).await.ok();

            }

            p.number

        }

    };

    Some(number)

}

/// Start a bump's pipeline, as a push of it would. Boxed as `Send`, which

/// breaks the cycle between a bump starting runs and a passing run

/// starting bumps that the compiler cannot see through.

fn start_runs(

    app: Shared,

    row: RepoRow,

    bookmark: String,

    state: String,

    heads: Vec<ChangeHash>,

) -> std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send>> {

    Box::pin(async move { crate::flow::on_push(&app, &row, &bookmark, &state, &heads, ACCOUNT).await })

}

/// The text of `.oxymodules` on `bookmark` of `name`, if any.

async fn oxymodules(app: &Shared, name: &RepoName, bookmark: &str) -> Option<String> {

    let b = bookmark.to_string();

    app.repos

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

            let heads = heads_of(r, &b)?;

            if heads.is_empty() {

                return Ok(None);

            }

            let loaded = cache.load(r, &heads)?;

            Ok(match loaded.snapshot.files.get(".oxymodules").map(|f| &f.content) {

                Some(oxy_core::state::FileContent::Text { bytes, .. }) => Some(String::from_utf8_lossy(bytes).into_owned()),

                _ => None,

            })

        })

        .await

        .ok()

        .flatten()

}

/// The upstream `upstream` reached `heads` on its default bookmark: by a

/// push (`green` false) or a passing run of that state (`green` true).

/// Bump and propose in every consumer that follows it that way (§4.3).

pub async fn propagate(app: &Shared, upstream: &RepoRow, heads: Vec<ChangeHash>, green: bool) {

    propagate_to(app, upstream, heads, green, None).await

}

/// `propagate`, to one consumer only when `only` names it.

async fn propagate_to(app: &Shared, upstream: &RepoRow, heads: Vec<ChangeHash>, green: bool, only: Option<i64>) {