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

                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 repo != upstream_id {

        return Ok(Bumped::NoPin);

    }

    let new_heads = match plan(app, &uname, old.clone(), new_heads.clone()).await? {

        Plan::To(heads, _) => heads,

        Plan::UpToDate => return Ok(Bumped::UpToDate),

        Plan::Unknown => return Ok(Bumped::Unknown),

        Plan::Conflicts => return Ok(Bumped::Conflicts),

    };

    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 { from: old, to: new_heads });

        }

    }

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

}

/// Whether `path` matches a `watch` pattern: `*` is any run of characters

/// within one path segment, `**` any run across segments, `?` one

/// character; a pattern without a wildcard also matches everything below

/// it, as a directory would.

pub fn watched(pattern: &str, path: &str) -> bool {

    fn m(p: &[u8], s: &[u8]) -> bool {

        match (p.first(), s.first()) {

            (None, None) => true,

            (Some(b'*'), _) if p.get(1) == Some(&b'*') => (0..=s.len()).any(|i| m(&p[2..], &s[i..])),

            (Some(b'*'), _) => (0..=s.len()).take_while(|&i| i == 0 || s[i - 1] != b'/').any(|i| m(&p[1..], &s[i..])),

            (Some(b'?'), Some(c)) if *c != b'/' => m(&p[1..], &s[1..]),

            (Some(a), Some(b)) if a == b => m(&p[1..], &s[1..]),

            _ => false,

        }

    }

    let p = pattern.trim().trim_matches('/');

    if p.is_empty() {

        return false;

    }

    m(p.as_bytes(), path.as_bytes()) || (!p.contains(['*', '?']) && path.starts_with(&format!("{p}/")))

}

/// What a bump brings: the upstream changes in `new` and not in `old`,

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

/// with those touching a `watch` pattern flagged (and listed again).

pub struct Changelog {

    pub lines: Vec<String>,

    pub watched: Vec<String>,

}

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

    let Some(name) = RepoName::new(&upstream.owner, &upstream.name) else { return Changelog { lines: Vec::new(), watched: 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<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();

                // The watched paths it touches, when the game watches any.

                let hits: Vec<String> = if watch.is_empty() {

                    Vec::new()

                } else {

                    crate::repos::view::change(r, cache, h)

                        .map(|d| d.files.into_iter().map(|f| f.path).filter(|p| watch.iter().any(|w| watched(w, p))).collect())

                        .unwrap_or_default()

                };

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

            }

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

            let line = |h: &ChangeHash, msg: &str, who: &str, flag: bool| {

                format!(

                    "- {}[`{}`]({base}/change/{}) {} ({who})",

                    if flag { "⚑ " } else { "" },

                    &h.to_hex()[..12],

                    h.to_hex(),

                    msg.replace('\n', " ")

                )

            };

            let watched = added

                .iter()

                .filter(|a| !a.4.is_empty())

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

                    format!(

                        "{} — touches {}",

                        line(h, msg, who, true),

                        hits.iter().map(|p| format!("`{p}`")).collect::<Vec<_>>().join(", ")

                    )

                })

                .collect();

            let lines = added.iter().map(|(_, h, msg, who, hits)| line(h, msg, who, !hits.is_empty())).collect();

            Ok(Changelog { lines, watched })

        })

        .await

        .unwrap_or(Changelog { lines: Vec::new(), watched: Vec::new() })

}

/// Why a bump was made, which its proposal says.

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

pub enum Why {