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

    /// 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 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 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 })

        })

        .await

}

/// When a consumer takes an upstream's moves (§4.2).

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

#[serde(rename_all = "lowercase")]

pub enum Follow {

    Off,

    /// When the upstream state's checks pass.

    Green,

    /// On every move.

    Any,

}

impl Follow {

    pub fn parse(s: &str) -> Option<Follow> {

        match s.trim() {

            "off" => Some(Follow::Off),

            "green" => Some(Follow::Green),

            "any" => Some(Follow::Any),

            _ => None,

        }

    }

}

/// How one consumer follows one subrepo, from its `.oxymodules` entry, its

/// repository setting and the site's default, in that order.

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

pub struct Rule {

    pub follow: Follow,

    /// Land a green bump itself (D5); D1 only proposes.

    pub land_auto: bool,

    /// The bookmark to bump, when `.oxymodules` names one (else learned).

    pub target: Option<String>,

    /// Paths the consumer watches: upstream changes touching them are

    /// flagged in a bump's changelog.

    pub watch: Vec<String>,

    /// Where `follow` came from: `.oxymodules`, `setting` or `site`.

    pub from: &'static str,

}

/// The Diffusion keys of `path`'s entry in `.oxymodules` text: `follow`,

/// `land`, `target` and `watch` (comma-separated, or a TOML list).

pub fn module_keys(text: &str, path: &str) -> (Option<Follow>, Option<bool>, Option<String>, Vec<String>) {

    let (mut follow, mut land, mut target, mut watch) = (None, None, None, Vec::new());

    let mut here = false;

    for line in text.lines() {

        let line = line.trim();

        if line.is_empty() || line.starts_with('#') {

            continue;

        }

        if let Some(inner) = line.strip_prefix('[').and_then(|l| l.strip_suffix(']')) {

            here = inner.trim().trim_matches('"').trim_end_matches('/') == path.trim_end_matches('/');

            continue;

        }

        let (true, Some((k, v))) = (here, line.split_once('=')) else { continue };

        let v = v.trim();

        let unquoted = v.trim_matches('"').trim();

        match k.trim() {

            "follow" => follow = Follow::parse(unquoted),

            "land" => land = Some(unquoted == "auto"),

            "target" if !unquoted.is_empty() => target = Some(unquoted.to_string()),

            "watch" => {

                watch = v

                    .trim_start_matches('[')

                    .trim_end_matches(']')

                    .split(',')

                    .map(|w| w.trim().trim_matches('"').trim().to_string())

                    .filter(|w| !w.is_empty())

                    .collect()

            }

            _ => {}

        }

    }

    (follow, land, target, watch)

}

/// How `row` follows its subrepo at `path`, given its `.oxymodules` text.

pub fn rule(site_default: &str, row: &RepoRow, oxymodules: Option<&str>, path: &str) -> Rule {

    let (follow, land, target, watch) = oxymodules.map(|t| module_keys(t, path)).unwrap_or_default();

    let (follow, from) = match follow {

        Some(f) => (f, ".oxymodules"),

        None => match Follow::parse(&row.diffusion) {

            Some(f) => (f, "setting"),

            None => (Follow::parse(site_default).unwrap_or(Follow::Off), "site"),

        },

    };

    Rule { follow, land_auto: land.unwrap_or(false), target, watch, from }

}

#[cfg(test)]

mod tests {

    use super::*;

    const MODULES: &str = r#"

[rstar-api]

url = https://oxygenhub.io/doomsdayone/rstar-api

follow = green

land = "propose"

target = develop

watch = ["src/x16/irq*", "src/x16/frame*"]

[other]

url = https://example/other

follow = any

"#;

    fn row(diffusion: &str) -> RepoRow {

        RepoRow {

            id: 1,

            owner: "o".into(),

            name: "n".into(),

            description: String::new(),

            default_bookmark: "main".into(),

            visibility: "public".into(),

            created_ms: 0,

            updated_ms: 0,

            git_mirror: false,

            git_mirror_state: String::new(),

            forked_from: None,

            topics: String::new(),

            flow: String::new(),

            diffusion: diffusion.into(),

            ..Default::default()

        }

    }

    #[test]

    fn oxymodules_keys_are_read_per_entry() {

        let (f, land, target, watch) = module_keys(MODULES, "rstar-api");

        assert_eq!((f, land, target.as_deref()), (Some(Follow::Green), Some(false), Some("develop")));

        assert_eq!(watch, vec!["src/x16/irq*", "src/x16/frame*"]);

        assert_eq!(module_keys(MODULES, "other").0, Some(Follow::Any));

        assert_eq!(module_keys(MODULES, "missing"), (None, None, None, vec![]));

        assert_eq!(module_keys("[a]\nwatch = x, y\n", "a").3, vec!["x", "y"]);

    }

    #[test]

    fn oxymodules_wins_then_the_setting_then_the_site() {

        let r = rule("off", &row("any"), Some(MODULES), "rstar-api");

        assert_eq!((r.follow, r.from), (Follow::Green, ".oxymodules"));

        let r = rule("off", &row("any"), Some("[rstar-api]\nurl = x\n"), "rstar-api");

        assert_eq!((r.follow, r.from), (Follow::Any, "setting"));

        let r = rule("green", &row(""), None, "rstar-api");

        assert_eq!((r.follow, r.from), (Follow::Green, "site"));

        assert_eq!(rule("nonsense", &row(""), None, "x").follow, Follow::Off);

    }

}