wsb_github/

mirror.rs

//! The machine's copy of what GitHub said.
//!
//! # Why a mirror rather than a cache
//!
//! Every list used to be fetched from the API when a page opened. That made
//! the page as fast as the network, as available as the credential, and as
//! honest as the last request — and when a request failed, the list rendered
//! EMPTY. A missing token, a rate limit or a closed laptop each produced "you
//! have no gists", which is a lie told confidently.
//!
//! So a copy is kept and every reader reads the copy. Reads never touch the
//! network, never need a token, and never fail because GitHub is having a
//! morning. What a read CAN say is how old its answer is and whether the last
//! refresh failed, which is the honest version of the same page.
//!
//! # One per machine
//!
//! GitHub's repositories, gists, notifications and project boards belong to
//! accounts, not to any checkout, and one board spans many repositories. So
//! there is one mirror for every project under `$PRJ`, in `$PRJ/.state`, and
//! every process that reads or writes GitHub through this crate shares it.
//!
//! # What is stored
//!
//! Whole payloads, as GitHub sent them. The columns beside them exist only to
//! be queried on — state, update time — and everything a reader renders comes
//! out of the payload.

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

use anyhow::{bail, Context, Result};
use chrono::Utc;
use r2d2::PooledConnection;
use r2d2_sqlite::SqliteConnectionManager;
use rusqlite::{params, OptionalExtension, Transaction};
use serde::{Deserialize, Serialize};

use crate::client::GitHubGist;

/// The schema version this build reads and writes.
const SCHEMA_VERSION: i64 = 1;

/// How many changes the change log keeps. A reader further behind than this
/// re-reads everything rather than missing what fell off the end.
const CHANGES_KEPT: i64 = 10_000;

/// The file the mirror lives in, under `$PRJ/.state`.
pub const FILE: &str = "github.db";

/// What one mirrored thing is called, and what a sync is about.
///
/// One vocabulary for every kind: the sync fetches, compares and records them
/// identically, and a vocabulary per kind would mean a copy of every query
/// that asks what is stale, what failed, or what moved.
///
/// The string form is what the database and every change notice carry, so it
/// is part of the contract rather than a debugging convenience.
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub enum Scope {
    /// One repository's issues, as `owner/repo`.
    Issues(String),
    /// One repository's pull requests.
    Pulls(String),
    /// What CI says about one commit of one repository.
    ///
    /// Per commit, because the same repository is checked out at different
    /// commits in different worktrees, and CI for the work in front of a
    /// person is CI for the commit that worktree is on.
    Checks { repo: String, commit: String },
    Releases(String),
    Tags(String),
    /// What the repository IS: private or public, archived, its topics, its
    /// stars.
    Repository(String),
    Milestones(String),
    Labels(String),
    /// Views and clones. Needs push access, so this is the one scope whose
    /// refusal is ordinary.
    Traffic(String),
    /// Comments on one issue or pull request: `owner/repo#number`.
    Comments { repo: String, number: i64 },
    /// The authenticated user's gists.
    Gists,
    /// What GitHub thinks needs this person, across every repository.
    Notifications,
    /// How much API budget is left.
    RateLimit,
    /// Who the token belongs to, and the organisations it is a member of —
    /// the owners whose boards are this person's without anybody naming them.
    Viewer,
    /// Which project boards one owner has.
    Projects(String),
    /// One board's cards, by the board's node id.
    ProjectItems(String),
}

impl Scope {
    pub fn key(&self) -> String {
        match self {
            Self::Issues(repo) => format!("issues:{repo}"),
            Self::Pulls(repo) => format!("pulls:{repo}"),
            Self::Checks { repo, commit } => format!("checks:{repo}@{commit}"),
            Self::Releases(repo) => format!("releases:{repo}"),
            Self::Tags(repo) => format!("tags:{repo}"),
            Self::Repository(repo) => format!("repository:{repo}"),
            Self::Milestones(repo) => format!("milestones:{repo}"),
            Self::Labels(repo) => format!("labels:{repo}"),
            Self::Traffic(repo) => format!("traffic:{repo}"),
            Self::Comments { repo, number } => format!("comments:{repo}#{number}"),
            Self::Gists => "gists".to_string(),
            Self::Notifications => "notifications".to_string(),
            Self::RateLimit => "rate_limit".to_string(),
            Self::Viewer => "viewer".to_string(),
            Self::Projects(login) => format!("projects:{login}"),
            Self::ProjectItems(id) => format!("project_items:{id}"),
        }
    }

