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;
const SCHEMA_VERSION: i64 = 1;
const CHANGES_KEPT: i64 = 10_000;
pub const FILE: &str = "github.db";
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub enum Scope {
Issues(String),
Pulls(String),
Checks { repo: String, commit: String },
Releases(String),
Tags(String),
Repository(String),
Milestones(String),
Labels(String),
Traffic(String),
Comments { repo: String, number: i64 },
Gists,
Notifications,
RateLimit,
Projects(String),
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::Projects(login) => format!("projects:{login}"),
Self::ProjectItems(id) => format!("project_items:{id}"),
}
}
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 => None,
Self::Projects(_) | Self::ProjectItems(_) => None,
}
}
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),
_ => None,
}
}
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::Projects(login) => format!("{login}'s projects"),
Self::ProjectItems(id) => format!("project {id}"),
other => other.repository().unwrap_or_default().to_string(),
}
}
pub fn is_authoritative(&self) -> bool {
!matches!(self, Self::Issues(_) | Self::Pulls(_))
}
}
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct SyncState {
pub etag: Option<String>,
pub cursor: Option<String>,
pub synced_at: Option<String>,
pub attempted_at: Option<String>,
pub last_error: Option<String>,
}
impl SyncState {
pub fn ever_synced(&self) -> bool {
self.synced_at.is_some()
}
pub fn standing(&self) -> Standing {
Standing {
synced_at: self.synced_at.clone(),
last_error: self.last_error.clone(),
ever_synced: self.ever_synced(),
}
}
}
#[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,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Document {
pub key: String,
pub payload: serde_json::Value,
pub state: Option<String>,
pub updated_at: Option<String>,
pub complete: bool,
pub fetched_at: String,
}
impl Document {
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(),
}
}
pub fn summary(key: impl Into<String>, payload: serde_json::Value) -> Self {
Self {
complete: false,
..Self::of(key, payload)
}
}
pub fn with_state(mut self, state: impl Into<String>) -> Self {
self.state = Some(state.into());
self
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Changes {
Moved { scopes: Vec<String>, latest: i64 },
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 {
pub fn of(prj: &wsb_prj::Prj) -> Result<Self> {
Self::open(&prj.state_dir().join(FILE))
}
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| {
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()))
}
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())
}
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)
}
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(())
}
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(())
}
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(())
}
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(())
}
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())
}
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)
}
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)
}
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)
}
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(())
}
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)
}
pub fn payloads(&self, scope: &Scope) -> Result<Vec<serde_json::Value>> {
Ok(self
.documents(scope)?
.into_iter()
.map(|document| document.payload)
.collect())
}
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()
}
pub fn only(&self, scope: &Scope) -> Result<Option<serde_json::Value>> {
Ok(self
.document(scope, SINGLETON)?
.map(|document| document.payload))
}
pub fn put_only(&self, scope: &Scope, payload: serde_json::Value) -> Result<bool> {
self.put(scope, &[Document::of(SINGLETON, payload)])
}
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)
}
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)
}
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),
)?)
}
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(())
}
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)
}
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)) => {
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)
}
pub const SINGLETON: &str = "-";
pub fn gist_is_whole(gist: &GitHubGist) -> bool {
gist.files.is_empty() || gist.files.values().all(|file| file.content.is_some())
}
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;