wsb_github/

sync.rs

//! Keeping the mirror current.
//!
//! One pass asks GitHub about whatever is due, conditionally — an ETag makes
//! an unchanged repository free, and `since` keeps a changed one small —
//! writes what it learns into the mirror, and says what moved. Whoever runs
//! passes — a daemon on a timer, `wsb github sync` from a shell — runs the
//! same pass, so there is one schedule and one way of writing what came back.
//!
//! # What a pass will not do
//!
//! * It never empties anything because a fetch failed. A failure is recorded
//!   beside the data, not instead of it.
//! * It never forgets anything while some project could not be read: a
//!   worktree that failed to answer is not a worktree that has nothing.

use std::collections::{BTreeMap, BTreeSet};
use std::time::Duration;

use anyhow::{anyhow, Result};
use serde::Serialize;

use crate::boards::BoardRef;
use crate::client::{Conditions, Fetched, GitHubClient, GitHubProject};
use crate::mirror::{gist_is_whole, newest_update, Document, Mirror, Scope, SyncState, SINGLETON};
use crate::targets::Targets;

/// How often a CONDITIONAL scope is re-asked about.
///
/// Two minutes is affordable because these carry an ETag: GitHub does not
/// count a 304 against the rate limit at all, so a repository nobody touched
/// costs nothing to ask about.
pub const SWEEP: Duration = Duration::from_secs(120);

/// How often a scope that CANNOT be conditional is re-asked about.
///
/// Checks and project boards have no ETag to present — the checks endpoints
/// answer 200 every time, and GraphQL carries none — so every one of these
/// costs a request against the limit.
pub const SWEEP_UNCONDITIONAL: Duration = Duration::from_secs(300);

/// How often traffic is re-asked about. GitHub aggregates views and clones by
/// day, so asking more than a few times a day cannot produce a different
/// answer.
pub const SWEEP_TRAFFIC: Duration = Duration::from_secs(6 * 60 * 60);

/// How long a scope that FAILED waits before it is tried again. Longer than
/// the sweep, because the common failures — no token, a rate limit, no
/// network — do not resolve in two minutes.
pub const AFTER_FAILURE: Duration = Duration::from_secs(600);

/// What a pass did.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
pub struct SyncReport {
    /// Scopes whose data actually changed.
    pub changed: Vec<String>,
    /// Scopes that were asked and answered "nothing new".
    pub unchanged: Vec<String>,
    /// Scopes that could not be asked, and why. Serialised as
    /// `{ "scope": …, "error": … }`, which is what a reader names them by.
    #[serde(serialize_with = "failures_as_objects")]
    pub failed: Vec<(String, String)>,
    /// True when there was no token to sync with. Not a failure of any scope:
    /// nothing was attempted, and nothing is wrong with what is held.
    pub no_token: bool,
    /// What could not be found out about what to mirror.
    pub problems: Vec<String>,
}

impl SyncReport {
    pub fn touched_anything(&self) -> bool {
        !self.changed.is_empty()
    }
}

fn failures_as_objects<S: serde::Serializer>(
    failed: &[(String, String)],
    serializer: S,
) -> Result<S::Ok, S::Error> {
    use serde::ser::SerializeSeq;
    #[derive(Serialize)]
    struct Failure<'a> {
        scope: &'a str,
        error: &'a str,
    }
    let mut seq = serializer.serialize_seq(Some(failed.len()))?;
    for (scope, error) in failed {
        seq.serialize_element(&Failure { scope, error })?;
    }
    seq.end()
}

/// Every scope the targets want, except the boards' cards.
///
/// The cards of a board are known only once its owner's board list is held —
/// a board is referenced by number and its cards by its node id — so those
/// are decided after that list is current; see [`board_scopes`].
pub fn wanted(mirror: &Mirror, targets: &Targets) -> Result<Vec<Scope>> {
    let mut scopes = vec![Scope::Gists, Scope::Notifications, Scope::RateLimit];
    for (repo, commits) in &targets.repositories {
        scopes.extend([
            Scope::Issues(repo.clone()),
            Scope::Pulls(repo.clone()),
            Scope::Repository(repo.clone()),
            Scope::Releases(repo.clone()),
            Scope::Tags(repo.clone()),
            Scope::Milestones(repo.clone()),
            Scope::Labels(repo.clone()),
            Scope::Traffic(repo.clone()),
        ]);
        scopes.extend(commits.iter().map(|commit| Scope::Checks {
            repo: repo.clone(),
            commit: commit.clone(),
        }));
    }
    scopes.extend(targets.board_owners().into_iter().map(Scope::Projects));
    // Comment threads are mirrored only where one is held: a thread nobody
    // opened is not worth a request per sweep, and asking for every issue's
    // comments would be one request per issue.
    scopes.extend(
        mirror
            .states()?
            .keys()
            .filter_map(|key| Scope::parse(key))
            .filter(|scope| match scope {
                Scope::Comments { repo, .. } => targets.repositories.contains_key(repo),
                _ => false,
            }),
    );
    Ok(scopes)
}

