Skip to main content

wsb_github/
mirror.rs

1//! The machine's copy of what GitHub said.
2//!
3//! # Why a mirror rather than a cache
4//!
5//! Every list used to be fetched from the API when a page opened. That made
6//! the page as fast as the network, as available as the credential, and as
7//! honest as the last request — and when a request failed, the list rendered
8//! EMPTY. A missing token, a rate limit or a closed laptop each produced "you
9//! have no gists", which is a lie told confidently.
10//!
11//! So a copy is kept and every reader reads the copy. Reads never touch the
12//! network, never need a token, and never fail because GitHub is having a
13//! morning. What a read CAN say is how old its answer is and whether the last
14//! refresh failed, which is the honest version of the same page.
15//!
16//! # One per machine
17//!
18//! GitHub's repositories, gists, notifications and project boards belong to
19//! accounts, not to any checkout, and one board spans many repositories. So
20//! there is one mirror for every project under `$PRJ`, in `$PRJ/.state`, and
21//! every process that reads or writes GitHub through this crate shares it.
22//!
23//! # What is stored
24//!
25//! Whole payloads, as GitHub sent them. The columns beside them exist only to
26//! be queried on — state, update time — and everything a reader renders comes
27//! out of the payload.
28
29use std::collections::{BTreeMap, BTreeSet};
30use std::path::{Path, PathBuf};
31use std::time::Duration;
32
33use anyhow::{bail, Context, Result};
34use chrono::Utc;
35use r2d2::PooledConnection;
36use r2d2_sqlite::SqliteConnectionManager;
37use rusqlite::{params, OptionalExtension, Transaction};
38use serde::{Deserialize, Serialize};
39
40use crate::client::GitHubGist;
41
42/// The schema version this build reads and writes.
43const SCHEMA_VERSION: i64 = 1;
44
45/// How many changes the change log keeps. A reader further behind than this
46/// re-reads everything rather than missing what fell off the end.
47const CHANGES_KEPT: i64 = 10_000;
48
49/// What one mirrored thing is called, and what a sync is about.
50///
51/// One vocabulary for every kind: the sync fetches, compares and records them
52/// identically, and a vocabulary per kind would mean a copy of every query
53/// that asks what is stale, what failed, or what moved.
54///
55/// The string form is what the database and every change notice carry, so it
56/// is part of the contract rather than a debugging convenience.
57#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
58pub enum Scope {
59    /// One repository's issues, as `owner/repo`.
60    Issues(String),
61    /// One repository's pull requests.
62    Pulls(String),
63    /// What CI says about one commit of one repository.
64    ///
65    /// Per commit, because the same repository is checked out at different
66    /// commits in different worktrees, and CI for the work in front of a
67    /// person is CI for the commit that worktree is on.
68    Checks { repo: String, commit: String },
69    Releases(String),
70    Tags(String),
71    /// What the repository IS: private or public, archived, its topics, its
72    /// stars.
73    Repository(String),
74    Milestones(String),
75    Labels(String),
76    /// Views and clones. Needs push access, so this is the one scope whose
77    /// refusal is ordinary.
78    Traffic(String),
79    /// Comments on one issue or pull request: `owner/repo#number`.
80    Comments { repo: String, number: i64 },
81    /// The authenticated user's gists.
82    Gists,
83    /// What GitHub thinks needs this person, across every repository.
84    Notifications,
85    /// How much API budget is left.
86    RateLimit,
87    /// Who the token belongs to, and the organisations it is a member of —
88    /// the owners whose boards are this person's without anybody naming them.
89    Viewer,
90    /// Which project boards one owner has.
91    Projects(String),
92    /// One board's cards, by the board's node id.
93    ProjectItems(String),
94}
95
96impl Scope {
97    pub fn key(&self) -> String {
98        match self {
99            Self::Issues(repo) => format!("issues:{repo}"),
100            Self::Pulls(repo) => format!("pulls:{repo}"),
101            Self::Checks { repo, commit } => format!("checks:{repo}@{commit}"),
102            Self::Releases(repo) => format!("releases:{repo}"),
103            Self::Tags(repo) => format!("tags:{repo}"),
104            Self::Repository(repo) => format!("repository:{repo}"),
105            Self::Milestones(repo) => format!("milestones:{repo}"),
106            Self::Labels(repo) => format!("labels:{repo}"),
107            Self::Traffic(repo) => format!("traffic:{repo}"),
108            Self::Comments { repo, number } => format!("comments:{repo}#{number}"),
109            Self::Gists => "gists".to_string(),
110            Self::Notifications => "notifications".to_string(),
111            Self::RateLimit => "rate_limit".to_string(),
112            Self::Viewer => "viewer".to_string(),
113            Self::Projects(login) => format!("projects:{login}"),
114            Self::ProjectItems(id) => format!("project_items:{id}"),
115        }
116    }
117
118    /// The repository this scope is about, when it is about one.
119    pub fn repository(&self) -> Option<&str> {
120        match self {
121            Self::Issues(repo)
122            | Self::Pulls(repo)
123            | Self::Checks { repo, .. }
124            | Self::Releases(repo)
125            | Self::Tags(repo)
126            | Self::Repository(repo)
127            | Self::Milestones(repo)
128            | Self::Labels(repo)
129            | Self::Traffic(repo)
130            | Self::Comments { repo, .. } => Some(repo),
131            Self::Gists | Self::Notifications | Self::RateLimit | Self::Viewer => None,
132            Self::Projects(_) | Self::ProjectItems(_) => None,
133        }
134    }
135
136    /// Read one back from what the database and a change notice carry.
137    ///
138    /// Every scope round-trips, or the process that wrote a row cannot say
139    /// what the row is about — see the round-trip test.
140    pub fn parse(key: &str) -> Option<Self> {
141        if let Some((kind, rest)) = key.split_once(':') {
142            if rest.is_empty() {
143                return None;
144            }
145            return Some(match kind {
146                "issues" => Self::Issues(rest.to_string()),
147                "pulls" => Self::Pulls(rest.to_string()),
148                "checks" => {
149                    let (repo, commit) = rest.rsplit_once('@')?;
150                    if repo.is_empty() || commit.is_empty() {
151                        return None;
152                    }
153                    Self::Checks {
154                        repo: repo.to_string(),
155                        commit: commit.to_string(),
156                    }
157                }
158                "releases" => Self::Releases(rest.to_string()),
159                "tags" => Self::Tags(rest.to_string()),
160                "repository" => Self::Repository(rest.to_string()),
161                "milestones" => Self::Milestones(rest.to_string()),
162                "labels" => Self::Labels(rest.to_string()),
163                "traffic" => Self::Traffic(rest.to_string()),
164                "comments" => {
165                    let (repo, number) = rest.rsplit_once('#')?;
166                    Self::Comments {
167                        repo: repo.to_string(),
168                        number: number.parse().ok()?,
169                    }
170                }
171                "projects" => Self::Projects(rest.to_string()),
172                "project_items" => Self::ProjectItems(rest.to_string()),
173                _ => return None,
174            });
175        }
176        match key {
177            "gists" => Some(Self::Gists),
178            "notifications" => Some(Self::Notifications),
179            "rate_limit" => Some(Self::RateLimit),
180            "viewer" => Some(Self::Viewer),
181            _ => None,
182        }
183    }
184
185    /// What a person calls it, for a message about it.
186    pub fn label(&self) -> String {
187        match self {
188            Self::Comments { repo, number } => format!("{repo}#{number} comments"),
189            Self::Checks { repo, commit } => {
190                format!("checks of {repo} at {}", &commit[..commit.len().min(12)])
191            }
192            Self::Gists => "gists".to_string(),
193            Self::Notifications => "notifications".to_string(),
194            Self::RateLimit => "the rate limit".to_string(),
195            Self::Viewer => "the token's account".to_string(),
196            Self::Projects(login) => format!("{login}'s projects"),
197            Self::ProjectItems(id) => format!("project {id}"),
198            other => other.repository().unwrap_or_default().to_string(),
199        }
200    }
201
202    /// Whether a whole answer replaces what is held.
203    ///
204    /// A LIST endpoint says what exists: a gist absent from it was deleted, a
205    /// release absent from it was removed. An INCREMENTAL one says only what
206    /// changed, so what it does not mention is still true — and clearing on
207    /// one of those empties the mirror every time nothing happened.
208    ///
209    /// Issues and pull requests are fetched with `since`, so they are
210    /// incremental. Everything else is asked for whole.
211    pub fn is_authoritative(&self) -> bool {
212        !matches!(self, Self::Issues(_) | Self::Pulls(_))
213    }
214}
215
216/// Where one scope stands: how current it is, and what went wrong last.
217///
218/// `synced_at` and `last_error` are both meaningful at once, and that pairing
219/// is the point: a scope that synced an hour ago and failed a minute ago has
220/// data worth showing AND a problem worth reporting.
221#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
222pub struct SyncState {
223    /// GitHub's ETag for the last response, replayed as `If-None-Match`.
224    pub etag: Option<String>,
225    /// The newest `updated_at` seen, for incremental fetches.
226    pub cursor: Option<String>,
227    /// When the last SUCCESSFUL sync finished.
228    pub synced_at: Option<String>,
229    /// When the last attempt was made, successful or not.
230    pub attempted_at: Option<String>,
231    pub last_error: Option<String>,
232}
233
234impl SyncState {
235    /// Whether this scope has ever been mirrored.
236    pub fn ever_synced(&self) -> bool {
237        self.synced_at.is_some()
238    }
239
240    /// What a reader is told about how current an answer is.
241    pub fn standing(&self) -> Standing {
242        Standing {
243            synced_at: self.synced_at.clone(),
244            last_error: self.last_error.clone(),
245            ever_synced: self.ever_synced(),
246        }
247    }
248}
249
250/// How current one scope's answer is, as every reader is told it.
251///
252/// "Never read" and "read, and empty" are different, and so are "read an hour
253/// ago" and "could not be read a minute ago": the three fields say which.
254#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
255pub struct Standing {
256    pub synced_at: Option<String>,
257    pub last_error: Option<String>,
258    pub ever_synced: bool,
259}
260
261/// One document, as GitHub last described it.
262#[derive(Debug, Clone, PartialEq, Eq)]
263pub struct Document {
264    /// What identifies it inside its scope: an issue number, a gist id, a tag
265    /// name, a project item id.
266    pub key: String,
267    pub payload: serde_json::Value,
268    pub state: Option<String>,
269    pub updated_at: Option<String>,
270    /// Whether the payload is the whole thing rather than a list summary.
271    pub complete: bool,
272    pub fetched_at: String,
273}
274
275impl Document {
276    /// One document from a value GitHub sent, with the columns a query needs
277    /// pulled out of it.
278    ///
279    /// The extraction is by field NAME, which is the same name in every kind
280    /// GitHub returns — `state` and `updated_at`. A kind that carries neither
281    /// stores neither, and is read back whole.
282    pub fn of(key: impl Into<String>, payload: serde_json::Value) -> Self {
283        let state = payload
284            .get("state")
285            .and_then(|state| state.as_str())
286            .map(str::to_string);
287        let updated_at = payload
288            .get("updated_at")
289            .or_else(|| payload.get("updatedAt"))
290            .or_else(|| payload.get("published_at"))
291            .and_then(|time| time.as_str())
292            .map(str::to_string);
293        Self {
294            key: key.into(),
295            payload,
296            state,
297            updated_at,
298            complete: true,
299            fetched_at: Utc::now().to_rfc3339(),
300        }
301    }
302
303    /// The same, for a payload that is a summary rather than the whole thing.
304    pub fn summary(key: impl Into<String>, payload: serde_json::Value) -> Self {
305        Self {
306            complete: false,
307            ..Self::of(key, payload)
308        }
309    }
310
311    /// The same, with a state this kind does not carry in its payload.
312    pub fn with_state(mut self, state: impl Into<String>) -> Self {
313        self.state = Some(state.into());
314        self
315    }
316}
317
318/// What changed since a reader last looked.
319#[derive(Debug, Clone, PartialEq, Eq)]
320pub enum Changes {
321    /// These scopes moved, and `latest` is the change to ask after next time.
322    Moved { scopes: Vec<String>, latest: i64 },
323    /// The reader is further behind than the log reaches: anything may have
324    /// moved. Never reported as "nothing moved", which would leave a page
325    /// showing what is no longer held.
326    Lost { latest: i64 },
327}
328
329#[derive(Clone)]
330pub struct Mirror {
331    pool: r2d2::Pool<SqliteConnectionManager>,
332    path: PathBuf,
333}
334
335impl std::fmt::Debug for Mirror {
336    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
337        f.debug_struct("Mirror").field("path", &self.path).finish()
338    }
339}
340
341impl Mirror {
342    /// The machine's mirror, in the workshop's database
343    /// ([`wsb_prj::database_path`]).
344    pub fn of_workshop() -> Result<Self> {
345        Self::open(&wsb_prj::database_path()?)
346    }
347
348    /// The mirror's tables in one database file, created if they are not
349    /// there. The file is shared: the mirror's tables are `github_*`, and its
350    /// schema version is its own table's, never the file's `user_version`.
351    ///
352    /// A file written by another version of this schema is refused: reading
353    /// one optimistically is how a page comes to render half a board.
354    pub fn open(path: &Path) -> Result<Self> {
355        if let Some(parent) = path.parent() {
356            std::fs::create_dir_all(parent)
357                .with_context(|| format!("cannot create {}", parent.display()))?;
358        }
359        let manager = SqliteConnectionManager::file(path).with_init(|connection| {
360            // Several processes share this file. WAL lets readers proceed
361            // while one writes, and the busy timeout makes a writer wait its
362            // turn rather than fail on a lock another process holds for a
363            // millisecond.
364            connection.busy_timeout(Duration::from_secs(10))?;
365            connection.pragma_update(None, "journal_mode", "WAL")?;
366            connection.pragma_update(None, "synchronous", "NORMAL")?;
367            Ok(())
368        });
369        let pool = r2d2::Pool::builder()
370            .max_size(4)
371            .build(manager)
372            .with_context(|| format!("cannot open the GitHub mirror at {}", path.display()))?;
373        let mirror = Self {
374            pool,
375            path: path.to_path_buf(),
376        };
377        mirror.migrate()?;
378        Ok(mirror)
379    }
380
381    pub fn path(&self) -> &Path {
382        &self.path
383    }
384
385    fn migrate(&self) -> Result<()> {
386        let mut conn = self.connection()?;
387        let transaction = conn.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
388        let held: bool = transaction.query_row(
389            "SELECT EXISTS (SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'github_schema')",
390            [],
391            |row| row.get(0),
392        )?;
393        let version: i64 = if held {
394            transaction.query_row("SELECT version FROM github_schema", [], |row| row.get(0))?
395        } else {
396            0
397        };
398        match version {
399            0 => {
400                transaction.execute_batch(include_str!("schema.sql"))?;
401                transaction.execute_batch("CREATE TABLE github_schema (version INTEGER NOT NULL)")?;
402                transaction.execute("INSERT INTO github_schema (version) VALUES (?1)", params![SCHEMA_VERSION])?;
403            }
404            SCHEMA_VERSION => {}
405            other => bail!(
406                "the GitHub mirror at {} is schema version {other}, and this build reads version \
407                 {SCHEMA_VERSION}; upgrade the program that reads it",
408                self.path.display()
409            ),
410        }
411        transaction.commit()?;
412        Ok(())
413    }
414
415    fn connection(&self) -> Result<PooledConnection<SqliteConnectionManager>> {
416        self.pool
417            .get()
418            .with_context(|| format!("cannot reach the GitHub mirror at {}", self.path.display()))
419    }
420
421    // ---------------------------------------------------------------- state
422
423    pub fn state(&self, scope: &Scope) -> Result<SyncState> {
424        let conn = self.connection()?;
425        let state = conn
426            .query_row(
427                "SELECT etag, cursor, synced_at, attempted_at, last_error
428                 FROM github_sync_state WHERE scope = ?1",
429                params![scope.key()],
430                |row| {
431                    Ok(SyncState {
432                        etag: row.get(0)?,
433                        cursor: row.get(1)?,
434                        synced_at: row.get(2)?,
435                        attempted_at: row.get(3)?,
436                        last_error: row.get(4)?,
437                    })
438                },
439            )
440            .optional()?;
441        Ok(state.unwrap_or_default())
442    }
443
444    /// Every scope this mirror knows about.
445    pub fn states(&self) -> Result<BTreeMap<String, SyncState>> {
446        let conn = self.connection()?;
447        let mut statement = conn.prepare(
448            "SELECT scope, etag, cursor, synced_at, attempted_at, last_error
449             FROM github_sync_state",
450        )?;
451        let rows = statement.query_map([], |row| {
452            Ok((
453                row.get::<_, String>(0)?,
454                SyncState {
455                    etag: row.get(1)?,
456                    cursor: row.get(2)?,
457                    synced_at: row.get(3)?,
458                    attempted_at: row.get(4)?,
459                    last_error: row.get(5)?,
460                },
461            ))
462        })?;
463        let mut out = BTreeMap::new();
464        for row in rows {
465            let (scope, state) = row?;
466            out.insert(scope, state);
467        }
468        Ok(out)
469    }
470
471    /// Record a sync that worked.
472    ///
473    /// Clears `last_error`: the scope is current, and a stale error beside
474    /// current data is a page reporting a problem that is over.
475    pub fn synced(&self, scope: &Scope, etag: Option<&str>, cursor: Option<&str>) -> Result<()> {
476        let now = Utc::now().to_rfc3339();
477        let conn = self.connection()?;
478        conn.execute(
479            "INSERT INTO github_sync_state (scope, etag, cursor, synced_at, attempted_at, last_error)
480             VALUES (?1, ?2, ?3, ?4, ?4, NULL)
481             ON CONFLICT(scope) DO UPDATE SET
482                 -- A conditional request that answered 304 carries no ETag of
483                 -- its own; the one we sent is still the current one.
484                 etag = COALESCE(excluded.etag, github_sync_state.etag),
485                 cursor = COALESCE(excluded.cursor, github_sync_state.cursor),
486                 synced_at = excluded.synced_at,
487                 attempted_at = excluded.attempted_at,
488                 last_error = NULL",
489            params![scope.key(), etag, cursor, now],
490        )?;
491        Ok(())
492    }
493
494    /// Record an attempt that failed.
495    ///
496    /// Leaves `synced_at`, the ETag and the cursor alone: the data this scope
497    /// holds is still the data GitHub last gave it, and the next attempt
498    /// should still be conditional.
499    pub fn failed(&self, scope: &Scope, error: &str) -> Result<()> {
500        let now = Utc::now().to_rfc3339();
501        let conn = self.connection()?;
502        conn.execute(
503            "INSERT INTO github_sync_state (scope, attempted_at, last_error)
504             VALUES (?1, ?2, ?3)
505             ON CONFLICT(scope) DO UPDATE SET
506                 attempted_at = excluded.attempted_at,
507                 last_error = excluded.last_error",
508            params![scope.key(), now, error],
509        )?;
510        Ok(())
511    }
512
513    /// Put a scope in the sweep before it has ever been read.
514    ///
515    /// What opening a comment thread does: a thread nobody has opened is not
516    /// worth a request per sweep, and one somebody opened is. Recorded as a
517    /// scope with no data and no attempt, which is exactly "not read yet".
518    pub fn hold(&self, scope: &Scope) -> Result<()> {
519        let conn = self.connection()?;
520        conn.execute(
521            "INSERT INTO github_sync_state (scope) VALUES (?1) ON CONFLICT(scope) DO NOTHING",
522            params![scope.key()],
523        )?;
524        Ok(())
525    }
526
527    /// Forget a scope entirely, data and state.
528    pub fn forget(&self, scope: &Scope) -> Result<()> {
529        let mut conn = self.connection()?;
530        let transaction = conn.transaction()?;
531        let removed = transaction.execute(
532            "DELETE FROM github_document WHERE scope = ?1",
533            params![scope.key()],
534        )?;
535        transaction.execute(
536            "DELETE FROM github_sync_state WHERE scope = ?1",
537            params![scope.key()],
538        )?;
539        if removed > 0 {
540            record_change(&transaction, scope)?;
541        }
542        transaction.commit()?;
543        Ok(())
544    }
545
546    /// Forget everything about one repository.
547    ///
548    /// For a repository no project has any more: its issues, its pull
549    /// requests, its releases and everything else go together, and a list of
550    /// kinds to delete would be a list to keep in step with every kind added
551    /// afterwards.
552    pub fn forget_repository(&self, repo: &str) -> Result<usize> {
553        let scopes: Vec<Scope> = self
554            .states()?
555            .keys()
556            .filter_map(|key| Scope::parse(key))
557            .filter(|scope| scope.repository() == Some(repo))
558            .collect();
559        for scope in &scopes {
560            self.forget(scope)?;
561        }
562        Ok(scopes.len())
563    }
564
565    // ------------------------------------------------------------ documents
566
567    /// Write what a fetch returned, and say whether anything changed.
568    ///
569    /// An AUTHORITATIVE scope replaces its contents: what the answer does not
570    /// mention no longer exists. An incremental one only adds and updates,
571    /// because its answer is "what changed", and clearing on that empties the
572    /// mirror every time nothing did.
573    pub fn put(&self, scope: &Scope, documents: &[Document]) -> Result<bool> {
574        let mut conn = self.connection()?;
575        let transaction = conn.transaction()?;
576        let mut changed = false;
577        if scope.is_authoritative() {
578            changed |= prune_scope(&transaction, scope, documents)?;
579        }
580        for document in documents {
581            changed |= put_document(&transaction, scope, document)?;
582        }
583        if changed {
584            record_change(&transaction, scope)?;
585        }
586        transaction.commit()?;
587        Ok(changed)
588    }
589
590    /// Write one document — what a create or an edit produced.
591    ///
592    /// Never prunes, whatever the scope: one document is not an answer about
593    /// what exists.
594    pub fn put_one(&self, scope: &Scope, document: Document) -> Result<bool> {
595        let mut conn = self.connection()?;
596        let transaction = conn.transaction()?;
597        let changed = put_document(&transaction, scope, &document)?;
598        if changed {
599            record_change(&transaction, scope)?;
600        }
601        transaction.commit()?;
602        Ok(changed)
603    }
604
605    /// Take one document out of a scope — what a deletion on GitHub produced.
606    pub fn remove_one(&self, scope: &Scope, key: &str) -> Result<bool> {
607        let mut conn = self.connection()?;
608        let transaction = conn.transaction()?;
609        let removed = transaction.execute(
610            "DELETE FROM github_document WHERE scope = ?1 AND key = ?2",
611            params![scope.key(), key],
612        )? > 0;
613        if removed {
614            record_change(&transaction, scope)?;
615        }
616        transaction.commit()?;
617        Ok(removed)
618    }
619
620    /// Atomically store one changed board and the complete card set GitHub
621    /// returned after that board changed.
622    ///
623    /// A removed column changes both documents at once: the board loses an
624    /// option and its cards lose that option's values. Committing those in two
625    /// transactions creates a state no GitHub answer ever had if the second
626    /// write fails, so this is one database act.
627    pub fn put_project_write(
628        &self,
629        boards_scope: &Scope,
630        board: Document,
631        items_scope: &Scope,
632        items: &[Document],
633    ) -> Result<()> {
634        if !matches!(boards_scope, Scope::Projects(_)) {
635            bail!(
636                "a project write needs a projects scope, got {}",
637                boards_scope.key()
638            );
639        }
640        match items_scope {
641            Scope::ProjectItems(project) if project == &board.key => {}
642            Scope::ProjectItems(project) => bail!(
643                "project {} cannot be written with the cards of project {project}",
644                board.key
645            ),
646            _ => bail!(
647                "a project write needs a project-items scope, got {}",
648                items_scope.key()
649            ),
650        }
651        let mut conn = self.connection()?;
652        let transaction = conn.transaction()?;
653        if put_document(&transaction, boards_scope, &board)? {
654            record_change(&transaction, boards_scope)?;
655        }
656        let mut items_changed = prune_scope(&transaction, items_scope, items)?;
657        for item in items {
658            items_changed |= put_document(&transaction, items_scope, item)?;
659        }
660        if items_changed {
661            record_change(&transaction, items_scope)?;
662        }
663        transaction.commit()?;
664        Ok(())
665    }
666
667    /// Everything one scope holds, newest movement first.
668    pub fn documents(&self, scope: &Scope) -> Result<Vec<Document>> {
669        let conn = self.connection()?;
670        let mut statement = conn.prepare(
671            "SELECT key, state, updated_at, payload_json, complete, fetched_at
672             FROM github_document WHERE scope = ?1
673             ORDER BY updated_at DESC, key",
674        )?;
675        let rows = statement.query_map(params![scope.key()], document_row)?;
676        let mut documents = Vec::new();
677        for row in rows {
678            documents.push(read_document(scope, row?)?);
679        }
680        Ok(documents)
681    }
682
683    /// The newest `updated_at` any document of `scope` holds: for a board's
684    /// cards, when anything on the board last changed, as GitHub reports it.
685    /// `None` when the scope holds nothing that says.
686    pub fn newest_update_in(&self, scope: &Scope) -> Result<Option<String>> {
687        let conn = self.connection()?;
688        Ok(conn.query_row(
689            "SELECT MAX(updated_at) FROM github_document WHERE scope = ?1",
690            params![scope.key()],
691            |row| row.get::<_, Option<String>>(0),
692        )?)
693    }
694
695    /// Just the payloads, which is what a reader serves.
696    pub fn payloads(&self, scope: &Scope) -> Result<Vec<serde_json::Value>> {
697        Ok(self
698            .documents(scope)?
699            .into_iter()
700            .map(|document| document.payload)
701            .collect())
702    }
703
704    /// One document, when the scope holds it.
705    pub fn document(&self, scope: &Scope, key: &str) -> Result<Option<Document>> {
706        let conn = self.connection()?;
707        let row = conn
708            .query_row(
709                "SELECT key, state, updated_at, payload_json, complete, fetched_at
710                 FROM github_document WHERE scope = ?1 AND key = ?2",
711                params![scope.key(), key],
712                document_row,
713            )
714            .optional()?;
715        row.map(|row| read_document(scope, row)).transpose()
716    }
717
718    /// The single document of a scope that holds exactly one: a repository's
719    /// metadata, a commit's checks, its traffic, the rate limit.
720    pub fn only(&self, scope: &Scope) -> Result<Option<serde_json::Value>> {
721        Ok(self
722            .document(scope, SINGLETON)?
723            .map(|document| document.payload))
724    }
725
726    /// Write the single document of such a scope.
727    pub fn put_only(&self, scope: &Scope, payload: serde_json::Value) -> Result<bool> {
728        self.put(scope, &[Document::of(SINGLETON, payload)])
729    }
730
731    /// How many documents of each state one scope holds.
732    ///
733    /// Counted in SQL: an explorer asks this for every row it draws, and
734    /// reading every payload to answer would make a redraw proportional to the
735    /// whole mirror.
736    pub fn counts_by_state(&self, scope: &Scope) -> Result<BTreeMap<String, i64>> {
737        let conn = self.connection()?;
738        let mut statement = conn.prepare(
739            "SELECT COALESCE(state, ''), COUNT(*) FROM github_document
740             WHERE scope = ?1 GROUP BY 1",
741        )?;
742        let rows = statement.query_map(params![scope.key()], |row| {
743            Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)?))
744        })?;
745        let mut counts = BTreeMap::new();
746        for row in rows {
747            let (state, count) = row?;
748            counts.insert(state, count);
749        }
750        Ok(counts)
751    }
752
753    /// How many documents are open, per repository, across every scope of one
754    /// kind (`issues` or `pulls`).
755    ///
756    /// One query for a whole tree rather than one per row.
757    pub fn open_counts(&self, kind: &str) -> Result<BTreeMap<String, i64>> {
758        let conn = self.connection()?;
759        let mut statement = conn.prepare(
760            "SELECT scope, COUNT(*) FROM github_document
761             WHERE scope LIKE ?1 AND state = 'open' GROUP BY scope",
762        )?;
763        let rows = statement.query_map(params![format!("{kind}:%")], |row| {
764            Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)?))
765        })?;
766        let mut counts = BTreeMap::new();
767        for row in rows {
768            let (scope, count) = row?;
769            let repo = Scope::parse(&scope)
770                .and_then(|scope| scope.repository().map(str::to_string))
771                .ok_or_else(|| anyhow::anyhow!("the mirror holds a scope it cannot read: {scope}"))?;
772            counts.insert(repo, count);
773        }
774        Ok(counts)
775    }
776
777    // -------------------------------------------------------------- changes
778
779    /// The newest change recorded, or 0 when nothing ever changed.
780    pub fn latest_change(&self) -> Result<i64> {
781        let conn = self.connection()?;
782        Ok(conn.query_row(
783            "SELECT COALESCE(MAX(seq), 0) FROM github_change",
784            [],
785            |row| row.get(0),
786        )?)
787    }
788
789    /// Which scopes changed after change `since`, written by whichever process.
790    pub fn changes_since(&self, since: i64) -> Result<Changes> {
791        let conn = self.connection()?;
792        let (oldest, latest): (Option<i64>, i64) = conn.query_row(
793            "SELECT MIN(seq), COALESCE(MAX(seq), 0) FROM github_change",
794            [],
795            |row| Ok((row.get(0)?, row.get(1)?)),
796        )?;
797        if latest <= since {
798            return Ok(Changes::Moved {
799                scopes: Vec::new(),
800                latest,
801            });
802        }
803        if let Some(oldest) = oldest {
804            if since + 1 < oldest {
805                return Ok(Changes::Lost { latest });
806            }
807        }
808        let mut statement =
809            conn.prepare("SELECT DISTINCT scope FROM github_change WHERE seq > ?1 ORDER BY scope")?;
810        let scopes = statement
811            .query_map(params![since], |row| row.get::<_, String>(0))?
812            .collect::<Result<Vec<_>, _>>()?;
813        Ok(Changes::Moved { scopes, latest })
814    }
815}
816
817type DocumentRow = (String, Option<String>, Option<String>, String, i64, String);
818
819fn document_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<DocumentRow> {
820    Ok((
821        row.get(0)?,
822        row.get(1)?,
823        row.get(2)?,
824        row.get(3)?,
825        row.get(4)?,
826        row.get(5)?,
827    ))
828}
829
830fn read_document(scope: &Scope, row: DocumentRow) -> Result<Document> {
831    let (key, state, updated_at, payload, complete, fetched_at) = row;
832    Ok(Document {
833        key,
834        payload: serde_json::from_str(&payload)
835            .with_context(|| format!("a stored {} is not readable", scope.key()))?,
836        state,
837        updated_at,
838        complete: complete != 0,
839        fetched_at,
840    })
841}
842
843fn record_change(transaction: &Transaction<'_>, scope: &Scope) -> Result<()> {
844    transaction.execute(
845        "INSERT INTO github_change (scope, changed_at) VALUES (?1, ?2)",
846        params![scope.key(), Utc::now().to_rfc3339()],
847    )?;
848    transaction.execute(
849        "DELETE FROM github_change WHERE seq <= (SELECT MAX(seq) FROM github_change) - ?1",
850        params![CHANGES_KEPT],
851    )?;
852    Ok(())
853}
854
855/// Remove what an authoritative answer no longer mentions. `true` when
856/// anything went.
857fn prune_scope(transaction: &Transaction<'_>, scope: &Scope, documents: &[Document]) -> Result<bool> {
858    let present: BTreeSet<&str> = documents
859        .iter()
860        .map(|document| document.key.as_str())
861        .collect();
862    let held: Vec<String> = {
863        let mut statement =
864            transaction.prepare("SELECT key FROM github_document WHERE scope = ?1")?;
865        let rows = statement.query_map(params![scope.key()], |row| row.get::<_, String>(0))?;
866        rows.collect::<Result<Vec<_>, _>>()?
867    };
868    let mut removed = false;
869    for key in held {
870        if !present.contains(key.as_str()) {
871            transaction.execute(
872                "DELETE FROM github_document WHERE scope = ?1 AND key = ?2",
873                params![scope.key(), key],
874            )?;
875            removed = true;
876        }
877    }
878    Ok(removed)
879}
880
881/// Write one document. `true` when what is held is now different.
882///
883/// Compared rather than assumed: a sync after a restart has no ETag to
884/// present, so it is answered with the same content again — and recording
885/// that as a change would tell every open page to re-read on a timer.
886fn put_document(transaction: &Transaction<'_>, scope: &Scope, document: &Document) -> Result<bool> {
887    let payload = serde_json::to_string(&document.payload)
888        .with_context(|| format!("cannot store {}/{}", scope.key(), document.key))?;
889    let held: Option<(Option<String>, Option<String>, String, i64)> = transaction
890        .query_row(
891            "SELECT state, updated_at, payload_json, complete FROM github_document
892             WHERE scope = ?1 AND key = ?2",
893            params![scope.key(), document.key],
894            |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
895        )
896        .optional()?;
897    let changed = match &held {
898        None => true,
899        Some((state, updated_at, held_payload, complete)) => {
900            // A summary never replaces a whole payload, so over a whole one
901            // it can change only the columns beside it.
902            let payload_moves = document.complete || *complete == 0;
903            *state != document.state
904                || *updated_at != document.updated_at
905                || (payload_moves && *held_payload != payload)
906        }
907    };
908    transaction.execute(
909        "INSERT INTO github_document
910             (scope, key, state, updated_at, payload_json, complete, fetched_at)
911         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)
912         ON CONFLICT(scope, key) DO UPDATE SET
913             state = excluded.state,
914             updated_at = excluded.updated_at,
915             -- A summary must not overwrite a payload already known to be
916             -- whole: list endpoints omit what an editor opens.
917             payload_json = CASE
918                 WHEN excluded.complete = 1 OR github_document.complete = 0
919                 THEN excluded.payload_json
920                 ELSE github_document.payload_json
921             END,
922             complete = MAX(excluded.complete, github_document.complete),
923             fetched_at = excluded.fetched_at",
924        params![
925            scope.key(),
926            document.key,
927            document.state,
928            document.updated_at,
929            payload,
930            document.complete as i64,
931            document.fetched_at
932        ],
933    )?;
934    Ok(changed)
935}
936
937/// The key a scope that holds exactly one document stores it under.
938pub const SINGLETON: &str = "-";
939
940/// Whether a gist payload carries file contents rather than just their names.
941///
942/// GitHub's list endpoint returns files with `content` absent; the single-gist
943/// endpoint fills them in. A gist with no files at all carries nothing either
944/// way, and is complete as it stands.
945pub fn gist_is_whole(gist: &GitHubGist) -> bool {
946    gist.files.is_empty() || gist.files.values().all(|file| file.content.is_some())
947}
948
949/// The newest `updated_at` in a set, which becomes the next fetch's `since`.
950///
951/// GitHub's `since` is inclusive, so the item that set the mark comes back
952/// with the next fetch and is written over itself. One row rewritten beats one
953/// update missed.
954pub fn newest_update<'a>(documents: impl IntoIterator<Item = &'a Document>) -> Option<String> {
955    documents
956        .into_iter()
957        .filter_map(|document| document.updated_at.clone())
958        .max()
959}
960
961#[cfg(test)]
962#[path = "mirror_tests.rs"]
963mod tests;