    /// The repository this scope is about, when it is about one.
    pub fn repository(&self) -> Option<&str> {
        match self {
            Self::Issues(repo)
            | Self::Pulls(repo)
            | Self::Checks { repo, .. }
            | Self::Releases(repo)
            | Self::Tags(repo)
            | Self::Repository(repo)
            | Self::Milestones(repo)
            | Self::Labels(repo)
            | Self::Traffic(repo)
            | Self::Comments { repo, .. } => Some(repo),
            Self::Gists | Self::Notifications | Self::RateLimit | Self::Viewer => None,
            Self::Projects(_) | Self::ProjectItems(_) => None,
        }
    }

    /// Read one back from what the database and a change notice carry.
    ///
    /// Every scope round-trips, or the process that wrote a row cannot say
    /// what the row is about — see the round-trip test.
    pub fn parse(key: &str) -> Option<Self> {
        if let Some((kind, rest)) = key.split_once(':') {
            if rest.is_empty() {
                return None;
            }
            return Some(match kind {
                "issues" => Self::Issues(rest.to_string()),
                "pulls" => Self::Pulls(rest.to_string()),
                "checks" => {
                    let (repo, commit) = rest.rsplit_once('@')?;
                    if repo.is_empty() || commit.is_empty() {
                        return None;
                    }
                    Self::Checks {
                        repo: repo.to_string(),
                        commit: commit.to_string(),
                    }
                }
                "releases" => Self::Releases(rest.to_string()),
                "tags" => Self::Tags(rest.to_string()),
                "repository" => Self::Repository(rest.to_string()),
                "milestones" => Self::Milestones(rest.to_string()),
                "labels" => Self::Labels(rest.to_string()),
                "traffic" => Self::Traffic(rest.to_string()),
                "comments" => {
                    let (repo, number) = rest.rsplit_once('#')?;
                    Self::Comments {
                        repo: repo.to_string(),
                        number: number.parse().ok()?,
                    }
                }
                "projects" => Self::Projects(rest.to_string()),
                "project_items" => Self::ProjectItems(rest.to_string()),
                _ => return None,
            });
        }
        match key {
            "gists" => Some(Self::Gists),
            "notifications" => Some(Self::Notifications),
            "rate_limit" => Some(Self::RateLimit),
            "viewer" => Some(Self::Viewer),
            _ => None,
        }
    }

    /// What a person calls it, for a message about it.
    pub fn label(&self) -> String {
        match self {
            Self::Comments { repo, number } => format!("{repo}#{number} comments"),
            Self::Checks { repo, commit } => {
                format!("checks of {repo} at {}", &commit[..commit.len().min(12)])
            }
            Self::Gists => "gists".to_string(),
            Self::Notifications => "notifications".to_string(),
            Self::RateLimit => "the rate limit".to_string(),
            Self::Viewer => "the token's account".to_string(),
            Self::Projects(login) => format!("{login}'s projects"),
            Self::ProjectItems(id) => format!("project {id}"),
            other => other.repository().unwrap_or_default().to_string(),
        }
    }

    /// Whether a whole answer replaces what is held.
    ///
    /// A LIST endpoint says what exists: a gist absent from it was deleted, a
    /// release absent from it was removed. An INCREMENTAL one says only what
    /// changed, so what it does not mention is still true — and clearing on
    /// one of those empties the mirror every time nothing happened.
    ///
    /// Issues and pull requests are fetched with `since`, so they are
    /// incremental. Everything else is asked for whole.
    pub fn is_authoritative(&self) -> bool {
        !matches!(self, Self::Issues(_) | Self::Pulls(_))
    }
}

