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