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