/// Where one scope stands: how current it is, and what went wrong last.
///
/// `synced_at` and `last_error` are both meaningful at once, and that pairing
/// is the point: a scope that synced an hour ago and failed a minute ago has
/// data worth showing AND a problem worth reporting.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct SyncState {
    /// GitHub's ETag for the last response, replayed as `If-None-Match`.
    pub etag: Option<String>,
    /// The newest `updated_at` seen, for incremental fetches.
    pub cursor: Option<String>,
    /// When the last SUCCESSFUL sync finished.
    pub synced_at: Option<String>,
    /// When the last attempt was made, successful or not.
    pub attempted_at: Option<String>,
    pub last_error: Option<String>,
}

impl SyncState {
    /// Whether this scope has ever been mirrored.
    pub fn ever_synced(&self) -> bool {
        self.synced_at.is_some()
    }

    /// What a reader is told about how current an answer is.
    pub fn standing(&self) -> Standing {
        Standing {
            synced_at: self.synced_at.clone(),
            last_error: self.last_error.clone(),
            ever_synced: self.ever_synced(),
        }
    }
}

/// How current one scope's answer is, as every reader is told it.
///
/// "Never read" and "read, and empty" are different, and so are "read an hour
/// ago" and "could not be read a minute ago": the three fields say which.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct Standing {
    pub synced_at: Option<String>,
    pub last_error: Option<String>,
    pub ever_synced: bool,
}

/// One document, as GitHub last described it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Document {
    /// What identifies it inside its scope: an issue number, a gist id, a tag
    /// name, a project item id.
    pub key: String,
    pub payload: serde_json::Value,
    pub state: Option<String>,
    pub updated_at: Option<String>,
    /// Whether the payload is the whole thing rather than a list summary.
    pub complete: bool,
    pub fetched_at: String,
}

impl Document {
    /// One document from a value GitHub sent, with the columns a query needs
    /// pulled out of it.
    ///
    /// The extraction is by field NAME, which is the same name in every kind
    /// GitHub returns — `state` and `updated_at`. A kind that carries neither
    /// stores neither, and is read back whole.
    pub fn of(key: impl Into<String>, payload: serde_json::Value) -> Self {
        let state = payload
            .get("state")
            .and_then(|state| state.as_str())
            .map(str::to_string);
        let updated_at = payload
            .get("updated_at")
            .or_else(|| payload.get("updatedAt"))
            .or_else(|| payload.get("published_at"))
            .and_then(|time| time.as_str())
            .map(str::to_string);
        Self {
            key: key.into(),
            payload,
            state,
            updated_at,
            complete: true,
            fetched_at: Utc::now().to_rfc3339(),
        }
    }

    /// The same, for a payload that is a summary rather than the whole thing.
    pub fn summary(key: impl Into<String>, payload: serde_json::Value) -> Self {
        Self {
            complete: false,
            ..Self::of(key, payload)
        }
    }

    /// The same, with a state this kind does not carry in its payload.
    pub fn with_state(mut self, state: impl Into<String>) -> Self {
        self.state = Some(state.into());
        self
    }
}

/// What changed since a reader last looked.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Changes {
    /// These scopes moved, and `latest` is the change to ask after next time.
    Moved { scopes: Vec<String>, latest: i64 },
    /// The reader is further behind than the log reaches: anything may have
    /// moved. Never reported as "nothing moved", which would leave a page
    /// showing what is no longer held.
    Lost { latest: i64 },
}

#[derive(Clone)]
pub struct Mirror {
    pool: r2d2::Pool<SqliteConnectionManager>,
    path: PathBuf,
}

impl std::fmt::Debug for Mirror {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Mirror").field("path", &self.path).finish()
    }
}

impl Mirror {
    /// The machine's mirror, in `$PRJ/.state`.
    pub fn of(prj: &wsb_prj::Prj) -> Result<Self> {
        Self::open(&prj.state_dir().join(FILE))
    }