/// The boards the targets reference, as the mirror knows them.
///
/// A board whose owner's list has not been read yet, or that the list does
/// not contain, is reported rather than skipped silently: a reference to a
/// board nobody can read is something a person must hear about.
pub fn resolve_boards(
    mirror: &Mirror,
    boards: &BTreeSet<BoardRef>,
) -> Result<(Vec<GitHubProject>, Vec<String>)> {
    let mut found = Vec::new();
    let mut problems = Vec::new();
    // Per owner: its board list, or why there is none to look in.
    let mut lists: BTreeMap<String, std::result::Result<Vec<GitHubProject>, String>> =
        BTreeMap::new();
    for board in boards {
        if !lists.contains_key(&board.owner) {
            let list = owner_boards(mirror, &board.owner)?;
            lists.insert(board.owner.clone(), list);
        }
        match &lists[&board.owner] {
            Err(reason) => problems.push(format!("{board}: {reason}")),
            Ok(list) => match list.iter().find(|project| project.number == board.number) {
                Some(project) => found.push(project.clone()),
                None => problems.push(format!("{} has no board {}", board.owner, board.number)),
            },
        }
    }
    Ok((found, problems))
}

/// One owner's boards as the mirror holds them, or why it holds none.
pub fn owner_boards(
    mirror: &Mirror,
    owner: &str,
) -> Result<std::result::Result<Vec<GitHubProject>, String>> {
    let scope = Scope::Projects(owner.to_string());
    let state = mirror.state(&scope)?;
    if !state.ever_synced() {
        return Ok(Err(match &state.last_error {
            Some(error) => format!("{owner}'s boards could not be read: {error}"),
            None => format!("{owner}'s boards have not been read yet"),
        }));
    }
    let boards = mirror
        .payloads(&scope)?
        .into_iter()
        .map(|payload| {
            serde_json::from_value::<GitHubProject>(payload).map_err(|e| {
                anyhow!("a mirrored board of {owner} cannot be read by this version ({e}); sync again")
            })
        })
        .collect::<Result<Vec<_>>>()?;
    Ok(Ok(boards))
}

/// The card scopes of every referenced board the mirror can resolve.
pub fn board_scopes(mirror: &Mirror, targets: &Targets) -> Result<(Vec<Scope>, Vec<String>)> {
    let (found, problems) = resolve_boards(mirror, &targets.boards)?;
    Ok((
        found
            .into_iter()
            .map(|board| Scope::ProjectItems(board.id))
            .collect(),
        problems,
    ))
}

/// Sync everything that is due, and say what moved.
///
/// `force` ignores the schedule — what a refresh button means — but not the
/// ETags: a forced sync of an unchanged repository is still one 304.
///
/// `client` is `None` when there is no token; the pass then forgets what no
/// project has any more and asks GitHub nothing.
pub async fn sync(
    mirror: &Mirror,
    client: Option<&GitHubClient>,
    targets: &Targets,
    force: bool,
) -> SyncReport {
    let mut report = SyncReport {
        problems: targets.problems.clone(),
        ..SyncReport::default()
    };

    let first = match wanted(mirror, targets) {
        Ok(scopes) => scopes,
        Err(error) => {
            report.problems.push(format!("the mirror cannot be read: {error:#}"));
            return report;
        }
    };
    // The board lists first, then the cards of the boards those lists name.
    if let Some(client) = client {
        run(mirror, client, &first, force, &mut report).await;
    } else {
        report.no_token = true;
    }
    let cards = match board_scopes(mirror, targets) {
        Ok((scopes, problems)) => {
            report.problems.extend(problems);
            scopes
        }
        Err(error) => {
            report.problems.push(format!("the boards cannot be resolved: {error:#}"));
            Vec::new()
        }
    };
    if let Some(client) = client {
        run(mirror, client, &cards, force, &mut report).await;
    }

    let mut all = first;
    all.extend(cards);
    forget_departed(mirror, targets, &all, &mut report);
    report
}

