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

}

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

    }

}