    /// The mirror in one file, created if it does not exist.
    ///
    /// A file written by another version of this schema is refused: reading
    /// one optimistically is how a page comes to render half a board.
    pub fn open(path: &Path) -> Result<Self> {
        if let Some(parent) = path.parent() {
            std::fs::create_dir_all(parent)
                .with_context(|| format!("cannot create {}", parent.display()))?;
        }
        let manager = SqliteConnectionManager::file(path).with_init(|connection| {
            // Several processes share this file. WAL lets readers proceed
            // while one writes, and the busy timeout makes a writer wait its
            // turn rather than fail on a lock another process holds for a
            // millisecond.
            connection.busy_timeout(Duration::from_secs(10))?;
            connection.pragma_update(None, "journal_mode", "WAL")?;
            connection.pragma_update(None, "synchronous", "NORMAL")?;
            Ok(())
        });
        let pool = r2d2::Pool::builder()
            .max_size(4)
            .build(manager)
            .with_context(|| format!("cannot open the GitHub mirror at {}", path.display()))?;
        let mirror = Self {
            pool,
            path: path.to_path_buf(),
        };
        mirror.migrate()?;
        Ok(mirror)
    }

    pub fn path(&self) -> &Path {
        &self.path
    }

    fn migrate(&self) -> Result<()> {
        let mut conn = self.connection()?;
        let transaction = conn.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
        let version: i64 = transaction.query_row("PRAGMA user_version", [], |row| row.get(0))?;
        match version {
            0 => {
                transaction.execute_batch(include_str!("schema.sql"))?;
                transaction.pragma_update(None, "user_version", SCHEMA_VERSION)?;
            }
            SCHEMA_VERSION => {}
            other => bail!(
                "the GitHub mirror at {} is schema version {other}, and this build reads version \
                 {SCHEMA_VERSION}; upgrade the program that reads it",
                self.path.display()
            ),
        }
        transaction.commit()?;
        Ok(())
    }

    fn connection(&self) -> Result<PooledConnection<SqliteConnectionManager>> {
        self.pool
            .get()
            .with_context(|| format!("cannot reach the GitHub mirror at {}", self.path.display()))
    }

    // ---------------------------------------------------------------- state

    pub fn state(&self, scope: &Scope) -> Result<SyncState> {
        let conn = self.connection()?;
        let state = conn
            .query_row(
                "SELECT etag, cursor, synced_at, attempted_at, last_error
                 FROM github_sync_state WHERE scope = ?1",
                params![scope.key()],
                |row| {
                    Ok(SyncState {
                        etag: row.get(0)?,
                        cursor: row.get(1)?,
                        synced_at: row.get(2)?,
                        attempted_at: row.get(3)?,
                        last_error: row.get(4)?,
                    })
                },
            )
            .optional()?;
        Ok(state.unwrap_or_default())
    }

    /// Every scope this mirror knows about.
    pub fn states(&self) -> Result<BTreeMap<String, SyncState>> {
        let conn = self.connection()?;
        let mut statement = conn.prepare(
            "SELECT scope, etag, cursor, synced_at, attempted_at, last_error
             FROM github_sync_state",
        )?;
        let rows = statement.query_map([], |row| {
            Ok((
                row.get::<_, String>(0)?,
                SyncState {
                    etag: row.get(1)?,
                    cursor: row.get(2)?,
                    synced_at: row.get(3)?,
                    attempted_at: row.get(4)?,
                    last_error: row.get(5)?,
                },
            ))
        })?;
        let mut out = BTreeMap::new();
        for row in rows {
            let (scope, state) = row?;
            out.insert(scope, state);
        }
        Ok(out)
    }

    /// Record a sync that worked.
    ///
    /// Clears `last_error`: the scope is current, and a stale error beside
    /// current data is a page reporting a problem that is over.
    pub fn synced(&self, scope: &Scope, etag: Option<&str>, cursor: Option<&str>) -> Result<()> {
        let now = Utc::now().to_rfc3339();
        let conn = self.connection()?;
        conn.execute(
            "INSERT INTO github_sync_state (scope, etag, cursor, synced_at, attempted_at, last_error)
             VALUES (?1, ?2, ?3, ?4, ?4, NULL)
             ON CONFLICT(scope) DO UPDATE SET
                 -- A conditional request that answered 304 carries no ETag of
                 -- its own; the one we sent is still the current one.
                 etag = COALESCE(excluded.etag, github_sync_state.etag),
                 cursor = COALESCE(excluded.cursor, github_sync_state.cursor),
                 synced_at = excluded.synced_at,
                 attempted_at = excluded.attempted_at,
                 last_error = NULL",
            params![scope.key(), etag, cursor, now],
        )?;
        Ok(())
    }

