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

}