/// Sync exactly these scopes, due or not as `force` says.
///
/// What a reader that is about to show one board does: it brings that board
/// current, and leaves the rest of the machine's mirror to its schedule.
pub async fn sync_scopes(
    mirror: &Mirror,
    client: &GitHubClient,
    scopes: &[Scope],
    force: bool,
) -> SyncReport {
    let mut report = SyncReport::default();
    run(mirror, client, scopes, force, &mut report).await;
    report
}

async fn run(
    mirror: &Mirror,
    client: &GitHubClient,
    scopes: &[Scope],
    force: bool,
    report: &mut SyncReport,
) {
    let now = chrono::Utc::now();
    for scope in scopes {
        let state = match mirror.state(scope) {
            Ok(state) => state,
            Err(error) => {
                report.failed.push((scope.key(), format!("{error:#}")));
                continue;
            }
        };
        if !force && !due(scope, &state, now) {
            continue;
        }
        match sync_one(client, mirror, scope, &state).await {
            Ok(true) => report.changed.push(scope.key()),
            Ok(false) => report.unchanged.push(scope.key()),
            Err(error) => {
                let message = format!("{error:#}");
                if let Err(write) = mirror.failed(scope, &message) {
                    report
                        .failed
                        .push((scope.key(), format!("{message}; and the failure could not be recorded: {write:#}")));
                    continue;
                }
                report.failed.push((scope.key(), message));
            }
        }
    }
}

/// Sync one scope. `true` when the mirror changed.
///
/// Every kind does the same three things: ask conditionally, write what came
/// back, record where the scope now stands. The differences between kinds are
/// which endpoint and which key — not how a failure is handled, or what an
/// unchanged answer means.
async fn sync_one(
    client: &GitHubClient,
    mirror: &Mirror,
    scope: &Scope,
    held: &SyncState,
) -> Result<bool> {
    let conditions = Conditions {
        etag: held.etag.clone(),
        since: held.cursor.clone(),
    };

    match scope {
        Scope::Issues(repo) => {
            let (owner, name) = split(repo)?;
            match client.list_issues_changed(owner, name, &conditions).await? {
                Fetched::Unchanged => unchanged(mirror, scope),
                Fetched::Fresh { value, etag } => {
                    let documents = documents_by(&value, |issue| issue.number.to_string())?;
                    incremental(mirror, scope, documents, etag)
                }
            }
        }
        Scope::Pulls(repo) => {
            let (owner, name) = split(repo)?;
            match client.list_pulls_changed(owner, name, &conditions).await? {
                Fetched::Unchanged => unchanged(mirror, scope),
                Fetched::Fresh { value, etag } => {
                    let documents = documents_by(&value, |pull| pull.number.to_string())?;
                    incremental(mirror, scope, documents, etag)
                }
            }
        }
        Scope::Repository(repo) => {
            let (owner, name) = split(repo)?;
            match client.repository_changed(owner, name, &conditions).await? {
                Fetched::Unchanged => unchanged(mirror, scope),
                Fetched::Fresh { value, etag } => {
                    let changed = mirror.put_only(scope, serde_json::to_value(&value)?)?;
                    mirror.synced(scope, etag.as_deref(), None)?;
                    Ok(changed)
                }
            }
        }
        Scope::Checks { repo, commit } => {
            let (owner, name) = split(repo)?;
            let checks = client.checks_for(owner, name, commit).await?;
            let document =
                Document::of(SINGLETON, serde_json::to_value(&checks)?).with_state(checks.state.clone());
            let changed = mirror.put(scope, &[document])?;
            mirror.synced(scope, None, None)?;
            Ok(changed)
        }
        Scope::Releases(repo) => {
            let (owner, name) = split(repo)?;
            match client.list_releases_changed(owner, name, &conditions).await? {
                Fetched::Unchanged => unchanged(mirror, scope),
                Fetched::Fresh { value, etag } => {
                    replace(mirror, scope, documents_by(&value, |release| release.id.to_string())?, etag)
                }
            }
        }
        Scope::Tags(repo) => {
            let (owner, name) = split(repo)?;
            match client.list_tags_changed(owner, name, &conditions).await? {
                Fetched::Unchanged => unchanged(mirror, scope),
                Fetched::Fresh { value, etag } => {
                    replace(mirror, scope, documents_by(&value, |tag| tag.name.clone())?, etag)
                }
            }
        }
        Scope::Milestones(repo) => {
            let (owner, name) = split(repo)?;
            match client.list_milestones_changed(owner, name, &conditions).await? {
                Fetched::Unchanged => unchanged(mirror, scope),
                Fetched::Fresh { value, etag } => replace(
                    mirror,
                    scope,
                    documents_by(&value, |milestone| milestone.number.to_string())?,
                    etag,
                ),
            }
        }
        Scope::Labels(repo) => {
            let (owner, name) = split(repo)?;
            match client.list_labels_changed(owner, name, &conditions).await? {
                Fetched::Unchanged => unchanged(mirror, scope),
                Fetched::Fresh { value, etag } => {
                    replace(mirror, scope, documents_by(&value, |label| label.name.clone())?, etag)
                }
            }
        }
        Scope::Traffic(repo) => {
            let (owner, name) = split(repo)?;
            // Needs push access. A repository somebody merely reads answers
            // 403, which is recorded as this scope's error.
            let traffic = client.traffic(owner, name).await?;
            let changed = mirror.put_only(scope, serde_json::to_value(&traffic)?)?;
            mirror.synced(scope, None, None)?;
            Ok(changed)
        }
        Scope::Comments { repo, number } => {
            let (owner, name) = split(repo)?;
            match client
                .list_comments_changed(owner, name, *number, &conditions)
                .await?
            {
                Fetched::Unchanged => unchanged(mirror, scope),
                Fetched::Fresh { value, etag } => {
                    replace(mirror, scope, documents_by(&value, |comment| comment.id.to_string())?, etag)
                }
            }
        }
        Scope::Gists => match client.list_gists_changed(&conditions).await? {
            Fetched::Unchanged => unchanged(mirror, scope),
            Fetched::Fresh { value, etag } => {
                let mut documents = Vec::new();
                for gist in &value {
                    let payload = serde_json::to_value(gist)?;
                    documents.push(if gist_is_whole(gist) {
                        Document::of(gist.id.clone(), payload)
                    } else {
                        // The list endpoint omits file bodies, so these rows
                        // are summaries and must say so — or an editor opens
                        // one and saves empty files over a gist.
                        Document::summary(gist.id.clone(), payload)
                    });
                }
                replace(mirror, scope, documents, etag)
            }
        },
        Scope::Notifications => match client.list_notifications_changed(&conditions).await? {
            Fetched::Unchanged => unchanged(mirror, scope),
            Fetched::Fresh { value, etag } => replace(
                mirror,
                scope,
                documents_by(&value, |notification| notification.id.clone())?,
                etag,
            ),
        },
        Scope::RateLimit => {
            let limit = client.rate_limit().await?;
            let changed = mirror.put_only(scope, serde_json::to_value(&limit)?)?;
            mirror.synced(scope, None, None)?;
            Ok(changed)
        }
        Scope::Projects(login) => {
            let projects = client.list_projects(login).await?;
            replace(mirror, scope, documents_by(&projects, |project| project.id.clone())?, None)
        }
        Scope::ProjectItems(id) => {
            let items = client.project_items(id).await?;
            replace(mirror, scope, documents_by(&items, |item| item.id.clone())?, None)
        }
    }
}