    /// Record an attempt that failed.
    ///
    /// Leaves `synced_at`, the ETag and the cursor alone: the data this scope
    /// holds is still the data GitHub last gave it, and the next attempt
    /// should still be conditional.
    pub fn failed(&self, scope: &Scope, error: &str) -> Result<()> {
        let now = Utc::now().to_rfc3339();
        let conn = self.connection()?;
        conn.execute(
            "INSERT INTO github_sync_state (scope, attempted_at, last_error)
             VALUES (?1, ?2, ?3)
             ON CONFLICT(scope) DO UPDATE SET
                 attempted_at = excluded.attempted_at,
                 last_error = excluded.last_error",
            params![scope.key(), now, error],
        )?;
        Ok(())
    }

    /// Put a scope in the sweep before it has ever been read.
    ///
    /// What opening a comment thread does: a thread nobody has opened is not
    /// worth a request per sweep, and one somebody opened is. Recorded as a
    /// scope with no data and no attempt, which is exactly "not read yet".
    pub fn hold(&self, scope: &Scope) -> Result<()> {
        let conn = self.connection()?;
        conn.execute(
            "INSERT INTO github_sync_state (scope) VALUES (?1) ON CONFLICT(scope) DO NOTHING",
            params![scope.key()],
        )?;
        Ok(())
    }

    /// Forget a scope entirely, data and state.
    pub fn forget(&self, scope: &Scope) -> Result<()> {
        let mut conn = self.connection()?;
        let transaction = conn.transaction()?;
        let removed = transaction.execute(
            "DELETE FROM github_document WHERE scope = ?1",
            params![scope.key()],
        )?;
        transaction.execute(
            "DELETE FROM github_sync_state WHERE scope = ?1",
            params![scope.key()],
        )?;
        if removed > 0 {
            record_change(&transaction, scope)?;
        }
        transaction.commit()?;
        Ok(())
    }

    /// Forget everything about one repository.
    ///
    /// For a repository no project has any more: its issues, its pull
    /// requests, its releases and everything else go together, and a list of
    /// kinds to delete would be a list to keep in step with every kind added
    /// afterwards.
    pub fn forget_repository(&self, repo: &str) -> Result<usize> {
        let scopes: Vec<Scope> = self
            .states()?
            .keys()
            .filter_map(|key| Scope::parse(key))
            .filter(|scope| scope.repository() == Some(repo))
            .collect();
        for scope in &scopes {
            self.forget(scope)?;
        }
        Ok(scopes.len())
    }

    // ------------------------------------------------------------ documents

    /// Write what a fetch returned, and say whether anything changed.
    ///
    /// An AUTHORITATIVE scope replaces its contents: what the answer does not
    /// mention no longer exists. An incremental one only adds and updates,
    /// because its answer is "what changed", and clearing on that empties the
    /// mirror every time nothing did.
    pub fn put(&self, scope: &Scope, documents: &[Document]) -> Result<bool> {
        let mut conn = self.connection()?;
        let transaction = conn.transaction()?;
        let mut changed = false;
        if scope.is_authoritative() {
            changed |= prune_scope(&transaction, scope, documents)?;
        }
        for document in documents {
            changed |= put_document(&transaction, scope, document)?;
        }
        if changed {
            record_change(&transaction, scope)?;
        }
        transaction.commit()?;
        Ok(changed)
    }

    /// Write one document — what a create or an edit produced.
    ///
    /// Never prunes, whatever the scope: one document is not an answer about
    /// what exists.
    pub fn put_one(&self, scope: &Scope, document: Document) -> Result<bool> {
        let mut conn = self.connection()?;
        let transaction = conn.transaction()?;
        let changed = put_document(&transaction, scope, &document)?;
        if changed {
            record_change(&transaction, scope)?;
        }
        transaction.commit()?;
        Ok(changed)
    }

    /// Take one document out of a scope — what a deletion on GitHub produced.
    pub fn remove_one(&self, scope: &Scope, key: &str) -> Result<bool> {
        let mut conn = self.connection()?;
        let transaction = conn.transaction()?;
        let removed = transaction.execute(
            "DELETE FROM github_document WHERE scope = ?1 AND key = ?2",
            params![scope.key(), key],
        )? > 0;
        if removed {
            record_change(&transaction, scope)?;
        }
        transaction.commit()?;
        Ok(removed)
    }

    /// Atomically store one changed board and the complete card set GitHub
    /// returned after that board changed.
    ///
    /// A removed column changes both documents at once: the board loses an
    /// option and its cards lose that option's values. Committing those in two
    /// transactions creates a state no GitHub answer ever had if the second
    /// write fails, so this is one database act.
    pub fn put_project_write(
        &self,
        boards_scope: &Scope,
        board: Document,
        items_scope: &Scope,
        items: &[Document],
    ) -> Result<()> {
        if !matches!(boards_scope, Scope::Projects(_)) {
            bail!(
                "a project write needs a projects scope, got {}",
                boards_scope.key()
            );
        }
        match items_scope {
            Scope::ProjectItems(project) if project == &board.key => {}
            Scope::ProjectItems(project) => bail!(
                "project {} cannot be written with the cards of project {project}",
                board.key
            ),
            _ => bail!(
                "a project write needs a project-items scope, got {}",
                items_scope.key()
            ),
        }
        let mut conn = self.connection()?;
        let transaction = conn.transaction()?;
        if put_document(&transaction, boards_scope, &board)? {
            record_change(&transaction, boards_scope)?;
        }
        let mut items_changed = prune_scope(&transaction, items_scope, items)?;
        for item in items {
            items_changed |= put_document(&transaction, items_scope, item)?;
        }
        if items_changed {
            record_change(&transaction, items_scope)?;
        }
        transaction.commit()?;
        Ok(())
    }

    /// Everything one scope holds, newest movement first.
    pub fn documents(&self, scope: &Scope) -> Result<Vec<Document>> {
        let conn = self.connection()?;
        let mut statement = conn.prepare(
            "SELECT key, state, updated_at, payload_json, complete, fetched_at
             FROM github_document WHERE scope = ?1
             ORDER BY updated_at DESC, key",
        )?;
        let rows = statement.query_map(params![scope.key()], document_row)?;
        let mut documents = Vec::new();
        for row in rows {
            documents.push(read_document(scope, row?)?);
        }
        Ok(documents)
    }

    /// Just the payloads, which is what a reader serves.
    pub fn payloads(&self, scope: &Scope) -> Result<Vec<serde_json::Value>> {
        Ok(self
            .documents(scope)?
            .into_iter()
            .map(|document| document.payload)
            .collect())
    }

    /// One document, when the scope holds it.
    pub fn document(&self, scope: &Scope, key: &str) -> Result<Option<Document>> {
        let conn = self.connection()?;
        let row = conn
            .query_row(
                "SELECT key, state, updated_at, payload_json, complete, fetched_at
                 FROM github_document WHERE scope = ?1 AND key = ?2",
                params![scope.key(), key],
                document_row,
            )
            .optional()?;
        row.map(|row| read_document(scope, row)).transpose()
    }

    /// The single document of a scope that holds exactly one: a repository's
    /// metadata, a commit's checks, its traffic, the rate limit.
    pub fn only(&self, scope: &Scope) -> Result<Option<serde_json::Value>> {
        Ok(self
            .document(scope, SINGLETON)?
            .map(|document| document.payload))
    }

    /// Write the single document of such a scope.
    pub fn put_only(&self, scope: &Scope, payload: serde_json::Value) -> Result<bool> {
        self.put(scope, &[Document::of(SINGLETON, payload)])
    }