/// A scope that answered "nothing new".
fn unchanged(mirror: &Mirror, scope: &Scope) -> Result<bool> {
    mirror.synced(scope, None, None)?;
    Ok(false)
}

/// Write an incremental answer, moving the cursor to the newest thing in it.
fn incremental(
    mirror: &Mirror,
    scope: &Scope,
    documents: Vec<Document>,
    etag: Option<String>,
) -> Result<bool> {
    let cursor = newest_update(documents.iter());
    let changed = mirror.put(scope, &documents)?;
    mirror.synced(scope, etag.as_deref(), cursor.as_deref())?;
    Ok(changed)
}

/// Write an authoritative answer.
fn replace(
    mirror: &Mirror,
    scope: &Scope,
    documents: Vec<Document>,
    etag: Option<String>,
) -> Result<bool> {
    let changed = mirror.put(scope, &documents)?;
    mirror.synced(scope, etag.as_deref(), None)?;
    Ok(changed)
}

/// Turn what GitHub sent into documents, keyed the way that kind is keyed.
fn documents_by<T: Serialize>(values: &[T], key: impl Fn(&T) -> String) -> Result<Vec<Document>> {
    values
        .iter()
        .map(|value| Ok(Document::of(key(value), serde_json::to_value(value)?)))
        .collect()
}

fn split(repo: &str) -> Result<(&str, &str)> {
    repo.split_once('/')
        .ok_or_else(|| anyhow!("{repo} is not owner/repo"))
}