    /// How many documents of each state one scope holds.
    ///
    /// Counted in SQL: an explorer asks this for every row it draws, and
    /// reading every payload to answer would make a redraw proportional to the
    /// whole mirror.
    pub fn counts_by_state(&self, scope: &Scope) -> Result<BTreeMap<String, i64>> {
        let conn = self.connection()?;
        let mut statement = conn.prepare(
            "SELECT COALESCE(state, ''), COUNT(*) FROM github_document
             WHERE scope = ?1 GROUP BY 1",
        )?;
        let rows = statement.query_map(params![scope.key()], |row| {
            Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)?))
        })?;
        let mut counts = BTreeMap::new();
        for row in rows {
            let (state, count) = row?;
            counts.insert(state, count);
        }
        Ok(counts)
    }

    /// How many documents are open, per repository, across every scope of one
    /// kind (`issues` or `pulls`).
    ///
    /// One query for a whole tree rather than one per row.
    pub fn open_counts(&self, kind: &str) -> Result<BTreeMap<String, i64>> {
        let conn = self.connection()?;
        let mut statement = conn.prepare(
            "SELECT scope, COUNT(*) FROM github_document
             WHERE scope LIKE ?1 AND state = 'open' GROUP BY scope",
        )?;
        let rows = statement.query_map(params![format!("{kind}:%")], |row| {
            Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)?))
        })?;
        let mut counts = BTreeMap::new();
        for row in rows {
            let (scope, count) = row?;
            let repo = Scope::parse(&scope)
                .and_then(|scope| scope.repository().map(str::to_string))
                .ok_or_else(|| anyhow::anyhow!("the mirror holds a scope it cannot read: {scope}"))?;
            counts.insert(repo, count);
        }
        Ok(counts)
    }

    // -------------------------------------------------------------- changes

    /// The newest change recorded, or 0 when nothing ever changed.
    pub fn latest_change(&self) -> Result<i64> {
        let conn = self.connection()?;
        Ok(conn.query_row(
            "SELECT COALESCE(MAX(seq), 0) FROM github_change",
            [],
            |row| row.get(0),
        )?)
    }

    /// Which scopes changed after change `since`, written by whichever process.
    pub fn changes_since(&self, since: i64) -> Result<Changes> {
        let conn = self.connection()?;
        let (oldest, latest): (Option<i64>, i64) = conn.query_row(
            "SELECT MIN(seq), COALESCE(MAX(seq), 0) FROM github_change",
            [],
            |row| Ok((row.get(0)?, row.get(1)?)),
        )?;
        if latest <= since {
            return Ok(Changes::Moved {
                scopes: Vec::new(),
                latest,
            });
        }
        if let Some(oldest) = oldest {
            if since + 1 < oldest {
                return Ok(Changes::Lost { latest });
            }
        }
        let mut statement =
            conn.prepare("SELECT DISTINCT scope FROM github_change WHERE seq > ?1 ORDER BY scope")?;
        let scopes = statement
            .query_map(params![since], |row| row.get::<_, String>(0))?
            .collect::<Result<Vec<_>, _>>()?;
        Ok(Changes::Moved { scopes, latest })
    }
}

type DocumentRow = (String, Option<String>, Option<String>, String, i64, String);

fn document_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<DocumentRow> {
    Ok((
        row.get(0)?,
        row.get(1)?,
        row.get(2)?,
        row.get(3)?,
        row.get(4)?,
        row.get(5)?,
    ))
}

fn read_document(scope: &Scope, row: DocumentRow) -> Result<Document> {
    let (key, state, updated_at, payload, complete, fetched_at) = row;
    Ok(Document {
        key,
        payload: serde_json::from_str(&payload)
            .with_context(|| format!("a stored {} is not readable", scope.key()))?,
        state,
        updated_at,
        complete: complete != 0,
        fetched_at,
    })
}

fn record_change(transaction: &Transaction<'_>, scope: &Scope) -> Result<()> {
    transaction.execute(
        "INSERT INTO github_change (scope, changed_at) VALUES (?1, ?2)",
        params![scope.key(), Utc::now().to_rfc3339()],
    )?;
    transaction.execute(
        "DELETE FROM github_change WHERE seq <= (SELECT MAX(seq) FROM github_change) - ?1",
        params![CHANGES_KEPT],
    )?;
    Ok(())
}

/// Remove what an authoritative answer no longer mentions. `true` when
/// anything went.
fn prune_scope(transaction: &Transaction<'_>, scope: &Scope, documents: &[Document]) -> Result<bool> {
    let present: BTreeSet<&str> = documents
        .iter()
        .map(|document| document.key.as_str())
        .collect();
    let held: Vec<String> = {
        let mut statement =
            transaction.prepare("SELECT key FROM github_document WHERE scope = ?1")?;
        let rows = statement.query_map(params![scope.key()], |row| row.get::<_, String>(0))?;
        rows.collect::<Result<Vec<_>, _>>()?
    };
    let mut removed = false;
    for key in held {
        if !present.contains(key.as_str()) {
            transaction.execute(
                "DELETE FROM github_document WHERE scope = ?1 AND key = ?2",
                params![scope.key(), key],
            )?;
            removed = true;
        }
    }
    Ok(removed)
}

/// Write one document. `true` when what is held is now different.
///
/// Compared rather than assumed: a sync after a restart has no ETag to
/// present, so it is answered with the same content again — and recording
/// that as a change would tell every open page to re-read on a timer.
fn put_document(transaction: &Transaction<'_>, scope: &Scope, document: &Document) -> Result<bool> {
    let payload = serde_json::to_string(&document.payload)
        .with_context(|| format!("cannot store {}/{}", scope.key(), document.key))?;
    let held: Option<(Option<String>, Option<String>, String, i64)> = transaction
        .query_row(
            "SELECT state, updated_at, payload_json, complete FROM github_document
             WHERE scope = ?1 AND key = ?2",
            params![scope.key(), document.key],
            |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
        )
        .optional()?;
    let changed = match &held {
        None => true,
        Some((state, updated_at, held_payload, complete)) => {
            // A summary never replaces a whole payload, so over a whole one
            // it can change only the columns beside it.
            let payload_moves = document.complete || *complete == 0;
            *state != document.state
                || *updated_at != document.updated_at
                || (payload_moves && *held_payload != payload)
        }
    };
    transaction.execute(
        "INSERT INTO github_document
             (scope, key, state, updated_at, payload_json, complete, fetched_at)
         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)
         ON CONFLICT(scope, key) DO UPDATE SET
             state = excluded.state,
             updated_at = excluded.updated_at,
             -- A summary must not overwrite a payload already known to be
             -- whole: list endpoints omit what an editor opens.
             payload_json = CASE
                 WHEN excluded.complete = 1 OR github_document.complete = 0
                 THEN excluded.payload_json
                 ELSE github_document.payload_json
             END,
             complete = MAX(excluded.complete, github_document.complete),
             fetched_at = excluded.fetched_at",
        params![
            scope.key(),
            document.key,
            document.state,
            document.updated_at,
            payload,
            document.complete as i64,
            document.fetched_at
        ],
    )?;
    Ok(changed)
}

/// The key a scope that holds exactly one document stores it under.
pub const SINGLETON: &str = "-";

/// Whether a gist payload carries file contents rather than just their names.
///
/// GitHub's list endpoint returns files with `content` absent; the single-gist
/// endpoint fills them in. A gist with no files at all carries nothing either
/// way, and is complete as it stands.
pub fn gist_is_whole(gist: &GitHubGist) -> bool {
    gist.files.is_empty() || gist.files.values().all(|file| file.content.is_some())
}

/// The newest `updated_at` in a set, which becomes the next fetch's `since`.
///
/// GitHub's `since` is inclusive, so the item that set the mark comes back
/// with the next fetch and is written over itself. One row rewritten beats one
/// update missed.
pub fn newest_update<'a>(documents: impl IntoIterator<Item = &'a Document>) -> Option<String> {
    documents
        .into_iter()
        .filter_map(|document| document.updated_at.clone())
        .max()
}

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