/// The schedule, as a decision about one state at one moment.
///
/// Separated from the database on purpose: this is the whole of the policy —
/// how often a healthy scope is re-asked, how long a failing one waits — and
/// it is decided by elapsed time, which a test can state and cannot wait for.
pub fn due(scope: &Scope, state: &SyncState, now: chrono::DateTime<chrono::Utc>) -> bool {
    let Some(attempted) = state.attempted_at.as_deref().and_then(parse_time) else {
        // Never asked about: a project opened for the first time must not
        // wait out a sweep before it shows anything.
        return true;
    };
    let waited = (now - attempted).to_std().unwrap_or(Duration::ZERO);
    if state.last_error.is_some() {
        return waited >= AFTER_FAILURE;
    }
    waited >= interval(scope)
}

/// How long a scope waits between askings, decided by whether the ask is
/// FREE: a conditional request that answers 304 costs nothing, and anything
/// that spends a request every time has to earn its frequency.
pub fn interval(scope: &Scope) -> Duration {
    match scope {
        Scope::Traffic(_) => SWEEP_TRAFFIC,
        Scope::Checks { .. } | Scope::Projects(_) | Scope::ProjectItems(_) => SWEEP_UNCONDITIONAL,
        _ => SWEEP,
    }
}

/// Drop what no project has any more.
///
/// Nothing is dropped while any project could not be read, or while a
/// referenced board could not be resolved: either way the list of what is
/// wanted is incomplete, and acting on an incomplete list deletes what is
/// still wanted.
fn forget_departed(mirror: &Mirror, targets: &Targets, wanted: &[Scope], report: &mut SyncReport) {
    if !report.problems.is_empty() {
        return;
    }
    let keep: BTreeSet<String> = wanted.iter().map(Scope::key).collect();
    let known = match mirror.states() {
        Ok(known) => known,
        Err(error) => {
            report.problems.push(format!("the mirror cannot be read: {error:#}"));
            return;
        }
    };
    for key in known.keys() {
        if keep.contains(key) {
            continue;
        }
        let Some(scope) = Scope::parse(key) else {
            report.problems.push(format!("the mirror holds a scope it cannot read: {key}"));
            continue;
        };
        let forget = match &scope {
            // The person's own, not any project's.
            Scope::Gists | Scope::Notifications | Scope::RateLimit => false,
            // Wanted ones are all in `keep`: a commit no worktree is on, an
            // owner whose boards nobody references, a board nobody references.
            Scope::Checks { .. } | Scope::Projects(_) | Scope::ProjectItems(_) => true,
            // A repository's own kinds, comment threads included, go when no
            // project has the repository any more.
            other => other
                .repository()
                .is_some_and(|repo| !targets.repositories.contains_key(repo)),
        };
        if forget {
            match mirror.forget(&scope) {
                Ok(()) => report.changed.push(scope.key()),
                Err(error) => report
                    .failed
                    .push((scope.key(), format!("cannot forget it: {error:#}"))),
            }
        }
    }
}

/// How often a long-running process runs a pass. Each pass asks only what is
/// due, so this is how quickly something due is noticed, not how often GitHub
/// is asked.
pub const TICK: Duration = Duration::from_secs(30);

/// Keep the machine's mirror current for as long as the caller runs, and hand
/// every pass's report to `on_pass`.
///
/// What a process that stays — a workbench, a daemon — runs. Every pass reads
/// what to mirror from the projects again and resolves the token again, so a
/// worktree added or a `gh auth login` made while it runs is picked up on the
/// next pass rather than at the next restart.
pub async fn keep_current(
    prj: wsb_prj::Prj,
    mirror: Mirror,
    mut on_pass: impl FnMut(&SyncReport) + Send,
) {
    loop {
        let targets_of = prj.clone();
        let read = tokio::task::spawn_blocking(move || {
            (Targets::of(&targets_of), crate::token::resolve())
        })
        .await;
        let report = match read {
            Ok((Ok(targets), token)) => {
                let client = token.ok().map(|token| GitHubClient::new(token.value()));
                sync(&mirror, client.as_ref(), &targets, false).await
            }
            Ok((Err(error), _)) => SyncReport {
                problems: vec![format!("the projects under {} cannot be read: {error:#}", prj.root().display())],
                ..SyncReport::default()
            },
            Err(error) => SyncReport {
                problems: vec![format!("reading what to mirror failed: {error}")],
                ..SyncReport::default()
            },
        };
        on_pass(&report);
        tokio::time::sleep(TICK).await;
    }
}

fn parse_time(value: &str) -> Option<chrono::DateTime<chrono::Utc>> {
    chrono::DateTime::parse_from_rfc3339(value)
        .ok()
        .map(|time| time.with_timezone(&chrono::Utc))
}

#[cfg(test)]
#[path = "sync_tests.rs"]
mod tests;