michal/tit
Browse tree · Show commit · Download archive
Blob: src/store/mod.rs
use std::ffi::OsString;
use std::fs;
use std::path::{Path, PathBuf};
use std::time::Duration;
use rusqlite::OpenFlags;
use rusqlite::backup::Backup;
use rusqlite::types::ValueRef;
use rusqlite::{Connection, OptionalExtension, TransactionBehavior};
use serde::Serialize;
use thiserror::Error;
mod event;
const BUSY_TIMEOUT: Duration = Duration::from_secs(5);
const BUSY_TIMEOUT_MILLISECONDS: i64 = 5_000;
const MAX_ACTIVE_FEED_TOKENS: i64 = 1;
const SCHEMA_VERSION: i64 = 23;
#[allow(
dead_code,
reason = "the integration test imports this module without the CLI operation"
)]
pub(crate) const DATABASE_FILE: &str = "tit.sqlite3";
#[allow(
dead_code,
reason = "M1A proves migrations before the M2 server calls them"
)]
const MIGRATIONS: [&str; 23] = [
include_str!("migrations/001_initial.sql"),
include_str!("migrations/002_state.sql"),
include_str!("migrations/003_git_intents.sql"),
include_str!("migrations/004_identity.sql"),
include_str!("migrations/005_repository.sql"),
include_str!("migrations/006_repository_events.sql"),
include_str!("migrations/007_account_lifecycle.sql"),
include_str!("migrations/008_web_sessions.sql"),
include_str!("migrations/009_repository_authorization.sql"),
include_str!("migrations/010_audit_history.sql"),
include_str!("migrations/011_domain_events.sql"),
include_str!("migrations/012_issues.sql"),
include_str!("migrations/013_watches.sql"),
include_str!("migrations/014_feed_tokens.sql"),
include_str!("migrations/015_pull_requests.sql"),
include_str!("migrations/016_pull_request_reviews.sql"),
include_str!("migrations/017_pull_request_merges.sql"),
include_str!("migrations/018_streamlined_login.sql"),
include_str!("migrations/019_product_reduction.sql"),
include_str!("migrations/020_repository_profiles.sql"),
include_str!("migrations/021_pull_request_lifecycle.sql"),
include_str!("migrations/022_account_key_management.sql"),
include_str!("migrations/023_default_branch.sql"),
];
#[allow(
dead_code,
reason = "integration tests compile storage without every account operation"
)]
#[derive(Debug, Error)]
pub(crate) enum StoreError {
#[error("SQLite error: {0}")]
Sqlite(#[from] rusqlite::Error),
#[allow(
dead_code,
reason = "M1A proves migrations before the M2 server calls them"
)]
#[error("database schema version {0} is newer than this executable")]
NewerSchema(i64),
#[allow(
dead_code,
reason = "the integration test imports this module without the CLI operation"
)]
#[error("database schema version is {actual}, expected {expected}")]
SchemaVersion { expected: i64, actual: i64 },
#[error("database integrity check failed: {0}")]
Integrity(String),
#[error("SQLite setting {name} is {actual}, expected {expected}")]
Setting {
name: &'static str,
expected: &'static str,
actual: String,
},
#[error("Git operation intent {0} is not in the required state")]
IntentState(String),
#[error("the instance already has an administrator")]
AlreadyInitialized,
#[error("account does not exist or is not active: {0}")]
AccountNotFound(String),
#[error("username is not available: {0}")]
UsernameUnavailable(String),
#[error("signup invitation is invalid, expired, or already used")]
InvalidInvitation,
#[error("recovery credential is invalid")]
InvalidRecovery,
#[error("SSH public key already exists")]
KeyExists,
#[error("active SSH public key does not exist")]
KeyNotFound,
#[error("an account must have at least one active SSH public key")]
LastKey,
#[error("login identity does not exist or is not active")]
LoginIdentity,
#[error("too many login challenges are active")]
LoginNonceLimit,
#[error("login challenge is invalid, expired, or already used")]
InvalidLoginChallenge,
#[error("SSH login approval is invalid, expired, or already used")]
InvalidLoginApproval,
#[error("SSH login approval is waiting for SSH authentication")]
LoginApprovalPending,
#[error("Web session is invalid or expired")]
InvalidSession,
#[error("repository does not exist: {0}/{1}")]
RepositoryNotFound(String, String),
#[error("repository already exists: {0}/{1}")]
RepositoryExists(String, String),
#[error("repository ID already exists")]
RepositoryIdentifierCollision,
#[error("repository is already archived: {0}/{1}")]
RepositoryArchived(String, String),
#[error("repository visibility is not valid")]
InvalidRepositoryVisibility,
#[error("repository default branch is not valid")]
InvalidDefaultBranch,
#[error("collaborator role is not valid")]
InvalidCollaboratorRole,
#[error("repository owner cannot be a collaborator")]
OwnerCollaborator,
#[error("collaborator account does not exist or is not active: {0}")]
CollaboratorNotFound(String),
#[allow(
dead_code,
reason = "some integration tests import storage without public event pages"
)]
#[error("repository event page limit is too large")]
EventLimit,
#[error("stored Git reference event is malformed")]
EventPayload,
#[error("audit event page limit is too large")]
AuditLimit,
#[error("issue does not exist: {0}/{1}#{2}")]
IssueNotFound(String, String, i64),
#[error("issue access is not authorized")]
IssueDenied,
#[error("issue is hidden by repository access policy")]
IssueHidden,
#[error("issue state is already {0}")]
IssueState(String),
#[error("pull request does not exist: {0}/{1}#{2}")]
PullRequestNotFound(String, String, i64),
#[error("pull-request access is not authorized")]
PullRequestDenied,
#[error("pull request is hidden by repository access policy")]
PullRequestHidden,
#[error("pull request is not open")]
PullRequestState,
#[error("pull-request revision does not exist")]
PullRequestRevisionNotFound,
#[error("pull-request review anchor does not match its revision")]
PullRequestReviewAnchor,
#[error("pull-request ref intent {0} is not in the required state")]
PullRequestIntentState(String),
#[error("repository watch access is not authorized")]
WatchDenied,
#[error("feed token is invalid or revoked")]
FeedTokenNotFound,
#[error("an account cannot have more than one active feed token")]
FeedTokenLimit,
#[error("feed token scope is not valid")]
InvalidFeedScope,
}
pub(crate) struct Store {
connection: Connection,
}
impl Store {
#[allow(
dead_code,
reason = "some integration tests compile storage without audited services"
)]
pub(crate) fn record_audit_event(&self, event: &NewAuditEvent<'_>) -> Result<(), StoreError> {
insert_audit_event(&self.connection, event)?;
Ok(())
}
#[allow(
dead_code,
reason = "some integration tests compile storage without the audit CLI"
)]
pub(crate) fn audit_events(&self, limit: usize) -> Result<Vec<AuditEventRecord>, StoreError> {
if limit == 0 || limit > 1_000 {
return Err(StoreError::AuditLimit);
}
let limit = i64::try_from(limit).map_err(|_| StoreError::AuditLimit)?;
let mut statement = self.connection.prepare(
"SELECT id, action, actor, target, outcome, correlation_id, created_at
FROM audit_event ORDER BY id DESC LIMIT ?1",
)?;
statement
.query_map([limit], |row| {
Ok(AuditEventRecord {
id: row.get(0)?,
action: row.get(1)?,
actor: row.get(2)?,
target: row.get(3)?,
outcome: row.get(4)?,
correlation_id: row.get(5)?,
created_at: row.get(6)?,
})
})?
.collect::<Result<Vec<_>, _>>()
.map_err(Into::into)
}
#[allow(
dead_code,
reason = "M1A proves migrations before the M2 server calls them"
)]
pub(crate) fn open(path: &Path) -> Result<Self, StoreError> {
let mut store = Self::open_unmigrated(path)?;
let current = store.schema_version()?;
if current > 0 && current < SCHEMA_VERSION {
store.backup(&migration_backup_path(path, current))?;
}
store.migrate()?;
Ok(store)
}
#[allow(
dead_code,
reason = "M1A proves migrations before the M2 server calls them"
)]
pub(crate) fn open_unmigrated(path: &Path) -> Result<Self, StoreError> {
let connection = Connection::open(path)?;
configure(&connection)?;
Ok(Self { connection })
}
pub(crate) fn open_read_only(path: &Path) -> Result<Self, StoreError> {
let connection = Connection::open_with_flags(
path,
OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX,
)?;
connection.busy_timeout(BUSY_TIMEOUT)?;
connection.pragma_update(None, "foreign_keys", true)?;
Ok(Self { connection })
}
#[allow(
dead_code,
reason = "M1A proves migrations before the M2 server calls them"
)]
pub(crate) fn migrate(&mut self) -> Result<(), StoreError> {
self.migrate_with_hook(|_| {})
}
#[allow(
dead_code,
reason = "M1A proves migrations before the M2 server calls them"
)]
pub(crate) fn migrate_with_hook(
&mut self,
mut after_migration: impl FnMut(i64),
) -> Result<(), StoreError> {
let current = self.schema_version()?;
if current > SCHEMA_VERSION {
return Err(StoreError::NewerSchema(current));
}
if current == SCHEMA_VERSION {
return Ok(());
}
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Exclusive)?;
for version in (current + 1)..=SCHEMA_VERSION {
transaction.execute_batch(MIGRATIONS[(version - 1) as usize])?;
transaction.pragma_update(None, "user_version", version)?;
after_migration(version);
}
transaction.commit()?;
if current < 23 {
self.backfill_default_branches();
}
Ok(())
}
fn backfill_default_branches(&self) {
let Some(database) = self.connection.path() else {
return;
};
let Some(instance) = Path::new(database).parent() else {
return;
};
let Ok(mut statement) = self.connection.prepare("SELECT id FROM repository") else {
return;
};
let Ok(ids) = statement.query_map([], |row| row.get::<_, String>(0)) else {
return;
};
for id in ids.flatten() {
let head = instance
.join("repositories")
.join(format!("{id}.git"))
.join("HEAD");
let Ok(contents) = fs::read(head) else {
continue;
};
let Some(name) = contents
.strip_suffix(b"\n")
.unwrap_or(&contents)
.strip_prefix(b"ref: ")
else {
continue;
};
let candidate = gix::bstr::BString::from(name);
if !name.starts_with(b"refs/heads/")
|| gix::refs::FullName::try_from(candidate).is_err()
{
continue;
}
let Ok(name) = std::str::from_utf8(name) else {
continue;
};
let _ = self.connection.execute(
"UPDATE repository_default_branch SET ref_name = ?2
WHERE repository_id = ?1",
rusqlite::params![id, name],
);
}
}
pub(crate) fn schema_version(&self) -> Result<i64, StoreError> {
Ok(self
.connection
.pragma_query_value(None, "user_version", |row| row.get(0))?)
}
pub(crate) fn integrity_check(&self) -> Result<(), StoreError> {
let result: String =
self.connection
.pragma_query_value(None, "integrity_check", |row| row.get(0))?;
if result != "ok" {
return Err(StoreError::Integrity(result));
}
let mut statement = self.connection.prepare("PRAGMA foreign_key_check")?;
let mut rows = statement.query([])?;
if let Some(row) = rows.next()? {
let table: String = row.get(0)?;
return Err(StoreError::Integrity(format!(
"foreign key violation in table {table}"
)));
}
Ok(())
}
#[allow(dead_code, reason = "M1A proves backup before the M2 server calls it")]
pub(crate) fn backup(&self, path: &Path) -> Result<(), StoreError> {
let mut destination = Connection::open(path)?;
let backup = Backup::new(&self.connection, &mut destination)?;
backup.run_to_completion(128, Duration::from_millis(1), None)?;
Ok(())
}
pub(crate) fn maintain(&mut self, cutoff: i64) -> Result<MaintenanceResult, StoreError> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)?;
let mut deleted = 0_usize;
for statement in [
"DELETE FROM signup_invitation WHERE expires_at < ?1",
"DELETE FROM login_nonce WHERE expires_at < ?1",
"DELETE FROM ssh_login_approval WHERE expires_at < ?1",
"DELETE FROM web_session
WHERE expires_at < ?1 OR (ended_at IS NOT NULL AND ended_at < ?1)",
"DELETE FROM feed_token WHERE revoked_at IS NOT NULL AND revoked_at < ?1",
"DELETE FROM audit_event WHERE created_at < ?1",
] {
deleted = deleted
.checked_add(transaction.execute(statement, [cutoff])?)
.ok_or(StoreError::EventLimit)?;
}
transaction.commit()?;
self.connection
.execute_batch("PRAGMA wal_checkpoint(TRUNCATE); VACUUM;")?;
Ok(MaintenanceResult { deleted })
}
#[cfg(test)]
#[allow(dead_code, reason = "integration storage tests use this test boundary")]
pub(crate) fn checkpoint(&self) -> Result<(), StoreError> {
self.connection
.execute_batch("PRAGMA wal_checkpoint(TRUNCATE)")?;
Ok(())
}
#[cfg(test)]
#[allow(dead_code, reason = "integration storage tests use this test boundary")]
pub(crate) fn vacuum(&self) -> Result<(), StoreError> {
self.connection.execute_batch("VACUUM")?;
Ok(())
}
#[allow(
dead_code,
reason = "M1A tests storage behavior through this narrow test boundary"
)]
pub(crate) fn connection(&self) -> &Connection {
&self.connection
}
#[allow(
dead_code,
reason = "M1A tests storage behavior through this narrow test boundary"
)]
pub(crate) fn connection_mut(&mut self) -> &mut Connection {
&mut self.connection
}
pub(crate) fn begin_git_intent(
&self,
intent: &GitOperationIntent<'_>,
) -> Result<(), StoreError> {
self.connection.execute(
"INSERT INTO git_operation_intent
(id, repository_path, actor, initial_refs, proposed_refs, event_payload,
quarantine_path, state, created_at)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, 'pending', ?8)",
rusqlite::params![
intent.id,
intent.repository_path,
intent.actor,
intent.initial_refs,
intent.proposed_refs,
intent.event_payload,
intent.quarantine_path,
intent.created_at,
],
)?;
Ok(())
}
pub(crate) fn begin_pull_request_merge(
&mut self,
merge: &NewPullRequestMerge<'_>,
intent: &GitOperationIntent<'_>,
) -> Result<(), StoreError> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)?;
let access = repository_issue_access(
&transaction,
merge.owner,
merge.repository,
Some(merge.actor),
)?;
if !access.can_read() {
return Err(StoreError::PullRequestHidden);
}
if !access.can_maintain() {
return Err(StoreError::PullRequestDenied);
}
if managed_repository_id(intent.repository_path) != Some(access.repository.id.as_str()) {
return Err(StoreError::Integrity(
"pull-request merge repository path does not match its repository".to_owned(),
));
}
let current = transaction
.query_row(
"SELECT pull_request.id, pull_request.state, pull_request.base_ref,
pull_request.base_object_id, pull_request.head_object_id,
pull_request_revision.id, pull_request_revision.number
FROM pull_request
JOIN pull_request_revision
ON pull_request_revision.pull_request_id = pull_request.id
WHERE pull_request.repository_id = ?1
AND pull_request.number = ?2
AND pull_request_revision.number =
(SELECT MAX(number) FROM pull_request_revision AS latest
WHERE latest.pull_request_id = pull_request.id)",
rusqlite::params![access.repository.id, merge.number],
|row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
row.get::<_, String>(3)?,
row.get::<_, String>(4)?,
row.get::<_, String>(5)?,
row.get::<_, i64>(6)?,
))
},
)
.optional()?
.ok_or_else(|| {
StoreError::PullRequestNotFound(
merge.owner.to_owned(),
merge.repository.to_owned(),
merge.number,
)
})?;
if current.1 != "open" {
return Err(StoreError::PullRequestState);
}
if current.2 != merge.base_ref
|| current.3 != merge.old_target
|| current.4 != merge.head_target
|| current.6 != merge.revision
{
return Err(StoreError::PullRequestRevisionNotFound);
}
let initial = parse_event_refs(intent.initial_refs)?;
let proposed = parse_event_refs(intent.proposed_refs)?;
if initial
!= [(
merge.old_target.to_owned(),
merge.base_ref.as_bytes().to_vec(),
)]
|| proposed
!= [(
merge.new_target.to_owned(),
merge.base_ref.as_bytes().to_vec(),
)]
{
return Err(StoreError::EventPayload);
}
transaction.execute(
"INSERT INTO git_operation_intent
(id, repository_path, actor, initial_refs, proposed_refs, event_payload,
quarantine_path, state, created_at)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, 'pending', ?8)",
rusqlite::params![
intent.id,
intent.repository_path,
intent.actor,
intent.initial_refs,
intent.proposed_refs,
intent.event_payload,
intent.quarantine_path,
intent.created_at,
],
)?;
transaction.execute(
"INSERT INTO pull_request_merge_intent
(intent_id, repository_id, pull_request_id, revision_id, revision_number,
method, base_ref, old_target, new_target, created_at)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
rusqlite::params![
intent.id,
access.repository.id,
current.0,
current.5,
merge.revision,
merge.method,
merge.base_ref,
merge.old_target,
merge.new_target,
merge.created_at,
],
)?;
transaction.commit()?;
Ok(())
}
pub(crate) fn mark_git_objects_promoted(
&self,
id: &str,
pack_name: Option<&str>,
) -> Result<(), StoreError> {
let changed = self.connection.execute(
"UPDATE git_operation_intent
SET state = 'promoted', pack_name = ?2
WHERE id = ?1 AND state = 'pending'",
rusqlite::params![id, pack_name],
)?;
require_one_intent(id, changed)
}
pub(crate) fn complete_git_intent(&mut self, id: &str) -> Result<(), StoreError> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)?;
let (payload, repository_path, actor, initial_refs, proposed_refs, created_at): (
Vec<u8>,
String,
String,
Vec<u8>,
Vec<u8>,
i64,
) = transaction.query_row(
"SELECT event_payload, repository_path, actor, initial_refs, proposed_refs, created_at
FROM git_operation_intent
WHERE id = ?1 AND state = 'promoted'",
[id],
|row| {
Ok((
row.get(0)?,
row.get(1)?,
row.get(2)?,
row.get(3)?,
row.get(4)?,
row.get(5)?,
))
},
)?;
let changed = transaction.execute(
"UPDATE git_operation_intent SET state = 'completed'
WHERE id = ?1 AND state = 'promoted'",
[id],
)?;
require_one_intent(id, changed)?;
transaction.execute(
"INSERT INTO git_operation_event (intent_id, payload) VALUES (?1, ?2)",
rusqlite::params![id, payload],
)?;
insert_push_events(
&transaction,
id,
&repository_path,
&actor,
&initial_refs,
&proposed_refs,
created_at,
)?;
complete_pull_request_merge(&transaction, id, &actor, created_at)?;
transaction.commit()?;
Ok(())
}
pub(crate) fn abandon_git_intent(&mut self, id: &str) -> Result<(), StoreError> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)?;
transaction.execute(
"DELETE FROM pull_request_merge_intent WHERE intent_id = ?1",
[id],
)?;
let changed = transaction.execute(
"UPDATE git_operation_intent SET state = 'abandoned'
WHERE id = ?1 AND state IN ('pending', 'promoted')",
[id],
)?;
require_one_intent(id, changed)?;
transaction.commit()?;
Ok(())
}
#[allow(
dead_code,
reason = "some integration tests compile storage without receive-pack"
)]
pub(crate) fn git_intent_completed(&self, id: &str) -> Result<bool, StoreError> {
Ok(self
.connection
.query_row(
"SELECT state = 'completed' FROM git_operation_intent WHERE id = ?1",
[id],
|row| row.get(0),
)
.optional()?
.unwrap_or(false))
}
pub(crate) fn incomplete_git_intents(&self) -> Result<Vec<GitIntentRecord>, StoreError> {
let mut statement = self.connection.prepare(
"SELECT id, repository_path, initial_refs, proposed_refs, quarantine_path,
state, pack_name
FROM git_operation_intent
WHERE state IN ('pending', 'promoted')
ORDER BY created_at, id",
)?;
let records = statement
.query_map([], |row| {
Ok(GitIntentRecord {
id: row.get(0)?,
repository_path: row.get(1)?,
initial_refs: row.get(2)?,
proposed_refs: row.get(3)?,
quarantine_path: row.get(4)?,
state: row.get(5)?,
pack_name: row.get(6)?,
})
})?
.collect::<Result<Vec<_>, _>>()?;
Ok(records)
}
pub(crate) fn create_initial_administrator(
&mut self,
administrator: &InitialAdministrator<'_>,
) -> Result<(), StoreError> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)?;
let accounts: i64 =
transaction.query_row("SELECT count(*) FROM account", [], |row| row.get(0))?;
if accounts != 0 {
return Err(StoreError::AlreadyInitialized);
}
transaction.execute(
"INSERT INTO account
(username, is_administrator, state, created_at)
VALUES (?1, 1, 'active', ?2)",
rusqlite::params![administrator.username, administrator.created_at],
)?;
let account_id = transaction.last_insert_rowid();
transaction.execute(
"INSERT INTO ssh_public_key
(account_id, canonical_key, fingerprint, created_at)
VALUES (?1, ?2, ?3, ?4)",
rusqlite::params![
account_id,
administrator.canonical_key,
administrator.fingerprint,
administrator.created_at,
],
)?;
transaction.execute(
"INSERT INTO recovery_credential
(account_id, credential_hash, created_at)
VALUES (?1, ?2, ?3)",
rusqlite::params![
account_id,
administrator.recovery_hash,
administrator.created_at,
],
)?;
transaction.commit()?;
Ok(())
}
#[allow(
dead_code,
reason = "some integration tests compile storage without accounts"
)]
pub(crate) fn create_signup_invitation(
&self,
code_hash: &[u8; 32],
created_at: i64,
expires_at: i64,
) -> Result<(), StoreError> {
self.connection.execute(
"INSERT INTO signup_invitation (code_hash, created_at, expires_at, consumed_at)
VALUES (?1, ?2, ?3, NULL)",
rusqlite::params![code_hash, created_at, expires_at],
)?;
Ok(())
}
#[allow(
dead_code,
reason = "some integration tests compile storage without accounts"
)]
pub(crate) fn create_account_with_invitation(
&mut self,
account: &InvitedAccount<'_>,
) -> Result<(), StoreError> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)?;
let consumed = transaction.execute(
"UPDATE signup_invitation SET consumed_at = ?2
WHERE code_hash = ?1 AND consumed_at IS NULL AND expires_at >= ?2",
rusqlite::params![account.invitation_hash, account.created_at],
)?;
if consumed != 1 {
return Err(StoreError::InvalidInvitation);
}
let inserted = transaction.execute(
"INSERT INTO account (username, is_administrator, state, created_at)
VALUES (?1, 0, 'active', ?2)",
rusqlite::params![account.username, account.created_at],
);
let account_id = match inserted {
Ok(1) => transaction.last_insert_rowid(),
Err(error) if is_unique_constraint(&error) => {
return Err(StoreError::UsernameUnavailable(account.username.to_owned()));
}
Err(error) => return Err(error.into()),
Ok(_) => unreachable!("an INSERT changes one row"),
};
insert_ssh_key(&transaction, account_id, &account.key, account.created_at)?;
transaction.execute(
"INSERT INTO recovery_credential (account_id, credential_hash, created_at)
VALUES (?1, ?2, ?3)",
rusqlite::params![account_id, account.recovery_hash, account.created_at],
)?;
insert_audit_event(
&transaction,
&NewAuditEvent {
action: "account.signup",
actor: account.username,
target: account.username,
outcome: "success",
correlation_id: account.correlation_id,
created_at: account.created_at,
},
)?;
transaction.commit()?;
Ok(())
}
#[allow(
dead_code,
reason = "some integration tests compile storage without accounts"
)]
pub(crate) fn recover_account(
&mut self,
recovery: &AccountRecovery<'_>,
) -> Result<(), StoreError> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)?;
let account_id = transaction
.query_row(
"SELECT account.id FROM account
JOIN recovery_credential ON recovery_credential.account_id = account.id
WHERE account.username = ?1 AND account.state = 'active'
AND recovery_credential.credential_hash = ?2",
rusqlite::params![recovery.username, recovery.old_recovery_hash],
|row| row.get::<_, i64>(0),
)
.optional()?
.ok_or(StoreError::InvalidRecovery)?;
transaction.execute(
"UPDATE ssh_public_key SET revoked_at = ?2
WHERE account_id = ?1 AND revoked_at IS NULL",
rusqlite::params![account_id, recovery.created_at],
)?;
let existing = transaction
.query_row(
"SELECT account_id FROM ssh_public_key WHERE fingerprint = ?1",
[recovery.key.fingerprint],
|row| row.get::<_, i64>(0),
)
.optional()?;
match existing {
Some(owner) if owner == account_id => {
transaction.execute(
"UPDATE ssh_public_key
SET canonical_key = ?2, label = ?3, revoked_at = NULL
WHERE account_id = ?1 AND fingerprint = ?4",
rusqlite::params![
account_id,
recovery.key.canonical_key,
recovery.key.label,
recovery.key.fingerprint,
],
)?;
}
Some(_) => return Err(StoreError::KeyExists),
None => insert_ssh_key(&transaction, account_id, &recovery.key, recovery.created_at)?,
}
transaction.execute(
"UPDATE recovery_credential
SET credential_hash = ?2, created_at = ?3 WHERE account_id = ?1",
rusqlite::params![account_id, recovery.new_recovery_hash, recovery.created_at],
)?;
end_sessions(&transaction, account_id, recovery.created_at)?;
revoke_feed_tokens(&transaction, account_id, recovery.created_at)?;
insert_audit_event(
&transaction,
&NewAuditEvent {
action: "account.recover",
actor: recovery.username,
target: recovery.username,
outcome: "success",
correlation_id: recovery.correlation_id,
created_at: recovery.created_at,
},
)?;
transaction.commit()?;
Ok(())
}
#[allow(
dead_code,
reason = "some integration tests compile storage without accounts"
)]
pub(crate) fn add_account_key(
&mut self,
username: &str,
key: &NewSshKey<'_>,
created_at: i64,
actor: &str,
correlation_id: &str,
) -> Result<(), StoreError> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)?;
let account_id = active_account_id(&transaction, username)?;
insert_ssh_key(&transaction, account_id, key, created_at)?;
end_sessions(&transaction, account_id, created_at)?;
let target = format!("{username}:{}", key.fingerprint);
insert_audit_event(
&transaction,
&NewAuditEvent {
action: "key.add",
actor,
target: &target,
outcome: "success",
correlation_id,
created_at,
},
)?;
transaction.commit()?;
Ok(())
}
#[allow(
dead_code,
reason = "some integration tests compile storage without accounts"
)]
pub(crate) fn revoke_account_key(
&mut self,
username: &str,
fingerprint: &str,
revoked_at: i64,
actor: &str,
correlation_id: &str,
) -> Result<(), StoreError> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)?;
let account_id = active_account_id(&transaction, username)?;
let active: i64 = transaction.query_row(
"SELECT count(*) FROM ssh_public_key WHERE account_id = ?1 AND revoked_at IS NULL",
[account_id],
|row| row.get(0),
)?;
if active <= 1 {
return Err(StoreError::LastKey);
}
let changed = transaction.execute(
"UPDATE ssh_public_key SET revoked_at = ?3
WHERE account_id = ?1 AND fingerprint = ?2 AND revoked_at IS NULL",
rusqlite::params![account_id, fingerprint, revoked_at],
)?;
if changed != 1 {
return Err(StoreError::KeyNotFound);
}
end_sessions(&transaction, account_id, revoked_at)?;
let target = format!("{username}:{fingerprint}");
insert_audit_event(
&transaction,
&NewAuditEvent {
action: "key.revoke",
actor,
target: &target,
outcome: "success",
correlation_id,
created_at: revoked_at,
},
)?;
transaction.commit()?;
Ok(())
}
pub(crate) fn complete_account_key_add(
&mut self,
authorization: &AccountKeyAuthorization<'_>,
key: &NewSshKey<'_>,
) -> Result<(), StoreError> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)?;
let (account_id, _) = consume_account_key_authorization(&transaction, authorization)?;
let duplicate_label: bool = transaction.query_row(
"SELECT EXISTS(
SELECT 1 FROM ssh_public_key
WHERE account_id = ?1 AND label = ?2 AND revoked_at IS NULL
)",
rusqlite::params![account_id, key.label],
|row| row.get(0),
)?;
if duplicate_label {
return Err(StoreError::KeyExists);
}
insert_ssh_key(&transaction, account_id, key, authorization.changed_at)?;
let target = format!("{}:{}", authorization.username, key.fingerprint);
insert_audit_event(
&transaction,
&NewAuditEvent {
action: "key.add",
actor: authorization.username,
target: &target,
outcome: "success",
correlation_id: authorization.correlation_id,
created_at: authorization.changed_at,
},
)?;
transaction.commit()?;
Ok(())
}
pub(crate) fn complete_account_key_revoke(
&mut self,
authorization: &AccountKeyAuthorization<'_>,
fingerprint: &str,
) -> Result<(), StoreError> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)?;
let (account_id, session_key_id) =
consume_account_key_authorization(&transaction, authorization)?;
let active: i64 = transaction.query_row(
"SELECT count(*) FROM ssh_public_key
WHERE account_id = ?1 AND revoked_at IS NULL",
[account_id],
|row| row.get(0),
)?;
if active <= 1 {
return Err(StoreError::LastKey);
}
let key_id = transaction
.query_row(
"SELECT id FROM ssh_public_key
WHERE account_id = ?1 AND fingerprint = ?2 AND revoked_at IS NULL",
rusqlite::params![account_id, fingerprint],
|row| row.get::<_, i64>(0),
)
.optional()?
.ok_or(StoreError::KeyNotFound)?;
let changed = transaction.execute(
"UPDATE ssh_public_key SET revoked_at = ?3
WHERE id = ?1 AND account_id = ?2 AND revoked_at IS NULL",
rusqlite::params![key_id, account_id, authorization.changed_at],
)?;
if changed != 1 {
return Err(StoreError::KeyNotFound);
}
if session_key_id.is_none() || session_key_id == Some(key_id) {
end_sessions(&transaction, account_id, authorization.changed_at)?;
}
let target = format!("{}:{fingerprint}", authorization.username);
insert_audit_event(
&transaction,
&NewAuditEvent {
action: "key.revoke",
actor: authorization.username,
target: &target,
outcome: "success",
correlation_id: authorization.correlation_id,
created_at: authorization.changed_at,
},
)?;
transaction.commit()?;
Ok(())
}
#[allow(
dead_code,
reason = "some integration tests compile storage without accounts"
)]
pub(crate) fn suspend_account(
&mut self,
username: &str,
suspended: bool,
changed_at: i64,
actor: &str,
correlation_id: &str,
) -> Result<(), StoreError> {
let state = if suspended { "suspended" } else { "active" };
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)?;
let changed = transaction.execute(
"UPDATE account SET state = ?2 WHERE username = ?1",
rusqlite::params![username, state],
)?;
if changed != 1 {
return Err(StoreError::AccountNotFound(username.to_owned()));
}
let account_id: i64 = transaction.query_row(
"SELECT id FROM account WHERE username = ?1",
[username],
|row| row.get(0),
)?;
end_sessions(&transaction, account_id, changed_at)?;
if suspended {
revoke_feed_tokens(&transaction, account_id, changed_at)?;
}
insert_audit_event(
&transaction,
&NewAuditEvent {
action: if suspended {
"account.suspend"
} else {
"account.resume"
},
actor,
target: username,
outcome: "success",
correlation_id,
created_at: changed_at,
},
)?;
transaction.commit()?;
Ok(())
}
#[allow(
dead_code,
reason = "some integration tests compile storage without Web login"
)]
pub(crate) fn create_login_nonce(
&mut self,
nonce: &NewLoginNonce<'_>,
) -> Result<(), StoreError> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)?;
transaction.execute(
"DELETE FROM login_nonce WHERE expires_at < ?1 OR consumed_at IS NOT NULL",
[nonce.created_at],
)?;
let active: i64 =
transaction.query_row("SELECT count(*) FROM login_nonce", [], |row| row.get(0))?;
if active >= 1_024 {
return Err(StoreError::LoginNonceLimit);
}
let account_id = transaction
.query_row(
"SELECT id FROM account
WHERE username = ?1 AND state = 'active'",
[nonce.username],
|row| row.get::<_, i64>(0),
)
.optional()?
.ok_or(StoreError::LoginIdentity)?;
transaction.execute(
"INSERT INTO login_nonce
(nonce_hash, csrf_hash, account_id, ssh_public_key_id,
created_at, expires_at, consumed_at)
VALUES (?1, ?2, ?3, NULL, ?4, ?5, NULL)",
rusqlite::params![
nonce.nonce_hash,
nonce.csrf_hash,
account_id,
nonce.created_at,
nonce.expires_at,
],
)?;
transaction.commit()?;
Ok(())
}
#[allow(
dead_code,
reason = "some integration tests compile storage without Web login"
)]
pub(crate) fn consume_login_nonce(
&mut self,
login: &NewWebSession<'_>,
) -> Result<(), StoreError> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)?;
let (account_id, key_id) = transaction
.query_row(
"SELECT login_nonce.account_id, ssh_public_key.id
FROM login_nonce
JOIN account ON account.id = login_nonce.account_id
JOIN ssh_public_key ON ssh_public_key.account_id = login_nonce.account_id
WHERE login_nonce.nonce_hash = ?1
AND login_nonce.consumed_at IS NULL AND login_nonce.expires_at >= ?2
AND account.username = ?3 AND account.state = 'active'
AND ssh_public_key.fingerprint = ?4 AND ssh_public_key.revoked_at IS NULL
AND (login_nonce.ssh_public_key_id IS NULL
OR login_nonce.ssh_public_key_id = ssh_public_key.id)
AND login_nonce.csrf_hash = ?5",
rusqlite::params![
login.nonce_hash,
login.created_at,
login.username,
login.fingerprint,
login.login_csrf_hash,
],
|row| Ok((row.get::<_, i64>(0)?, row.get::<_, i64>(1)?)),
)
.optional()?
.ok_or(StoreError::InvalidLoginChallenge)?;
let changed = transaction.execute(
"UPDATE login_nonce SET consumed_at = ?2
WHERE nonce_hash = ?1 AND consumed_at IS NULL",
rusqlite::params![login.nonce_hash, login.created_at],
)?;
if changed != 1 {
return Err(StoreError::InvalidLoginChallenge);
}
transaction.execute(
"INSERT INTO web_session
(session_hash, csrf_hash, account_id, created_at, expires_at, ended_at,
ssh_public_key_id)
VALUES (?1, ?2, ?3, ?4, ?5, NULL, ?6)",
rusqlite::params![
login.session_hash,
login.csrf_hash,
account_id,
login.created_at,
login.expires_at,
key_id,
],
)?;
insert_audit_event(
&transaction,
&NewAuditEvent {
action: "login",
actor: login.username,
target: login.username,
outcome: "success",
correlation_id: login.correlation_id,
created_at: login.created_at,
},
)?;
transaction.commit()?;
Ok(())
}
pub(crate) fn create_login_approval(
&mut self,
approval: &NewLoginApproval<'_>,
) -> Result<(), StoreError> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)?;
transaction.execute(
"DELETE FROM ssh_login_approval
WHERE expires_at < ?1 OR consumed_at IS NOT NULL",
[approval.created_at],
)?;
let active: i64 =
transaction.query_row("SELECT count(*) FROM ssh_login_approval", [], |row| {
row.get(0)
})?;
if active >= 1_024 {
return Err(StoreError::LoginNonceLimit);
}
let expected_account_id = approval
.expected_username
.map(|username| active_account_id(&transaction, username))
.transpose()?;
transaction.execute(
"INSERT INTO ssh_login_approval
(secret_hash, csrf_hash, account_id, ssh_public_key_id,
created_at, expires_at, approved_at, consumed_at, purpose, expected_account_id)
VALUES (?1, ?2, NULL, NULL, ?3, ?4, NULL, NULL, ?5, ?6)",
rusqlite::params![
approval.secret_hash,
approval.csrf_hash,
approval.created_at,
approval.expires_at,
approval.purpose,
expected_account_id,
],
)?;
transaction.commit()?;
Ok(())
}
pub(crate) fn approve_login(&mut self, approval: &ApproveLogin<'_>) -> Result<(), StoreError> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)?;
let (account_id, key_id) = transaction
.query_row(
"SELECT account.id, ssh_public_key.id
FROM account
JOIN ssh_public_key ON ssh_public_key.account_id = account.id
WHERE account.username = ?1 AND account.state = 'active'
AND ssh_public_key.fingerprint = ?2
AND ssh_public_key.revoked_at IS NULL",
rusqlite::params![approval.username, approval.fingerprint],
|row| Ok((row.get::<_, i64>(0)?, row.get::<_, i64>(1)?)),
)
.optional()?
.ok_or(StoreError::LoginIdentity)?;
let changed = transaction.execute(
"UPDATE ssh_login_approval
SET account_id = ?2, ssh_public_key_id = ?3, approved_at = ?4
WHERE secret_hash = ?1 AND account_id IS NULL
AND approved_at IS NULL AND consumed_at IS NULL AND expires_at >= ?4
AND (expected_account_id IS NULL OR expected_account_id = ?2)",
rusqlite::params![
approval.secret_hash,
account_id,
key_id,
approval.approved_at,
],
)?;
if changed != 1 {
return Err(StoreError::InvalidLoginApproval);
}
transaction.commit()?;
Ok(())
}
pub(crate) fn consume_login_approval(
&mut self,
login: &NewApprovedWebSession<'_>,
) -> Result<String, StoreError> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)?;
let approval = transaction
.query_row(
"SELECT account.id, account.username, ssh_login_approval.approved_at,
ssh_login_approval.ssh_public_key_id
FROM ssh_login_approval
LEFT JOIN account ON account.id = ssh_login_approval.account_id
LEFT JOIN ssh_public_key
ON ssh_public_key.id = ssh_login_approval.ssh_public_key_id
WHERE ssh_login_approval.secret_hash = ?1
AND ssh_login_approval.csrf_hash = ?2
AND ssh_login_approval.purpose = 'login'
AND ssh_login_approval.expected_account_id IS NULL
AND ssh_login_approval.consumed_at IS NULL
AND ssh_login_approval.expires_at >= ?3
AND (ssh_login_approval.account_id IS NULL
OR (account.state = 'active' AND ssh_public_key.revoked_at IS NULL))",
rusqlite::params![login.secret_hash, login.login_csrf_hash, login.created_at,],
|row| {
Ok((
row.get::<_, Option<i64>>(0)?,
row.get::<_, Option<String>>(1)?,
row.get::<_, Option<i64>>(2)?,
row.get::<_, Option<i64>>(3)?,
))
},
)
.optional()?
.ok_or(StoreError::InvalidLoginApproval)?;
let (Some(account_id), Some(username), Some(_), Some(key_id)) = approval else {
return Err(StoreError::LoginApprovalPending);
};
let changed = transaction.execute(
"UPDATE ssh_login_approval SET consumed_at = ?2
WHERE secret_hash = ?1 AND consumed_at IS NULL",
rusqlite::params![login.secret_hash, login.created_at],
)?;
if changed != 1 {
return Err(StoreError::InvalidLoginApproval);
}
transaction.execute(
"INSERT INTO web_session
(session_hash, csrf_hash, account_id, created_at, expires_at, ended_at,
ssh_public_key_id)
VALUES (?1, ?2, ?3, ?4, ?5, NULL, ?6)",
rusqlite::params![
login.session_hash,
login.csrf_hash,
account_id,
login.created_at,
login.expires_at,
key_id,
],
)?;
insert_audit_event(
&transaction,
&NewAuditEvent {
action: "login",
actor: &username,
target: &username,
outcome: "success",
correlation_id: login.correlation_id,
created_at: login.created_at,
},
)?;
transaction.commit()?;
Ok(username)
}
#[allow(
dead_code,
reason = "some integration tests compile storage without Web login"
)]
pub(crate) fn web_session(
&self,
session_hash: &[u8; 32],
csrf_hash: Option<&[u8; 32]>,
now: i64,
) -> Result<WebSessionRecord, StoreError> {
self.connection
.query_row(
"SELECT account.username, account.is_administrator, web_session.expires_at,
web_session.ssh_public_key_id
FROM web_session
JOIN account ON account.id = web_session.account_id
WHERE web_session.session_hash = ?1 AND web_session.ended_at IS NULL
AND web_session.expires_at >= ?3 AND account.state = 'active'
AND (?2 IS NULL OR web_session.csrf_hash = ?2)",
rusqlite::params![session_hash, csrf_hash, now],
|row| {
Ok(WebSessionRecord {
username: row.get(0)?,
is_administrator: row.get(1)?,
expires_at: row.get(2)?,
ssh_public_key_id: row.get(3)?,
})
},
)
.optional()?
.ok_or(StoreError::InvalidSession)
}
#[allow(
dead_code,
reason = "some integration tests compile storage without Web login"
)]
pub(crate) fn end_account_sessions(
&mut self,
username: &str,
ended_at: i64,
) -> Result<(), StoreError> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)?;
let account_id: i64 = transaction
.query_row(
"SELECT id FROM account WHERE username = ?1",
[username],
|row| row.get(0),
)
.optional()?
.ok_or_else(|| StoreError::AccountNotFound(username.to_owned()))?;
end_sessions(&transaction, account_id, ended_at)?;
transaction.commit()?;
Ok(())
}
pub(crate) fn create_repository(
&mut self,
repository: &NewRepository<'_>,
) -> Result<(), StoreError> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)?;
let owner_id = active_account_id(&transaction, repository.owner)?;
let result = transaction.execute(
"INSERT INTO repository
(id, owner_account_id, slug, visibility, state, object_format, created_at, archived_at)
VALUES (?1, ?2, ?3, 'public', 'active', ?4, ?5, NULL)",
rusqlite::params![
repository.id,
owner_id,
repository.slug,
repository.object_format,
repository.created_at,
],
);
match result {
Ok(1) => {
transaction.execute(
"INSERT INTO repository_default_branch (repository_id, ref_name)
VALUES (?1, ?2)",
rusqlite::params![repository.id, repository.default_branch],
)?;
let event = event::repository(
repository.origin.event_kind(),
repository.owner,
repository.slug,
repository.object_format,
);
insert_domain_event(
&transaction,
&NewDomainEvent {
repository_id: repository.id,
source_intent_id: None,
source_ordinal: None,
issue_id: None,
pull_request_id: None,
event: &event,
actor: repository.owner,
ref_name: None,
old_target: None,
new_target: None,
created_at: repository.created_at,
},
)?;
for reference in repository.initial_references {
let kind = if reference.name.starts_with(b"refs/tags/") {
event::EventKind::TagCreated
} else if reference.name.starts_with(b"refs/heads/") {
event::EventKind::RefCreated
} else {
return Err(StoreError::EventPayload);
};
let event =
event::reference(kind, &reference.name, None, Some(&reference.target));
insert_domain_event(
&transaction,
&NewDomainEvent {
repository_id: repository.id,
source_intent_id: None,
source_ordinal: None,
issue_id: None,
pull_request_id: None,
event: &event,
actor: repository.owner,
ref_name: Some(&reference.name),
old_target: None,
new_target: Some(&reference.target),
created_at: repository.created_at,
},
)?;
}
let target = format!("{}/{}", repository.owner, repository.slug);
insert_audit_event(
&transaction,
&NewAuditEvent {
action: repository.origin.audit_action(),
actor: repository.actor,
target: &target,
outcome: "success",
correlation_id: repository.correlation_id,
created_at: repository.created_at,
},
)?;
transaction.commit()?;
}
Ok(_) => unreachable!("an INSERT changes one row"),
Err(error) if is_unique_constraint(&error) => {
let duplicate_id: bool = transaction.query_row(
"SELECT EXISTS(SELECT 1 FROM repository WHERE id = ?1)",
[repository.id],
|row| row.get(0),
)?;
return if duplicate_id {
Err(StoreError::RepositoryIdentifierCollision)
} else {
Err(StoreError::RepositoryExists(
repository.owner.to_owned(),
repository.slug.to_owned(),
))
};
}
Err(error) => return Err(error.into()),
}
Ok(())
}
pub(crate) fn rename_repository(
&mut self,
owner: &str,
old_slug: &str,
new_slug: &str,
changed_at: i64,
actor: &str,
correlation_id: &str,
) -> Result<(), StoreError> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)?;
let owner_id = active_account_id(&transaction, owner)?;
let result = transaction.execute(
"UPDATE repository SET slug = ?3
WHERE owner_account_id = ?1 AND slug = ?2 AND state = 'active'",
rusqlite::params![owner_id, old_slug, new_slug],
);
match result {
Ok(1) => {
let target = format!("{owner}/{old_slug}->{new_slug}");
insert_audit_event(
&transaction,
&NewAuditEvent {
action: "repository.rename",
actor,
target: &target,
outcome: "success",
correlation_id,
created_at: changed_at,
},
)?;
transaction.commit()?;
}
Ok(0) => {
return Err(repository_state_error(
&transaction,
owner_id,
owner,
old_slug,
)?);
}
Ok(_) => unreachable!("an owner and slug identify one repository"),
Err(error) if is_unique_constraint(&error) => {
return Err(StoreError::RepositoryExists(
owner.to_owned(),
new_slug.to_owned(),
));
}
Err(error) => return Err(error.into()),
}
Ok(())
}
pub(crate) fn archive_repository(
&mut self,
owner: &str,
slug: &str,
archived_at: i64,
actor: &str,
correlation_id: &str,
) -> Result<(), StoreError> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)?;
let owner_id = active_account_id(&transaction, owner)?;
let changed = transaction.execute(
"UPDATE repository SET state = 'archived', archived_at = ?3
WHERE owner_account_id = ?1 AND slug = ?2 AND state = 'active'",
rusqlite::params![owner_id, slug, archived_at],
)?;
if changed == 0 {
return Err(repository_state_error(&transaction, owner_id, owner, slug)?);
}
let target = format!("{owner}/{slug}");
insert_audit_event(
&transaction,
&NewAuditEvent {
action: "repository.archive",
actor,
target: &target,
outcome: "success",
correlation_id,
created_at: archived_at,
},
)?;
transaction.commit()?;
Ok(())
}
#[allow(
dead_code,
reason = "some integration tests compile storage without admin commands"
)]
pub(crate) fn set_repository_visibility(
&mut self,
owner: &str,
slug: &str,
visibility: &str,
changed_at: i64,
actor: &str,
correlation_id: &str,
) -> Result<(), StoreError> {
if !matches!(visibility, "public" | "private") {
return Err(StoreError::InvalidRepositoryVisibility);
}
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)?;
let owner_id = active_account_id(&transaction, owner)?;
let changed = transaction.execute(
"UPDATE repository SET visibility = ?3
WHERE owner_account_id = ?1 AND slug = ?2 AND state = 'active'",
rusqlite::params![owner_id, slug, visibility],
)?;
if changed == 0 {
return Err(repository_state_error(&transaction, owner_id, owner, slug)?);
}
let target = format!("{owner}/{slug}:{visibility}");
insert_audit_event(
&transaction,
&NewAuditEvent {
action: "repository.visibility",
actor,
target: &target,
outcome: "success",
correlation_id,
created_at: changed_at,
},
)?;
transaction.commit()?;
Ok(())
}
#[allow(
dead_code,
reason = "some integration tests compile storage without admin commands"
)]
pub(crate) fn set_repository_collaborator(
&mut self,
owner: &str,
slug: &str,
username: &str,
role: &str,
audit: &AuditContext<'_>,
) -> Result<(), StoreError> {
if !matches!(role, "maintainer" | "writer" | "reader") {
return Err(StoreError::InvalidCollaboratorRole);
}
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)?;
let owner_id = active_account_id(&transaction, owner)?;
let repository_id: Option<String> = transaction
.query_row(
"SELECT id FROM repository
WHERE owner_account_id = ?1 AND slug = ?2 AND state = 'active'",
rusqlite::params![owner_id, slug],
|row| row.get(0),
)
.optional()?;
let Some(repository_id) = repository_id else {
return Err(repository_state_error(&transaction, owner_id, owner, slug)?);
};
let collaborator_id = match active_account_id(&transaction, username) {
Ok(account_id) => account_id,
Err(StoreError::AccountNotFound(_)) => {
return Err(StoreError::CollaboratorNotFound(username.to_owned()));
}
Err(error) => return Err(error),
};
if collaborator_id == owner_id {
return Err(StoreError::OwnerCollaborator);
}
transaction.execute(
"INSERT INTO repository_collaborator
(repository_id, account_id, role, created_at)
VALUES (?1, ?2, ?3, ?4)
ON CONFLICT (repository_id, account_id)
DO UPDATE SET role = excluded.role",
rusqlite::params![repository_id, collaborator_id, role, audit.created_at],
)?;
let target = format!("{owner}/{slug}:{username}:{role}");
insert_audit_event(
&transaction,
&NewAuditEvent {
action: "collaborator.set",
actor: audit.actor,
target: &target,
outcome: "success",
correlation_id: audit.correlation_id,
created_at: audit.created_at,
},
)?;
transaction.commit()?;
Ok(())
}
#[allow(
dead_code,
reason = "some integration tests compile storage without admin commands"
)]
pub(crate) fn remove_repository_collaborator(
&mut self,
owner: &str,
slug: &str,
username: &str,
changed_at: i64,
actor: &str,
correlation_id: &str,
) -> Result<(), StoreError> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)?;
let owner_id = active_account_id(&transaction, owner)?;
let repository_id: Option<String> = transaction
.query_row(
"SELECT id FROM repository
WHERE owner_account_id = ?1 AND slug = ?2 AND state = 'active'",
rusqlite::params![owner_id, slug],
|row| row.get(0),
)
.optional()?;
let Some(repository_id) = repository_id else {
return Err(repository_state_error(&transaction, owner_id, owner, slug)?);
};
let collaborator_id: Option<i64> = transaction
.query_row(
"SELECT id FROM account WHERE username = ?1",
[username],
|row| row.get(0),
)
.optional()?;
let Some(collaborator_id) = collaborator_id else {
return Err(StoreError::CollaboratorNotFound(username.to_owned()));
};
if collaborator_id == owner_id {
return Err(StoreError::OwnerCollaborator);
}
let changed = transaction.execute(
"DELETE FROM repository_collaborator
WHERE repository_id = ?1 AND account_id = ?2",
rusqlite::params![repository_id, collaborator_id],
)?;
if changed == 0 {
return Err(StoreError::CollaboratorNotFound(username.to_owned()));
}
let target = format!("{owner}/{slug}:{username}");
insert_audit_event(
&transaction,
&NewAuditEvent {
action: "collaborator.remove",
actor,
target: &target,
outcome: "success",
correlation_id,
created_at: changed_at,
},
)?;
transaction.commit()?;
Ok(())
}
pub(crate) fn repository(
&self,
owner: &str,
slug: &str,
) -> Result<RepositoryRecord, StoreError> {
let result = self.connection.query_row(
"SELECT repository.id, account.username, repository.slug,
repository.visibility, repository.state, repository.object_format,
repository.created_at, repository.archived_at
FROM repository
JOIN account ON account.id = repository.owner_account_id
WHERE account.username = ?1 AND repository.slug = ?2",
rusqlite::params![owner, slug],
repository_from_row,
);
match result {
Ok(repository) => Ok(repository),
Err(rusqlite::Error::QueryReturnedNoRows) => Err(StoreError::RepositoryNotFound(
owner.to_owned(),
slug.to_owned(),
)),
Err(error) => Err(error.into()),
}
}
pub(crate) fn repository_settings(
&self,
owner: &str,
slug: &str,
actor: &str,
) -> Result<RepositorySettings, StoreError> {
let access = repository_issue_access(&self.connection, owner, slug, Some(actor))?;
if !access.can_maintain() {
return Err(StoreError::PullRequestDenied);
}
let description = self
.connection
.query_row(
"SELECT description FROM repository_profile WHERE repository_id = ?1",
[&access.repository.id],
|row| row.get(0),
)
.optional()?
.unwrap_or_default();
let mut statement = self.connection.prepare(
"SELECT account.username, repository_collaborator.role
FROM repository_collaborator
JOIN account ON account.id = repository_collaborator.account_id
WHERE repository_collaborator.repository_id = ?1
ORDER BY account.username",
)?;
let collaborators = statement
.query_map([&access.repository.id], |row| {
Ok(RepositoryCollaboratorRecord {
username: row.get(0)?,
role: row.get(1)?,
})
})?
.collect::<Result<Vec<_>, _>>()?;
Ok(RepositorySettings {
repository: access.repository,
description,
collaborators,
default_branch: self.repository_default_branch(owner, slug)?,
branches: Vec::new(),
})
}
pub(crate) fn repository_default_branch(
&self,
owner: &str,
slug: &str,
) -> Result<String, StoreError> {
self.connection
.query_row(
"SELECT repository_default_branch.ref_name
FROM repository_default_branch
JOIN repository ON repository.id = repository_default_branch.repository_id
JOIN account ON account.id = repository.owner_account_id
WHERE account.username = ?1 AND repository.slug = ?2",
rusqlite::params![owner, slug],
|row| row.get(0),
)
.optional()?
.ok_or_else(|| StoreError::RepositoryNotFound(owner.to_owned(), slug.to_owned()))
}
pub(crate) fn update_repository_default_branch(
&mut self,
owner: &str,
slug: &str,
actor: &str,
default_branch: &str,
changed_at: i64,
correlation_id: &str,
) -> Result<(), StoreError> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)?;
let access = repository_issue_access(&transaction, owner, slug, Some(actor))?;
if !access.can_maintain() {
return Err(StoreError::PullRequestDenied);
}
let changed = transaction.execute(
"UPDATE repository_default_branch SET ref_name = ?2
WHERE repository_id = ?1",
rusqlite::params![access.repository.id, default_branch],
)?;
if changed != 1 {
return Err(StoreError::InvalidDefaultBranch);
}
let target = format!("{owner}/{slug}:{default_branch}");
insert_audit_event(
&transaction,
&NewAuditEvent {
action: "repository.default-branch",
actor,
target: &target,
outcome: "success",
correlation_id,
created_at: changed_at,
},
)?;
transaction.commit()?;
Ok(())
}
pub(crate) fn repository_description(&self, repository_id: &str) -> Result<String, StoreError> {
self.connection
.query_row(
"SELECT COALESCE(repository_profile.description, '')
FROM repository
LEFT JOIN repository_profile
ON repository_profile.repository_id = repository.id
WHERE repository.id = ?1",
[repository_id],
|row| row.get(0),
)
.optional()?
.ok_or_else(|| StoreError::Integrity(format!("repository {repository_id} disappeared")))
}
#[allow(
clippy::too_many_arguments,
reason = "an audited repository settings change includes repository identity and audit context"
)]
pub(crate) fn update_repository_settings(
&mut self,
owner: &str,
slug: &str,
actor: &str,
description: &str,
visibility: &str,
changed_at: i64,
correlation_id: &str,
) -> Result<(), StoreError> {
if !matches!(visibility, "public" | "private") {
return Err(StoreError::InvalidRepositoryVisibility);
}
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)?;
let access = repository_issue_access(&transaction, owner, slug, Some(actor))?;
if !access.can_maintain() {
return Err(StoreError::PullRequestDenied);
}
transaction.execute(
"INSERT INTO repository_profile (repository_id, description)
VALUES (?1, ?2)
ON CONFLICT (repository_id) DO UPDATE SET description = excluded.description",
rusqlite::params![access.repository.id, description],
)?;
transaction.execute(
"UPDATE repository SET visibility = ?2 WHERE id = ?1",
rusqlite::params![access.repository.id, visibility],
)?;
insert_audit_event(
&transaction,
&NewAuditEvent {
action: "repository.settings",
actor,
target: &format!("{owner}/{slug}"),
outcome: "success",
correlation_id,
created_at: changed_at,
},
)?;
transaction.commit()?;
Ok(())
}
#[allow(
clippy::too_many_arguments,
reason = "an audited collaborator change includes repository identity and audit context"
)]
pub(crate) fn update_repository_collaborator(
&mut self,
owner: &str,
slug: &str,
actor: &str,
username: &str,
role: Option<&str>,
changed_at: i64,
correlation_id: &str,
) -> Result<(), StoreError> {
if role.is_some_and(|role| !matches!(role, "maintainer" | "writer" | "reader")) {
return Err(StoreError::InvalidCollaboratorRole);
}
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)?;
let access = repository_issue_access(&transaction, owner, slug, Some(actor))?;
if !access.can_maintain() {
return Err(StoreError::PullRequestDenied);
}
let collaborator_id = match active_account_id(&transaction, username) {
Ok(id) => id,
Err(StoreError::AccountNotFound(_)) => {
return Err(StoreError::CollaboratorNotFound(username.to_owned()));
}
Err(error) => return Err(error),
};
if username == owner {
return Err(StoreError::OwnerCollaborator);
}
let changed = match role {
Some(role) => transaction.execute(
"INSERT INTO repository_collaborator
(repository_id, account_id, role, created_at)
VALUES (?1, ?2, ?3, ?4)
ON CONFLICT (repository_id, account_id) DO UPDATE SET role = excluded.role",
rusqlite::params![access.repository.id, collaborator_id, role, changed_at],
)?,
None => transaction.execute(
"DELETE FROM repository_collaborator
WHERE repository_id = ?1 AND account_id = ?2",
rusqlite::params![access.repository.id, collaborator_id],
)?,
};
if changed == 0 {
return Err(StoreError::CollaboratorNotFound(username.to_owned()));
}
insert_audit_event(
&transaction,
&NewAuditEvent {
action: if role.is_some() {
"collaborator.set"
} else {
"collaborator.remove"
},
actor,
target: &format!("{owner}/{slug}:{username}"),
outcome: "success",
correlation_id,
created_at: changed_at,
},
)?;
transaction.commit()?;
Ok(())
}
pub(crate) fn archive_repository_for_actor(
&mut self,
owner: &str,
slug: &str,
actor: &str,
changed_at: i64,
correlation_id: &str,
) -> Result<(), StoreError> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)?;
let access = repository_issue_access(&transaction, owner, slug, Some(actor))?;
if !access.can_maintain() {
return Err(StoreError::PullRequestDenied);
}
transaction.execute(
"UPDATE repository SET state = 'archived', archived_at = ?2 WHERE id = ?1",
rusqlite::params![access.repository.id, changed_at],
)?;
insert_audit_event(
&transaction,
&NewAuditEvent {
action: "repository.archive",
actor,
target: &format!("{owner}/{slug}"),
outcome: "success",
correlation_id,
created_at: changed_at,
},
)?;
transaction.commit()?;
Ok(())
}
pub(crate) fn rename_repository_for_owner(
&mut self,
owner: &str,
old_slug: &str,
new_slug: &str,
actor: &str,
changed_at: i64,
correlation_id: &str,
) -> Result<(), StoreError> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)?;
let owner_id = active_account_id(&transaction, owner)?;
let actor_id = active_account_id(&transaction, actor)?;
if actor_id != owner_id {
return Err(StoreError::PullRequestDenied);
}
let result = transaction.execute(
"UPDATE repository SET slug = ?3
WHERE owner_account_id = ?1 AND slug = ?2 AND state = 'active'",
rusqlite::params![owner_id, old_slug, new_slug],
);
match result {
Ok(1) => {}
Ok(0) => {
return Err(repository_state_error(
&transaction,
owner_id,
owner,
old_slug,
)?);
}
Ok(_) => unreachable!("an owner and slug identify one repository"),
Err(error) if is_unique_constraint(&error) => {
return Err(StoreError::RepositoryExists(
owner.to_owned(),
new_slug.to_owned(),
));
}
Err(error) => return Err(error.into()),
}
insert_audit_event(
&transaction,
&NewAuditEvent {
action: "repository.rename",
actor,
target: &format!("{owner}/{old_slug}->{new_slug}"),
outcome: "success",
correlation_id,
created_at: changed_at,
},
)?;
transaction.commit()?;
Ok(())
}
pub(crate) fn unarchive_repository_for_owner(
&mut self,
owner: &str,
slug: &str,
actor: &str,
changed_at: i64,
correlation_id: &str,
) -> Result<(), StoreError> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)?;
let owner_id = active_account_id(&transaction, owner)?;
let actor_id = active_account_id(&transaction, actor)?;
if actor_id != owner_id {
return Err(StoreError::PullRequestDenied);
}
let changed = transaction.execute(
"UPDATE repository
SET state = 'active', archived_at = NULL
WHERE owner_account_id = ?1 AND slug = ?2 AND state = 'archived'",
rusqlite::params![owner_id, slug],
)?;
if changed != 1 {
return Err(repository_state_error(&transaction, owner_id, owner, slug)?);
}
insert_audit_event(
&transaction,
&NewAuditEvent {
action: "repository.unarchive",
actor,
target: &format!("{owner}/{slug}"),
outcome: "success",
correlation_id,
created_at: changed_at,
},
)?;
transaction.commit()?;
Ok(())
}
#[allow(
dead_code,
reason = "some integration tests compile storage without authorization"
)]
pub(crate) fn repository_authorization(
&self,
owner: &str,
slug: &str,
username: Option<&str>,
) -> Result<RepositoryAuthorizationRecord, StoreError> {
let result = self.connection.query_row(
"SELECT repository.id, owner.username, repository.slug,
repository.visibility, repository.state, repository.object_format,
repository.created_at, repository.archived_at,
CASE
WHEN actor.state != 'active' THEN NULL
WHEN actor.id = repository.owner_account_id THEN 'owner'
ELSE repository_collaborator.role
END
FROM repository
JOIN account AS owner ON owner.id = repository.owner_account_id
LEFT JOIN account AS actor ON actor.username = ?3
LEFT JOIN repository_collaborator
ON repository_collaborator.repository_id = repository.id
AND repository_collaborator.account_id = actor.id
WHERE owner.username = ?1 AND repository.slug = ?2",
rusqlite::params![owner, slug, username],
|row| {
Ok(RepositoryAuthorizationRecord {
repository: repository_from_row(row)?,
role: row.get(8)?,
})
},
);
match result {
Ok(record) => Ok(record),
Err(rusqlite::Error::QueryReturnedNoRows) => Err(StoreError::RepositoryNotFound(
owner.to_owned(),
slug.to_owned(),
)),
Err(error) => Err(error.into()),
}
}
pub(crate) fn create_issue(&mut self, issue: &NewIssue<'_>) -> Result<IssueRecord, StoreError> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)?;
let access = repository_issue_access(
&transaction,
issue.owner,
issue.repository,
Some(issue.actor),
)?;
if !access.can_read() {
return Err(StoreError::IssueHidden);
}
let actor_id = access.active_actor_id.ok_or(StoreError::IssueDenied)?;
transaction.execute(
"INSERT INTO repository_counter (repository_id)
VALUES (?1) ON CONFLICT (repository_id) DO NOTHING",
[&access.repository.id],
)?;
let number = transaction.query_row(
"UPDATE repository_counter
SET next_issue_number = next_issue_number + 1
WHERE repository_id = ?1
RETURNING next_issue_number - 1",
[&access.repository.id],
|row| row.get(0),
)?;
let id: String =
transaction.query_row("SELECT lower(hex(randomblob(16)))", [], |row| row.get(0))?;
transaction.execute(
"INSERT INTO issue
(id, repository_id, number, title, body, state, author_account_id,
created_at, updated_at)
VALUES (?1, ?2, ?3, ?4, ?5, 'open', ?6, ?7, ?7)",
rusqlite::params![
id,
access.repository.id,
number,
issue.title,
issue.body,
actor_id,
issue.created_at,
],
)?;
let event = event::issue(
event::EventKind::IssueCreated,
&id,
number,
issue.title,
issue.body,
);
insert_domain_event(
&transaction,
&NewDomainEvent {
repository_id: &access.repository.id,
source_intent_id: None,
source_ordinal: None,
issue_id: Some(&id),
pull_request_id: None,
event: &event,
actor: issue.actor,
ref_name: None,
old_target: None,
new_target: None,
created_at: issue.created_at,
},
)?;
transaction.commit()?;
Ok(IssueRecord {
id,
number,
title: issue.title.to_owned(),
body: issue.body.to_owned(),
state: "open".to_owned(),
author: issue.actor.to_owned(),
created_at: issue.created_at,
updated_at: issue.created_at,
closed_at: None,
})
}
pub(crate) fn begin_pull_request_open(
&mut self,
intent: &NewPullRequestRefIntent<'_>,
) -> Result<PullRequestRefIntentRecord, StoreError> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)?;
let access = repository_issue_access(
&transaction,
intent.owner,
intent.repository,
Some(intent.actor),
)?;
if !access.can_read() {
return Err(StoreError::PullRequestHidden);
}
if !access.can_write_repository() {
return Err(StoreError::PullRequestDenied);
}
let actor_id = access
.active_actor_id
.ok_or(StoreError::PullRequestDenied)?;
transaction.execute(
"INSERT INTO repository_counter (repository_id)
VALUES (?1) ON CONFLICT (repository_id) DO NOTHING",
[&access.repository.id],
)?;
let number = transaction.query_row(
"UPDATE repository_counter
SET next_pull_request_number = next_pull_request_number + 1
WHERE repository_id = ?1
RETURNING next_pull_request_number - 1",
[&access.repository.id],
|row| row.get(0),
)?;
insert_pull_request_intent(
&transaction,
intent,
&access.repository.id,
actor_id,
number,
1,
None,
"open",
)?;
transaction.commit()?;
Ok(PullRequestRefIntentRecord::from_new(
intent,
access.repository.id,
actor_id,
number,
1,
None,
"open",
))
}
pub(crate) fn begin_pull_request_revision(
&mut self,
number: i64,
intent: &NewPullRequestRefIntent<'_>,
) -> Result<PullRequestRefIntentRecord, StoreError> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)?;
let access = repository_issue_access(
&transaction,
intent.owner,
intent.repository,
Some(intent.actor),
)?;
if !access.can_read() {
return Err(StoreError::PullRequestHidden);
}
if !access.can_write_repository() {
return Err(StoreError::PullRequestDenied);
}
let actor_id = access
.active_actor_id
.ok_or(StoreError::PullRequestDenied)?;
let stored = transaction
.query_row(
"SELECT id, title, body, state, base_ref, head_ref, head_object_id,
(SELECT COALESCE(MAX(number), 0) + 1
FROM pull_request_revision
WHERE pull_request_id = pull_request.id)
FROM pull_request
WHERE repository_id = ?1 AND number = ?2",
rusqlite::params![access.repository.id, number],
|row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
row.get::<_, String>(3)?,
row.get::<_, String>(4)?,
row.get::<_, String>(5)?,
row.get::<_, String>(6)?,
row.get::<_, i64>(7)?,
))
},
)
.optional()?
.ok_or_else(|| {
StoreError::PullRequestNotFound(
intent.owner.to_owned(),
intent.repository.to_owned(),
number,
)
})?;
let (pull_request_id, title, body, state, base_ref, head_ref, old_head, revision) = stored;
if state != "open" {
return Err(StoreError::PullRequestState);
}
if pull_request_id != intent.pull_request_id
|| title != intent.title
|| body != intent.body
|| base_ref != intent.base_ref
|| head_ref != intent.head_ref
{
return Err(StoreError::PullRequestIntentState(intent.id.to_owned()));
}
insert_pull_request_intent(
&transaction,
intent,
&access.repository.id,
actor_id,
number,
revision,
Some(&old_head),
"revise",
)?;
transaction.commit()?;
Ok(PullRequestRefIntentRecord::from_new(
intent,
access.repository.id,
actor_id,
number,
revision,
Some(old_head),
"revise",
))
}
pub(crate) fn complete_pull_request_ref_intent(&mut self, id: &str) -> Result<(), StoreError> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)?;
let intent = pull_request_intent(&transaction, id)?;
if intent.state != "pending" {
return Err(StoreError::PullRequestIntentState(id.to_owned()));
}
if intent.operation == "open" {
transaction.execute(
"INSERT INTO pull_request
(id, repository_id, number, title, body, state, author_account_id,
base_ref, head_ref, base_object_id, head_object_id, created_at, updated_at)
VALUES (?1, ?2, ?3, ?4, ?5, 'open', ?6, ?7, ?8, ?9, ?10, ?11, ?11)",
rusqlite::params![
intent.pull_request_id,
intent.repository_id,
intent.pull_request_number,
intent.title,
intent.body,
intent.author_account_id,
intent.base_ref,
intent.head_ref,
intent.base_object_id,
intent.head_object_id,
intent.created_at,
],
)?;
} else {
let changed = transaction.execute(
"UPDATE pull_request
SET base_object_id = ?2, head_object_id = ?3, updated_at = ?4
WHERE id = ?1 AND state = 'open' AND head_object_id = ?5",
rusqlite::params![
intent.pull_request_id,
intent.base_object_id,
intent.head_object_id,
intent.created_at,
intent.old_head_object_id,
],
)?;
if changed != 1 {
return Err(StoreError::PullRequestIntentState(id.to_owned()));
}
}
let revision_id: String =
transaction.query_row("SELECT lower(hex(randomblob(16)))", [], |row| row.get(0))?;
transaction.execute(
"INSERT INTO pull_request_revision
(id, pull_request_id, number, author_account_id, base_object_id,
head_object_id, created_at)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
rusqlite::params![
revision_id,
intent.pull_request_id,
intent.revision_number,
intent.author_account_id,
intent.base_object_id,
intent.head_object_id,
intent.created_at,
],
)?;
let kind = if intent.operation == "open" {
event::EventKind::PullRequestCreated
} else {
event::EventKind::PullRequestRevised
};
let event = event::pull_request(
kind,
&intent.pull_request_id,
intent.pull_request_number,
intent.revision_number,
&intent.title,
&intent.base_ref,
&intent.head_ref,
&intent.base_object_id,
&intent.head_object_id,
);
insert_domain_event(
&transaction,
&NewDomainEvent {
repository_id: &intent.repository_id,
source_intent_id: None,
source_ordinal: None,
issue_id: None,
pull_request_id: Some(&intent.pull_request_id),
event: &event,
actor: &intent.actor,
ref_name: None,
old_target: None,
new_target: None,
created_at: intent.created_at,
},
)?;
transaction.execute(
"UPDATE pull_request_ref_intent SET state = 'completed'
WHERE id = ?1 AND state = 'pending'",
[id],
)?;
transaction.commit()?;
Ok(())
}
pub(crate) fn abandon_pull_request_ref_intent(&self, id: &str) -> Result<(), StoreError> {
let changed = self.connection.execute(
"UPDATE pull_request_ref_intent SET state = 'abandoned'
WHERE id = ?1 AND state = 'pending'",
[id],
)?;
if changed == 1 {
Ok(())
} else {
Err(StoreError::PullRequestIntentState(id.to_owned()))
}
}
pub(crate) fn incomplete_pull_request_ref_intents(
&self,
) -> Result<Vec<PullRequestRefIntentRecord>, StoreError> {
let mut statement = self.connection.prepare(
"SELECT id, repository_id, pull_request_id, pull_request_number,
revision_number, operation, title, body, author_account_id, actor,
base_ref, head_ref, base_object_id, old_head_object_id,
head_object_id, state, created_at
FROM pull_request_ref_intent WHERE state = 'pending'
ORDER BY created_at, id",
)?;
statement
.query_map([], pull_request_intent_from_row)?
.collect::<Result<Vec<_>, _>>()
.map_err(Into::into)
}
pub(crate) fn pull_request(
&self,
owner: &str,
repository: &str,
number: i64,
actor: Option<&str>,
) -> Result<PullRequestDetail, StoreError> {
self.pull_request_with_activity(owner, repository, number, actor, None)
}
#[allow(
clippy::too_many_arguments,
reason = "the pull-request detail has independent review and timeline pages"
)]
pub(crate) fn pull_request_detail_page(
&self,
owner: &str,
repository: &str,
number: i64,
actor: Option<&str>,
reviews_page: usize,
timeline_page: usize,
page_size: usize,
) -> Result<PullRequestDetail, StoreError> {
self.pull_request_with_activity(
owner,
repository,
number,
actor,
Some((reviews_page, timeline_page, page_size)),
)
}
fn pull_request_with_activity(
&self,
owner: &str,
repository: &str,
number: i64,
actor: Option<&str>,
activity: Option<(usize, usize, usize)>,
) -> Result<PullRequestDetail, StoreError> {
let access = repository_issue_access(&self.connection, owner, repository, actor)?;
if !access.can_read() {
return Err(StoreError::PullRequestHidden);
}
let pull_request = self
.connection
.query_row(
"SELECT pull_request.id, pull_request.number, pull_request.title,
pull_request.body, pull_request.state, author.username,
pull_request.base_ref, pull_request.head_ref,
pull_request.base_object_id, pull_request.head_object_id,
pull_request.created_at, pull_request.updated_at
FROM pull_request
JOIN account AS author ON author.id = pull_request.author_account_id
WHERE pull_request.repository_id = ?1 AND pull_request.number = ?2",
rusqlite::params![access.repository.id, number],
pull_request_from_row,
)
.optional()?
.ok_or_else(|| {
StoreError::PullRequestNotFound(owner.to_owned(), repository.to_owned(), number)
})?;
let mut statement = self.connection.prepare(
"SELECT pull_request_revision.id, pull_request_revision.number,
author.username, pull_request_revision.base_object_id,
pull_request_revision.head_object_id, pull_request_revision.created_at
FROM pull_request_revision
JOIN account AS author ON author.id = pull_request_revision.author_account_id
WHERE pull_request_revision.pull_request_id = ?1
ORDER BY pull_request_revision.number",
)?;
let revisions = statement
.query_map([&pull_request.id], |row| {
Ok(PullRequestRevisionRecord {
id: row.get(0)?,
number: row.get(1)?,
author: row.get(2)?,
base_object_id: row.get(3)?,
head_object_id: row.get(4)?,
created_at: row.get(5)?,
})
})?
.collect::<Result<Vec<_>, _>>()?;
let (reviews, reviews_page, reviews_has_next) = {
let mut statement = self.connection.prepare(
"SELECT pull_request_review.id, pull_request_revision.number,
author.username, pull_request_review.kind,
pull_request_review.body, pull_request_review.commit_object_id,
pull_request_review.path, pull_request_review.side,
pull_request_review.line, pull_request_review.created_at
FROM pull_request_review
JOIN pull_request_revision
ON pull_request_revision.id = pull_request_review.revision_id
JOIN account AS author
ON author.id = pull_request_review.author_account_id
WHERE pull_request_review.pull_request_id = ?1
ORDER BY pull_request_review.created_at DESC, pull_request_review.id DESC
LIMIT ?2 OFFSET ?3",
)?;
let (page, page_size) =
activity.map_or((1, usize::MAX - 1), |value| (value.0, value.2));
let limit = i64::try_from(page_size.saturating_add(1)).unwrap_or(i64::MAX);
let offset = if activity.is_some() {
page_offset(page, page_size)?
} else {
0
};
let mut records = statement
.query_map(
rusqlite::params![pull_request.id, limit, offset],
pull_request_review_from_row,
)?
.collect::<Result<Vec<_>, _>>()?;
let has_next = activity.is_some() && records.len() > page_size;
if activity.is_some() {
records.truncate(page_size);
}
records.reverse();
(records, page, has_next)
};
let (timeline, timeline_page, timeline_has_next) = {
let mut statement = self.connection.prepare(
"SELECT sequence, kind, actor, payload, created_at
FROM repository_event
WHERE pull_request_id = ?1 ORDER BY sequence DESC
LIMIT ?2 OFFSET ?3",
)?;
let (page, page_size) =
activity.map_or((1, usize::MAX - 1), |value| (value.1, value.2));
let limit = i64::try_from(page_size.saturating_add(1)).unwrap_or(i64::MAX);
let offset = if activity.is_some() {
page_offset(page, page_size)?
} else {
0
};
let mut records = statement
.query_map(rusqlite::params![pull_request.id, limit, offset], |row| {
Ok(PullRequestTimelineRecord {
sequence: row.get(0)?,
kind: row.get(1)?,
actor: row.get(2)?,
payload: row.get(3)?,
created_at: row.get(4)?,
})
})?
.collect::<Result<Vec<_>, _>>()?;
let has_next = activity.is_some() && records.len() > page_size;
if activity.is_some() {
records.truncate(page_size);
}
records.reverse();
(records, page, has_next)
};
let is_open = pull_request.state == "open";
let author_account_id: i64 = self.connection.query_row(
"SELECT author_account_id FROM pull_request WHERE id = ?1",
[&pull_request.id],
|row| row.get(0),
)?;
let can_edit = pull_request.state != "merged" && access.can_write_issue(author_account_id);
let can_revise = is_open && access.can_write_repository();
let can_merge = access.can_maintain() && pull_request.state == "open";
Ok(PullRequestDetail {
can_review: is_open && access.active_actor_id.is_some() && access.can_read(),
repository: access.repository,
pull_request,
revisions,
reviews,
timeline,
reviews_page,
reviews_has_next,
timeline_page,
timeline_has_next,
can_edit,
can_change_state: can_edit,
can_revise,
can_merge,
})
}
pub(crate) fn create_pull_request_review(
&mut self,
review: &NewPullRequestReview<'_>,
) -> Result<String, StoreError> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)?;
let access = repository_issue_access(
&transaction,
review.owner,
review.repository,
Some(review.actor),
)?;
if !access.can_read() {
return Err(StoreError::PullRequestHidden);
}
let actor_id = access
.active_actor_id
.ok_or(StoreError::PullRequestDenied)?;
let pull_request = transaction
.query_row(
"SELECT id, state FROM pull_request
WHERE repository_id = ?1 AND number = ?2",
rusqlite::params![access.repository.id, review.number],
|row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)),
)
.optional()?
.ok_or_else(|| {
StoreError::PullRequestNotFound(
review.owner.to_owned(),
review.repository.to_owned(),
review.number,
)
})?;
if pull_request.1 != "open" {
return Err(StoreError::PullRequestState);
}
let revision = transaction
.query_row(
"SELECT id, base_object_id, head_object_id
FROM pull_request_revision
WHERE pull_request_id = ?1 AND number = ?2",
rusqlite::params![pull_request.0, review.revision],
|row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
))
},
)
.optional()?
.ok_or(StoreError::PullRequestRevisionNotFound)?;
if review.kind == "line-comment" {
let expected = match review.side {
Some("base") => revision.1.as_str(),
Some("head") => revision.2.as_str(),
_ => return Err(StoreError::PullRequestReviewAnchor),
};
if review.commit_object_id != Some(expected) {
return Err(StoreError::PullRequestReviewAnchor);
}
}
let id: String =
transaction.query_row("SELECT lower(hex(randomblob(16)))", [], |row| row.get(0))?;
transaction.execute(
"INSERT INTO pull_request_review
(id, pull_request_id, revision_id, author_account_id, kind, body,
commit_object_id, path, side, line, created_at)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)",
rusqlite::params![
id,
pull_request.0,
revision.0,
actor_id,
review.kind,
review.body,
review.commit_object_id,
review.path,
review.side,
review.line,
review.created_at,
],
)?;
transaction.execute(
"UPDATE pull_request SET updated_at = ?2 WHERE id = ?1",
rusqlite::params![pull_request.0, review.created_at],
)?;
let kind = match review.kind {
"comment" => event::EventKind::PullRequestCommented,
"line-comment" => event::EventKind::PullRequestLineCommented,
"approved" => event::EventKind::PullRequestApproved,
"changes-requested" => event::EventKind::PullRequestChangesRequested,
_ => return Err(StoreError::PullRequestReviewAnchor),
};
let event = event::pull_request_review(
kind,
&pull_request.0,
review.number,
&id,
review.revision,
review.body,
review.commit_object_id,
review.path,
review.side,
review.line,
);
insert_pull_request_event(
&transaction,
&access.repository.id,
&pull_request.0,
review.actor,
review.created_at,
&event,
)?;
transaction.commit()?;
Ok(id)
}
pub(crate) fn pull_requests(
&self,
owner: &str,
repository: &str,
actor: Option<&str>,
) -> Result<(RepositoryRecord, Vec<PullRequestRecord>, bool), StoreError> {
let access = repository_issue_access(&self.connection, owner, repository, actor)?;
if !access.can_read() {
return Err(StoreError::PullRequestHidden);
}
let mut statement = self.connection.prepare(
"SELECT pull_request.id, pull_request.number, pull_request.title,
pull_request.body, pull_request.state, author.username,
pull_request.base_ref, pull_request.head_ref,
pull_request.base_object_id, pull_request.head_object_id,
pull_request.created_at, pull_request.updated_at
FROM pull_request
JOIN account AS author ON author.id = pull_request.author_account_id
WHERE pull_request.repository_id = ?1
ORDER BY pull_request.number DESC",
)?;
let pull_requests = statement
.query_map([&access.repository.id], pull_request_from_row)?
.collect::<Result<Vec<_>, _>>()?;
let can_create = access.can_write_repository();
Ok((access.repository, pull_requests, can_create))
}
pub(crate) fn pull_request_page(
&self,
owner: &str,
repository: &str,
actor: Option<&str>,
state: &str,
page: usize,
page_size: usize,
) -> Result<(RepositoryRecord, RecordPage<PullRequestRecord>, bool), StoreError> {
let access = repository_issue_access(&self.connection, owner, repository, actor)?;
if !access.can_read() {
return Err(StoreError::PullRequestHidden);
}
let offset = page
.checked_sub(1)
.and_then(|page| page.checked_mul(page_size))
.ok_or(StoreError::EventLimit)?;
let limit = i64::try_from(page_size.checked_add(1).ok_or(StoreError::EventLimit)?)
.map_err(|_| StoreError::EventLimit)?;
let offset = i64::try_from(offset).map_err(|_| StoreError::EventLimit)?;
let mut statement = self.connection.prepare(
"SELECT pull_request.id, pull_request.number, pull_request.title,
pull_request.body, pull_request.state, author.username,
pull_request.base_ref, pull_request.head_ref,
pull_request.base_object_id, pull_request.head_object_id,
pull_request.created_at, pull_request.updated_at
FROM pull_request
JOIN account AS author ON author.id = pull_request.author_account_id
WHERE pull_request.repository_id = ?1
AND (?2 = 'all' OR pull_request.state = ?2)
ORDER BY pull_request.number DESC
LIMIT ?3 OFFSET ?4",
)?;
let mut items = statement
.query_map(
rusqlite::params![access.repository.id, state, limit, offset],
pull_request_from_row,
)?
.collect::<Result<Vec<_>, _>>()?;
let has_next = items.len() > page_size;
items.truncate(page_size);
let can_create = access.can_write_repository();
Ok((
access.repository,
RecordPage {
items,
page,
has_next,
},
can_create,
))
}
#[allow(
clippy::too_many_arguments,
reason = "a pull-request edit includes repository identity, content, actor, and time"
)]
pub(crate) fn edit_pull_request(
&mut self,
owner: &str,
repository: &str,
number: i64,
actor: &str,
title: &str,
body: &str,
changed_at: i64,
) -> Result<(), StoreError> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)?;
let access = repository_issue_access(&transaction, owner, repository, Some(actor))?;
let current = transaction
.query_row(
"SELECT id, author_account_id, state
FROM pull_request WHERE repository_id = ?1 AND number = ?2",
rusqlite::params![access.repository.id, number],
|row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, i64>(1)?,
row.get::<_, String>(2)?,
))
},
)
.optional()?
.ok_or_else(|| {
StoreError::PullRequestNotFound(owner.to_owned(), repository.to_owned(), number)
})?;
if !access.can_write_issue(current.1) {
return Err(StoreError::PullRequestDenied);
}
if current.2 == "merged" {
return Err(StoreError::PullRequestState);
}
transaction.execute(
"UPDATE pull_request SET title = ?2, body = ?3, updated_at = ?4 WHERE id = ?1",
rusqlite::params![current.0, title, body, changed_at],
)?;
let event = event::pull_request_change(
event::EventKind::PullRequestEdited,
¤t.0,
number,
title,
body,
¤t.2,
);
insert_pull_request_event(
&transaction,
&access.repository.id,
¤t.0,
actor,
changed_at,
&event,
)?;
transaction.commit()?;
Ok(())
}
pub(crate) fn set_pull_request_state(
&mut self,
owner: &str,
repository: &str,
number: i64,
actor: &str,
state: &str,
changed_at: i64,
) -> Result<(), StoreError> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)?;
let access = repository_issue_access(&transaction, owner, repository, Some(actor))?;
let current = transaction
.query_row(
"SELECT id, author_account_id, title, body, state
FROM pull_request WHERE repository_id = ?1 AND number = ?2",
rusqlite::params![access.repository.id, number],
|row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, i64>(1)?,
row.get::<_, String>(2)?,
row.get::<_, String>(3)?,
row.get::<_, String>(4)?,
))
},
)
.optional()?
.ok_or_else(|| {
StoreError::PullRequestNotFound(owner.to_owned(), repository.to_owned(), number)
})?;
if !access.can_write_issue(current.1) {
return Err(StoreError::PullRequestDenied);
}
if current.4 == "merged" || current.4 == state {
return Err(StoreError::PullRequestState);
}
transaction.execute(
"UPDATE pull_request SET state = ?2, updated_at = ?3 WHERE id = ?1",
rusqlite::params![current.0, state, changed_at],
)?;
let kind = if state == "closed" {
event::EventKind::PullRequestClosed
} else {
event::EventKind::PullRequestReopened
};
let event =
event::pull_request_change(kind, ¤t.0, number, ¤t.2, ¤t.3, state);
insert_pull_request_event(
&transaction,
&access.repository.id,
¤t.0,
actor,
changed_at,
&event,
)?;
transaction.commit()?;
Ok(())
}
pub(crate) fn edit_issue(
&mut self,
change: &IssueChange<'_>,
title: &str,
body: &str,
) -> Result<(), StoreError> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)?;
let (access, current) = issue_mutation_context(
&transaction,
change.owner,
change.repository,
change.number,
change.actor,
)?;
if !access.can_write_issue(current.author_account_id) {
return Err(StoreError::IssueDenied);
}
transaction.execute(
"UPDATE issue SET title = ?2, body = ?3, updated_at = ?4 WHERE id = ?1",
rusqlite::params![current.issue.id, title, body, change.changed_at],
)?;
let event = event::issue(
event::EventKind::IssueEdited,
¤t.issue.id,
change.number,
title,
body,
);
insert_issue_event(
&transaction,
&access.repository.id,
¤t.issue.id,
change.actor,
change.changed_at,
&event,
)?;
transaction.commit()?;
Ok(())
}
pub(crate) fn comment_issue(
&mut self,
owner: &str,
repository: &str,
number: i64,
actor: &str,
body: &str,
created_at: i64,
) -> Result<String, StoreError> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)?;
let (access, current) =
issue_mutation_context(&transaction, owner, repository, number, actor)?;
let actor_id = access.active_actor_id.ok_or(StoreError::IssueDenied)?;
if !access.can_read() {
return Err(StoreError::IssueDenied);
}
let comment_id: String =
transaction.query_row("SELECT lower(hex(randomblob(16)))", [], |row| row.get(0))?;
transaction.execute(
"INSERT INTO issue_comment
(id, issue_id, author_account_id, body, created_at)
VALUES (?1, ?2, ?3, ?4, ?5)",
rusqlite::params![comment_id, current.issue.id, actor_id, body, created_at],
)?;
transaction.execute(
"UPDATE issue SET updated_at = ?2 WHERE id = ?1",
rusqlite::params![current.issue.id, created_at],
)?;
let event = event::issue_comment(¤t.issue.id, number, &comment_id, actor, body);
insert_issue_event(
&transaction,
&access.repository.id,
¤t.issue.id,
actor,
created_at,
&event,
)?;
transaction.commit()?;
Ok(comment_id)
}
pub(crate) fn set_issue_state(
&mut self,
owner: &str,
repository: &str,
number: i64,
actor: &str,
state: &str,
changed_at: i64,
) -> Result<(), StoreError> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)?;
let (access, current) =
issue_mutation_context(&transaction, owner, repository, number, actor)?;
if !access.can_write_issue(current.author_account_id) {
return Err(StoreError::IssueDenied);
}
if current.issue.state == state {
return Err(StoreError::IssueState(state.to_owned()));
}
let (closed_at, kind) = match state {
"closed" => (Some(changed_at), event::EventKind::IssueClosed),
"open" => (None, event::EventKind::IssueReopened),
_ => return Err(StoreError::IssueState(state.to_owned())),
};
transaction.execute(
"UPDATE issue
SET state = ?2, updated_at = ?3, closed_at = ?4
WHERE id = ?1",
rusqlite::params![current.issue.id, state, changed_at, closed_at],
)?;
let event = event::issue_state(kind, ¤t.issue.id, number, state);
insert_issue_event(
&transaction,
&access.repository.id,
¤t.issue.id,
actor,
changed_at,
&event,
)?;
transaction.commit()?;
Ok(())
}
pub(crate) fn issues(
&self,
owner: &str,
repository: &str,
actor: Option<&str>,
) -> Result<(RepositoryRecord, Vec<IssueRecord>), StoreError> {
let access = repository_issue_access(&self.connection, owner, repository, actor)?;
if !access.can_read() {
return Err(StoreError::IssueHidden);
}
let mut statement = self.connection.prepare(
"SELECT issue.id, issue.number, issue.title, issue.body, issue.state,
account.username, issue.created_at, issue.updated_at, issue.closed_at
FROM issue
JOIN account ON account.id = issue.author_account_id
WHERE issue.repository_id = ?1
ORDER BY issue.number DESC LIMIT 1000",
)?;
let issues = statement
.query_map([&access.repository.id], issue_from_row)?
.collect::<Result<Vec<_>, _>>()?;
drop(statement);
Ok((access.repository, issues))
}
pub(crate) fn issue_page(
&self,
owner: &str,
repository: &str,
actor: Option<&str>,
state: &str,
page: usize,
page_size: usize,
) -> Result<(RepositoryRecord, RecordPage<IssueRecord>), StoreError> {
let access = repository_issue_access(&self.connection, owner, repository, actor)?;
if !access.can_read() {
return Err(StoreError::IssueHidden);
}
let offset = page
.checked_sub(1)
.and_then(|page| page.checked_mul(page_size))
.ok_or(StoreError::EventLimit)?;
let limit = i64::try_from(page_size.checked_add(1).ok_or(StoreError::EventLimit)?)
.map_err(|_| StoreError::EventLimit)?;
let offset = i64::try_from(offset).map_err(|_| StoreError::EventLimit)?;
let mut statement = self.connection.prepare(
"SELECT issue.id, issue.number, issue.title, issue.body, issue.state,
account.username, issue.created_at, issue.updated_at, issue.closed_at
FROM issue
JOIN account ON account.id = issue.author_account_id
WHERE issue.repository_id = ?1
AND (?2 = 'all' OR issue.state = ?2)
ORDER BY issue.number DESC
LIMIT ?3 OFFSET ?4",
)?;
let mut items = statement
.query_map(
rusqlite::params![access.repository.id, state, limit, offset],
issue_from_row,
)?
.collect::<Result<Vec<_>, _>>()?;
let has_next = items.len() > page_size;
items.truncate(page_size);
Ok((
access.repository,
RecordPage {
items,
page,
has_next,
},
))
}
#[allow(
clippy::too_many_arguments,
reason = "the issue detail has independent comment and timeline pages"
)]
pub(crate) fn issue_detail(
&self,
owner: &str,
repository: &str,
number: i64,
actor: Option<&str>,
comments_page: usize,
timeline_page: usize,
page_size: usize,
) -> Result<IssueDetail, StoreError> {
let access = repository_issue_access(&self.connection, owner, repository, actor)?;
if !access.can_read() {
return Err(StoreError::IssueHidden);
}
let issue = find_issue(
&self.connection,
&access.repository,
owner,
repository,
number,
)?;
let limit = i64::try_from(page_size.checked_add(1).ok_or(StoreError::EventLimit)?)
.map_err(|_| StoreError::EventLimit)?;
let comments_offset = page_offset(comments_page, page_size)?;
let timeline_offset = page_offset(timeline_page, page_size)?;
let (comments, comments_has_next) = {
let mut statement = self.connection.prepare(
"SELECT issue_comment.id, account.username, issue_comment.body,
issue_comment.created_at
FROM issue_comment
JOIN account ON account.id = issue_comment.author_account_id
WHERE issue_comment.issue_id = ?1
ORDER BY issue_comment.created_at DESC, issue_comment.id DESC
LIMIT ?2 OFFSET ?3",
)?;
let mut records = statement
.query_map(
rusqlite::params![issue.issue.id, limit, comments_offset],
|row| {
Ok(IssueCommentRecord {
id: row.get(0)?,
author: row.get(1)?,
body: row.get(2)?,
created_at: row.get(3)?,
})
},
)?
.collect::<Result<Vec<_>, _>>()?;
let has_next = records.len() > page_size;
records.truncate(page_size);
records.reverse();
(records, has_next)
};
let (timeline, timeline_has_next) = {
let mut statement = self.connection.prepare(
"SELECT sequence, kind, actor, payload, created_at
FROM repository_event WHERE issue_id = ?1 ORDER BY sequence DESC
LIMIT ?2 OFFSET ?3",
)?;
let mut records = statement
.query_map(
rusqlite::params![issue.issue.id, limit, timeline_offset],
|row| {
Ok(IssueTimelineRecord {
sequence: row.get(0)?,
kind: row.get(1)?,
actor: row.get(2)?,
payload: row.get(3)?,
created_at: row.get(4)?,
})
},
)?
.collect::<Result<Vec<_>, _>>()?;
let has_next = records.len() > page_size;
records.truncate(page_size);
records.reverse();
(records, has_next)
};
Ok(IssueDetail {
can_comment: access.active_actor_id.is_some() && access.can_read(),
can_edit: access.can_write_issue(issue.author_account_id),
repository: access.repository,
issue: issue.issue,
comments,
timeline,
comments_page,
comments_has_next,
timeline_page,
timeline_has_next,
})
}
pub(crate) fn watch(
&self,
owner: &str,
repository: &str,
actor: Option<&str>,
) -> Result<(RepositoryRecord, Option<WatchRecord>), StoreError> {
let access = repository_issue_access(&self.connection, owner, repository, actor)?;
if !access.can_read() {
return Err(StoreError::WatchDenied);
}
let watch = match access.active_actor_id {
Some(actor_id) => self
.connection
.query_row(
"SELECT id, created_at, updated_at
FROM watch WHERE repository_id = ?1 AND account_id = ?2",
rusqlite::params![access.repository.id, actor_id],
|row| {
Ok(WatchRecord {
id: row.get(0)?,
created_at: row.get(1)?,
updated_at: row.get(2)?,
})
},
)
.optional()?,
None => None,
};
Ok((access.repository, watch))
}
pub(crate) fn set_watch(
&mut self,
owner: &str,
repository: &str,
actor: &str,
watching: bool,
changed_at: i64,
) -> Result<Option<WatchRecord>, StoreError> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)?;
let access = repository_issue_access(&transaction, owner, repository, Some(actor))?;
let actor_id = access.active_actor_id.ok_or(StoreError::WatchDenied)?;
if !access.can_read() {
return Err(StoreError::WatchDenied);
}
if !watching {
transaction.execute(
"DELETE FROM watch WHERE repository_id = ?1 AND account_id = ?2",
rusqlite::params![access.repository.id, actor_id],
)?;
transaction.commit()?;
return Ok(None);
}
transaction.execute(
"INSERT INTO watch
(id, repository_id, account_id, created_at, updated_at)
VALUES (lower(hex(randomblob(16))), ?1, ?2, ?3, ?3)
ON CONFLICT (repository_id, account_id) DO UPDATE SET
updated_at = excluded.updated_at",
rusqlite::params![access.repository.id, actor_id, changed_at],
)?;
let record = transaction.query_row(
"SELECT id, created_at, updated_at
FROM watch WHERE repository_id = ?1 AND account_id = ?2",
rusqlite::params![access.repository.id, actor_id],
|row| {
Ok(WatchRecord {
id: row.get(0)?,
created_at: row.get(1)?,
updated_at: row.get(2)?,
})
},
)?;
transaction.commit()?;
Ok(Some(record))
}
pub(crate) fn create_feed_token(
&mut self,
actor: &str,
token_hash: &[u8; 32],
created_at: i64,
) -> Result<FeedTokenRecord, StoreError> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)?;
let account_id = active_account_id(&transaction, actor)?;
let active_tokens: i64 = transaction.query_row(
"SELECT count(*) FROM feed_token
WHERE account_id = ?1 AND revoked_at IS NULL",
[account_id],
|row| row.get(0),
)?;
if active_tokens >= MAX_ACTIVE_FEED_TOKENS {
return Err(StoreError::FeedTokenLimit);
}
transaction.execute(
"INSERT INTO feed_token
(id, token_hash, account_id, scope, repository_id, created_at, revoked_at)
VALUES (lower(hex(randomblob(16))), ?1, ?2, 'watched', NULL, ?3, NULL)",
rusqlite::params![token_hash.as_slice(), account_id, created_at],
)?;
let record = transaction.query_row(
"SELECT id, scope, created_at, revoked_at
FROM feed_token
WHERE rowid = last_insert_rowid()",
[],
feed_token_from_row,
)?;
transaction.commit()?;
Ok(record)
}
pub(crate) fn feed_tokens(&self, actor: &str) -> Result<Vec<FeedTokenRecord>, StoreError> {
let account_id = active_account_id(&self.connection, actor)?;
let mut statement = self.connection.prepare(
"SELECT id, scope, created_at, revoked_at
FROM feed_token
WHERE account_id = ?1
ORDER BY created_at DESC, id
LIMIT 100",
)?;
statement
.query_map([account_id], feed_token_from_row)?
.collect::<Result<Vec<_>, _>>()
.map_err(Into::into)
}
pub(crate) fn rotate_feed_token(
&mut self,
actor: &str,
id: &str,
token_hash: &[u8; 32],
changed_at: i64,
) -> Result<FeedTokenRecord, StoreError> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)?;
let account_id = active_account_id(&transaction, actor)?;
transaction
.query_row(
"SELECT 1 FROM feed_token
WHERE id = ?1 AND account_id = ?2 AND revoked_at IS NULL",
rusqlite::params![id, account_id],
|_| Ok(()),
)
.optional()?
.ok_or(StoreError::FeedTokenNotFound)?;
transaction.execute(
"UPDATE feed_token SET revoked_at = ?3
WHERE id = ?1 AND account_id = ?2 AND revoked_at IS NULL",
rusqlite::params![id, account_id, changed_at],
)?;
transaction.execute(
"INSERT INTO feed_token
(id, token_hash, account_id, scope, repository_id, created_at, revoked_at)
VALUES (lower(hex(randomblob(16))), ?1, ?2, 'watched', NULL, ?3, NULL)",
rusqlite::params![token_hash.as_slice(), account_id, changed_at],
)?;
let record = transaction.query_row(
"SELECT id, scope, created_at, revoked_at
FROM feed_token
WHERE rowid = last_insert_rowid()",
[],
feed_token_from_row,
)?;
transaction.commit()?;
Ok(record)
}
pub(crate) fn revoke_feed_token(
&mut self,
actor: &str,
id: &str,
revoked_at: i64,
) -> Result<(), StoreError> {
let account_id = active_account_id(&self.connection, actor)?;
let changed = self.connection.execute(
"UPDATE feed_token SET revoked_at = ?3
WHERE id = ?1 AND account_id = ?2 AND revoked_at IS NULL",
rusqlite::params![id, account_id, revoked_at],
)?;
if changed == 0 {
return Err(StoreError::FeedTokenNotFound);
}
Ok(())
}
pub(crate) fn token_feed_events(
&self,
token_hash: &[u8; 32],
limit: usize,
) -> Result<TokenFeedPage, StoreError> {
let limit = i64::try_from(limit).map_err(|_| StoreError::EventLimit)?;
let grant = self
.connection
.query_row(
"SELECT feed_token.scope, account.id, account.username
FROM feed_token
JOIN account ON account.id = feed_token.account_id
WHERE feed_token.token_hash = ?1 AND feed_token.revoked_at IS NULL
AND account.state = 'active'",
[token_hash.as_slice()],
|row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, i64>(1)?,
row.get::<_, String>(2)?,
))
},
)
.optional()?
.ok_or(StoreError::FeedTokenNotFound)?;
if grant.0 != "watched" {
return Err(StoreError::InvalidFeedScope);
}
let events = watched_activity_events(&self.connection, grant.1, None, limit)?;
Ok(TokenFeedPage {
username: grant.2,
events,
})
}
pub(crate) fn watched_activity_page(
&self,
actor: &str,
before: Option<&ActivityCursor>,
limit: usize,
) -> Result<ActivityPage, StoreError> {
if limit == 0 || limit > 100 {
return Err(StoreError::EventLimit);
}
let account_id = active_account_id(&self.connection, actor)?;
let query_limit = limit.checked_add(1).ok_or(StoreError::EventLimit)?;
let query_limit = i64::try_from(query_limit).map_err(|_| StoreError::EventLimit)?;
let mut events =
watched_activity_events(&self.connection, account_id, before, query_limit)?;
let has_next = events.len() > limit;
events.truncate(limit);
let next_before = has_next.then(|| {
let event = &events[events.len() - 1].event;
ActivityCursor {
created_at: event.created_at,
event_id: event.event_id.clone(),
}
});
Ok(ActivityPage {
events,
next_before,
})
}
pub(crate) fn visit_metadata_search_candidates(
&self,
actor: Option<&str>,
limit: usize,
mut visit: impl FnMut(MetadataSearchCandidate) -> bool,
) -> Result<bool, StoreError> {
let query_limit = limit.checked_add(1).ok_or(StoreError::EventLimit)?;
let query_limit = i64::try_from(query_limit).map_err(|_| StoreError::EventLimit)?;
let mut statement = self.connection.prepare(
"WITH visible_repository AS (
SELECT repository.id, owner.username AS owner, repository.slug
FROM repository
JOIN account AS owner ON owner.id = repository.owner_account_id
LEFT JOIN account AS actor
ON actor.username = ?1 AND actor.state = 'active'
WHERE repository.state = 'active'
AND (repository.visibility = 'public'
OR repository.owner_account_id = actor.id
OR EXISTS (
SELECT 1 FROM repository_collaborator
WHERE repository_collaborator.repository_id = repository.id
AND repository_collaborator.account_id = actor.id
))
)
SELECT 'repository', id, owner, slug,
CAST(owner || '/' || slug AS BLOB), X''
FROM visible_repository
ORDER BY owner, slug
LIMIT ?2",
)?;
let mut rows = statement.query(rusqlite::params![actor, query_limit])?;
let mut visited = 0_usize;
while let Some(row) = rows.next()? {
if visited == limit {
return Ok(true);
}
visited += 1;
let candidate = MetadataSearchCandidate {
kind: row.get(0)?,
record_id: row.get(1)?,
owner: row.get(2)?,
repository: row.get(3)?,
title: row.get(4)?,
body: row.get(5)?,
};
if !visit(candidate) {
return Ok(true);
}
}
Ok(false)
}
#[allow(
dead_code,
reason = "some integration tests compile storage without authorization"
)]
pub(crate) fn active_repositories(&self) -> Result<Vec<RepositoryRecord>, StoreError> {
let mut statement = self.connection.prepare(
"SELECT repository.id, account.username, repository.slug,
repository.visibility, repository.state, repository.object_format,
repository.created_at, repository.archived_at
FROM repository
JOIN account ON account.id = repository.owner_account_id
WHERE repository.state = 'active'
ORDER BY account.username, repository.slug",
)?;
statement
.query_map([], repository_from_row)?
.collect::<Result<Vec<_>, _>>()
.map_err(Into::into)
}
pub(crate) fn all_repositories(&self) -> Result<Vec<RepositoryRecord>, StoreError> {
let mut statement = self.connection.prepare(
"SELECT repository.id, account.username, repository.slug,
repository.visibility, repository.state, repository.object_format,
repository.created_at, repository.archived_at
FROM repository
JOIN account ON account.id = repository.owner_account_id
ORDER BY account.username, repository.slug",
)?;
statement
.query_map([], repository_from_row)?
.collect::<Result<Vec<_>, _>>()
.map_err(Into::into)
}
pub(crate) fn index_names(&self) -> Result<Vec<String>, StoreError> {
let mut statement = self.connection.prepare(
"SELECT name FROM sqlite_schema
WHERE type = 'index' AND name NOT LIKE 'sqlite_%'
ORDER BY name",
)?;
statement
.query_map([], |row| row.get(0))?
.collect::<Result<Vec<_>, _>>()
.map_err(Into::into)
}
pub(crate) fn inspect_account(&self, username: &str) -> Result<AccountInspection, StoreError> {
let account = self
.connection
.query_row(
"SELECT id, username, is_administrator, state, created_at
FROM account WHERE username = ?1",
[username],
|row| {
Ok((
row.get::<_, i64>(0)?,
row.get::<_, String>(1)?,
row.get::<_, bool>(2)?,
row.get::<_, String>(3)?,
row.get::<_, i64>(4)?,
))
},
)
.optional()?
.ok_or_else(|| StoreError::AccountNotFound(username.to_owned()))?;
let mut statement = self.connection.prepare(
"SELECT label, fingerprint, created_at, last_used_at, revoked_at
FROM ssh_public_key WHERE account_id = ?1 ORDER BY id",
)?;
let keys = statement
.query_map([account.0], |row| {
Ok(KeyInspection {
label: row.get(0)?,
fingerprint: row.get(1)?,
created_at: row.get(2)?,
last_used_at: row.get(3)?,
revoked_at: row.get(4)?,
})
})?
.collect::<Result<Vec<_>, _>>()?;
Ok(AccountInspection {
record_type: "account",
username: account.1,
administrator: account.2,
state: account.3,
created_at: account.4,
keys,
})
}
pub(crate) fn inspect_git_intent(&self, id: &str) -> Result<GitIntentInspection, StoreError> {
self.connection
.query_row(
"SELECT id, repository_path, actor, quarantine_path, state, pack_name, created_at
FROM git_operation_intent WHERE id = ?1",
[id],
|row| {
Ok(GitIntentInspection {
record_type: "git-intent",
id: row.get(0)?,
repository_path: row.get(1)?,
actor: row.get(2)?,
quarantine_path: row.get(3)?,
state: row.get(4)?,
pack_name: row.get(5)?,
created_at: row.get(6)?,
})
},
)
.optional()?
.ok_or_else(|| StoreError::Integrity(format!("Git intent does not exist: {id}")))
}
pub(crate) fn dump_rows(
&self,
mut visit: impl FnMut(DumpRow) -> bool,
) -> Result<(), StoreError> {
let mut tables_statement = self.connection.prepare(
"SELECT name FROM sqlite_schema
WHERE type = 'table' AND name NOT LIKE 'sqlite_%'
ORDER BY name",
)?;
let tables = tables_statement
.query_map([], |row| row.get::<_, String>(0))?
.collect::<Result<Vec<_>, _>>()?;
for table in tables {
let columns = self.table_columns(&table)?;
let order = columns
.iter()
.filter(|column| column.primary_key > 0)
.map(|column| (column.primary_key, quote_identifier(&column.name)))
.collect::<Vec<_>>();
let order = if order.is_empty() {
"rowid".to_owned()
} else {
let mut order = order;
order.sort_by_key(|item| item.0);
order
.into_iter()
.map(|item| item.1)
.collect::<Vec<_>>()
.join(", ")
};
let sql = format!(
"SELECT * FROM {} ORDER BY {order}",
quote_identifier(&table)
);
let mut statement = self.connection.prepare(&sql)?;
let mut rows = statement.query([])?;
while let Some(row) = rows.next()? {
let mut values = Vec::with_capacity(columns.len());
for (index, column) in columns.iter().enumerate() {
values.push(DumpColumn {
name: column.name.clone(),
value: dump_value(row.get_ref(index)?),
});
}
let output = DumpRow {
record_type: "sqlite-row",
table: table.clone(),
columns: values,
};
if !visit(output) {
return Ok(());
}
}
}
Ok(())
}
fn table_columns(&self, table: &str) -> Result<Vec<TableColumn>, StoreError> {
let mut statement = self.connection.prepare(
"SELECT name, pk FROM pragma_table_info(?1)
ORDER BY cid",
)?;
statement
.query_map([table], |row| {
Ok(TableColumn {
name: row.get(0)?,
primary_key: row.get(1)?,
})
})?
.collect::<Result<Vec<_>, _>>()
.map_err(Into::into)
}
#[allow(
dead_code,
reason = "some integration tests import the store without public HTTP routes"
)]
pub(crate) fn public_repository(
&self,
owner: &str,
slug: &str,
) -> Result<RepositoryRecord, StoreError> {
let result = self.connection.query_row(
"SELECT repository.id, account.username, repository.slug,
repository.visibility, repository.state, repository.object_format,
repository.created_at, repository.archived_at
FROM repository
JOIN account ON account.id = repository.owner_account_id
WHERE account.username = ?1 AND repository.slug = ?2
AND repository.visibility = 'public' AND repository.state = 'active'",
rusqlite::params![owner, slug],
repository_from_row,
);
match result {
Ok(repository) => Ok(repository),
Err(rusqlite::Error::QueryReturnedNoRows) => Err(StoreError::RepositoryNotFound(
owner.to_owned(),
slug.to_owned(),
)),
Err(error) => Err(error.into()),
}
}
#[allow(
dead_code,
reason = "some integration tests import storage without the server"
)]
pub(crate) fn active_ssh_public_keys(&self) -> Result<Vec<String>, StoreError> {
let mut statement = self.connection.prepare(
"SELECT ssh_public_key.canonical_key
FROM ssh_public_key
JOIN account ON account.id = ssh_public_key.account_id
WHERE account.state = 'active' AND ssh_public_key.revoked_at IS NULL
ORDER BY ssh_public_key.id",
)?;
statement
.query_map([], |row| row.get(0))?
.collect::<Result<Vec<_>, _>>()
.map_err(Into::into)
}
#[allow(
dead_code,
reason = "some integration tests compile storage without the production SSH server"
)]
pub(crate) fn active_ssh_identities(&self) -> Result<Vec<ActiveSshIdentity>, StoreError> {
let mut statement = self.connection.prepare(
"SELECT account.username, ssh_public_key.canonical_key,
ssh_public_key.fingerprint
FROM ssh_public_key
JOIN account ON account.id = ssh_public_key.account_id
WHERE account.state = 'active' AND ssh_public_key.revoked_at IS NULL
ORDER BY ssh_public_key.id",
)?;
statement
.query_map([], |row| {
Ok(ActiveSshIdentity {
username: row.get(0)?,
canonical_key: row.get(1)?,
fingerprint: row.get(2)?,
})
})?
.collect::<Result<Vec<_>, _>>()
.map_err(Into::into)
}
#[allow(
dead_code,
reason = "some integration tests import storage without the server"
)]
pub(crate) fn active_public_repositories(&self) -> Result<Vec<RepositoryRecord>, StoreError> {
let mut statement = self.connection.prepare(
"SELECT repository.id, account.username, repository.slug,
repository.visibility, repository.state, repository.object_format,
repository.created_at, repository.archived_at
FROM repository
JOIN account ON account.id = repository.owner_account_id
WHERE repository.visibility = 'public' AND repository.state = 'active'
ORDER BY account.username, repository.slug",
)?;
statement
.query_map([], repository_from_row)?
.collect::<Result<Vec<_>, _>>()
.map_err(Into::into)
}
pub(crate) fn home_repositories(
&self,
owner: Option<&str>,
limit: usize,
) -> Result<Vec<HomeRepositoryRecord>, StoreError> {
let limit = i64::try_from(limit).expect("the home repository limit fits in SQLite");
let mut statement = self.connection.prepare(
"SELECT account.username, repository.slug, repository.visibility, repository.state,
COALESCE(repository_profile.description, ''),
COALESCE(MAX(repository_event.created_at), repository.created_at)
FROM repository
JOIN account ON account.id = repository.owner_account_id
LEFT JOIN repository_profile ON repository_profile.repository_id = repository.id
LEFT JOIN repository_event ON repository_event.repository_id = repository.id
WHERE ((?1 IS NULL AND repository.visibility = 'public'
AND repository.state = 'active')
OR account.username = ?1)
GROUP BY repository.id, account.username, repository.slug,
repository.visibility, repository.state, repository_profile.description,
repository.created_at
ORDER BY 6 DESC, account.username, repository.slug
LIMIT ?2",
)?;
statement
.query_map(rusqlite::params![owner, limit], |row| {
Ok(HomeRepositoryRecord {
owner: row.get(0)?,
slug: row.get(1)?,
visibility: row.get(2)?,
state: row.get(3)?,
description: row.get(4)?,
updated_at: row.get(5)?,
})
})?
.collect::<Result<Vec<_>, _>>()
.map_err(Into::into)
}
pub(crate) fn public_profile(&self, username: &str) -> Result<PublicProfile, StoreError> {
let profile = self.connection.query_row(
"SELECT username, bio, contact_email
FROM account
WHERE username = ?1 AND state = 'active'",
[username],
|row| {
Ok(PublicProfile {
username: row.get(0)?,
bio: row.get(1)?,
contact_email: row.get(2)?,
repositories: Vec::new(),
page: 1,
has_next: false,
})
},
);
let mut profile = match profile {
Ok(profile) => profile,
Err(rusqlite::Error::QueryReturnedNoRows) => {
return Err(StoreError::AccountNotFound(username.to_owned()));
}
Err(error) => return Err(error.into()),
};
let mut statement = self.connection.prepare(
"SELECT account.username, repository.slug, repository.visibility, repository.state,
COALESCE(repository_profile.description, ''),
COALESCE(MAX(repository_event.created_at), repository.created_at)
FROM repository
JOIN account ON account.id = repository.owner_account_id
LEFT JOIN repository_profile ON repository_profile.repository_id = repository.id
LEFT JOIN repository_event ON repository_event.repository_id = repository.id
WHERE account.username = ?1
AND repository.state = 'active'
AND repository.visibility = 'public'
GROUP BY repository.id, account.username, repository.slug,
repository.visibility, repository.state, repository_profile.description,
repository.created_at
ORDER BY 6 DESC, repository.slug
LIMIT 100",
)?;
profile.repositories = statement
.query_map([username], |row| {
Ok(HomeRepositoryRecord {
owner: row.get(0)?,
slug: row.get(1)?,
visibility: row.get(2)?,
state: row.get(3)?,
description: row.get(4)?,
updated_at: row.get(5)?,
})
})?
.collect::<Result<Vec<_>, _>>()?;
Ok(profile)
}
pub(crate) fn public_profile_page(
&self,
username: &str,
page: usize,
page_size: usize,
) -> Result<PublicProfile, StoreError> {
let mut profile = self.public_profile(username)?;
let offset = page
.checked_sub(1)
.and_then(|page| page.checked_mul(page_size))
.ok_or(StoreError::EventLimit)?;
let limit = i64::try_from(page_size.checked_add(1).ok_or(StoreError::EventLimit)?)
.map_err(|_| StoreError::EventLimit)?;
let offset = i64::try_from(offset).map_err(|_| StoreError::EventLimit)?;
let mut statement = self.connection.prepare(
"SELECT account.username, repository.slug, repository.visibility, repository.state,
COALESCE(repository_profile.description, ''),
COALESCE(MAX(repository_event.created_at), repository.created_at)
FROM repository
JOIN account ON account.id = repository.owner_account_id
LEFT JOIN repository_profile ON repository_profile.repository_id = repository.id
LEFT JOIN repository_event ON repository_event.repository_id = repository.id
WHERE account.username = ?1
AND repository.state = 'active'
AND repository.visibility = 'public'
GROUP BY repository.id, account.username, repository.slug,
repository.visibility, repository.state, repository_profile.description,
repository.created_at
ORDER BY 6 DESC, repository.slug
LIMIT ?2 OFFSET ?3",
)?;
let mut repositories = statement
.query_map(rusqlite::params![username, limit, offset], |row| {
Ok(HomeRepositoryRecord {
owner: row.get(0)?,
slug: row.get(1)?,
visibility: row.get(2)?,
state: row.get(3)?,
description: row.get(4)?,
updated_at: row.get(5)?,
})
})?
.collect::<Result<Vec<_>, _>>()?;
profile.has_next = repositories.len() > page_size;
repositories.truncate(page_size);
profile.repositories = repositories;
profile.page = page;
Ok(profile)
}
pub(crate) fn update_profile(
&mut self,
username: &str,
bio: &str,
contact_email: &str,
) -> Result<(), StoreError> {
let changed = self.connection.execute(
"UPDATE account SET bio = ?2, contact_email = ?3
WHERE username = ?1 AND state = 'active'",
rusqlite::params![username, bio, contact_email],
)?;
if changed == 1 {
Ok(())
} else {
Err(StoreError::AccountNotFound(username.to_owned()))
}
}
#[allow(
dead_code,
reason = "some integration tests import storage without public event pages"
)]
pub(crate) fn public_repository_events(
&self,
owner: &str,
slug: &str,
before: Option<i64>,
limit: usize,
) -> Result<(RepositoryRecord, Vec<RepositoryEventRecord>), StoreError> {
let repository = self.public_repository(owner, slug)?;
self.repository_events_for(repository, before, limit, false)
}
#[allow(
dead_code,
reason = "some integration tests use only public event queries"
)]
pub(crate) fn repository_events(
&self,
owner: &str,
slug: &str,
before: Option<i64>,
limit: usize,
) -> Result<(RepositoryRecord, Vec<RepositoryEventRecord>), StoreError> {
let repository = self.repository(owner, slug)?;
self.repository_events_for(repository, before, limit, false)
}
pub(crate) fn repository_issue_events(
&self,
owner: &str,
slug: &str,
before: Option<i64>,
limit: usize,
) -> Result<(RepositoryRecord, Vec<RepositoryEventRecord>), StoreError> {
let repository = self.repository(owner, slug)?;
self.repository_events_for(repository, before, limit, true)
}
fn repository_events_for(
&self,
repository: RepositoryRecord,
before: Option<i64>,
limit: usize,
issues_only: bool,
) -> Result<(RepositoryRecord, Vec<RepositoryEventRecord>), StoreError> {
let limit = i64::try_from(limit).map_err(|_| StoreError::EventLimit)?;
let mut statement = self.connection.prepare(
"SELECT event_id, sequence, kind, actor, ref_name, old_target, new_target,
payload_version, payload, created_at
FROM repository_event
WHERE repository_id = ?1 AND (?2 IS NULL OR sequence < ?2)
AND (?4 = 0 OR kind LIKE 'issue-%')
ORDER BY sequence DESC
LIMIT ?3",
)?;
let events = statement
.query_map(
rusqlite::params![repository.id, before, limit, issues_only],
|row| {
Ok(RepositoryEventRecord {
event_id: row.get(0)?,
sequence: row.get(1)?,
kind: row.get(2)?,
actor: row.get(3)?,
ref_name: row.get(4)?,
old_target: row.get(5)?,
new_target: row.get(6)?,
payload_version: row.get(7)?,
payload: row.get(8)?,
created_at: row.get(9)?,
})
},
)?
.collect::<Result<Vec<_>, _>>()?;
Ok((repository, events))
}
}
pub(crate) struct InitialAdministrator<'a> {
pub(crate) username: &'a str,
pub(crate) canonical_key: &'a str,
pub(crate) fingerprint: &'a str,
pub(crate) recovery_hash: &'a [u8; 32],
pub(crate) created_at: i64,
}
#[allow(
dead_code,
reason = "some integration tests compile storage without accounts"
)]
pub(crate) struct NewSshKey<'a> {
pub(crate) canonical_key: &'a str,
pub(crate) fingerprint: &'a str,
pub(crate) label: &'a str,
}
#[allow(
dead_code,
reason = "some integration tests compile storage without accounts"
)]
pub(crate) struct InvitedAccount<'a> {
pub(crate) invitation_hash: &'a [u8; 32],
pub(crate) username: &'a str,
pub(crate) key: NewSshKey<'a>,
pub(crate) recovery_hash: &'a [u8; 32],
pub(crate) created_at: i64,
pub(crate) correlation_id: &'a str,
}
#[allow(
dead_code,
reason = "some integration tests compile storage without accounts"
)]
pub(crate) struct AccountRecovery<'a> {
pub(crate) username: &'a str,
pub(crate) old_recovery_hash: &'a [u8; 32],
pub(crate) key: NewSshKey<'a>,
pub(crate) new_recovery_hash: &'a [u8; 32],
pub(crate) created_at: i64,
pub(crate) correlation_id: &'a str,
}
#[allow(
dead_code,
reason = "some integration tests compile storage without Web login"
)]
pub(crate) struct NewLoginNonce<'a> {
pub(crate) nonce_hash: &'a [u8; 32],
pub(crate) csrf_hash: &'a [u8; 32],
pub(crate) username: &'a str,
pub(crate) created_at: i64,
pub(crate) expires_at: i64,
}
pub(crate) struct NewLoginApproval<'a> {
pub(crate) secret_hash: &'a [u8; 32],
pub(crate) csrf_hash: &'a [u8; 32],
pub(crate) purpose: &'a str,
pub(crate) expected_username: Option<&'a str>,
pub(crate) created_at: i64,
pub(crate) expires_at: i64,
}
pub(crate) struct AccountKeyAuthorization<'a> {
pub(crate) username: &'a str,
pub(crate) session_hash: &'a [u8; 32],
pub(crate) csrf_hash: &'a [u8; 32],
pub(crate) secret_hash: &'a [u8; 32],
pub(crate) changed_at: i64,
pub(crate) correlation_id: &'a str,
}
pub(crate) struct ApproveLogin<'a> {
pub(crate) secret_hash: &'a [u8; 32],
pub(crate) username: &'a str,
pub(crate) fingerprint: &'a str,
pub(crate) approved_at: i64,
}
pub(crate) struct NewApprovedWebSession<'a> {
pub(crate) secret_hash: &'a [u8; 32],
pub(crate) login_csrf_hash: &'a [u8; 32],
pub(crate) session_hash: &'a [u8; 32],
pub(crate) csrf_hash: &'a [u8; 32],
pub(crate) created_at: i64,
pub(crate) expires_at: i64,
pub(crate) correlation_id: &'a str,
}
#[allow(
dead_code,
reason = "some integration tests compile storage without Web login"
)]
pub(crate) struct NewWebSession<'a> {
pub(crate) nonce_hash: &'a [u8; 32],
pub(crate) login_csrf_hash: &'a [u8; 32],
pub(crate) username: &'a str,
pub(crate) fingerprint: &'a str,
pub(crate) session_hash: &'a [u8; 32],
pub(crate) csrf_hash: &'a [u8; 32],
pub(crate) created_at: i64,
pub(crate) expires_at: i64,
pub(crate) correlation_id: &'a str,
}
#[allow(
dead_code,
reason = "some integration tests compile storage without Web login"
)]
pub(crate) struct WebSessionRecord {
pub(crate) username: String,
pub(crate) is_administrator: bool,
pub(crate) expires_at: i64,
pub(crate) ssh_public_key_id: Option<i64>,
}
pub(crate) struct NewRepository<'a> {
pub(crate) id: &'a str,
pub(crate) owner: &'a str,
pub(crate) slug: &'a str,
pub(crate) object_format: &'a str,
pub(crate) default_branch: &'a str,
pub(crate) created_at: i64,
pub(crate) origin: RepositoryOrigin,
pub(crate) initial_references: &'a [NewRepositoryReference],
pub(crate) actor: &'a str,
pub(crate) correlation_id: &'a str,
}
pub(crate) struct MaintenanceResult {
pub(crate) deleted: usize,
}
pub(crate) struct RecordPage<T> {
pub(crate) items: Vec<T>,
pub(crate) page: usize,
pub(crate) has_next: bool,
}
#[derive(Clone, Copy)]
#[allow(
dead_code,
reason = "some integration tests create repositories without the import operation"
)]
pub(crate) enum RepositoryOrigin {
Created,
Imported,
}
impl RepositoryOrigin {
fn event_kind(self) -> event::EventKind {
match self {
Self::Created => event::EventKind::RepositoryCreated,
Self::Imported => event::EventKind::RepositoryImported,
}
}
fn audit_action(self) -> &'static str {
match self {
Self::Created => "repository.create",
Self::Imported => "repository.import",
}
}
}
pub(crate) struct NewRepositoryReference {
pub(crate) name: Vec<u8>,
pub(crate) target: String,
}
#[derive(Debug, Eq, PartialEq)]
pub(crate) struct RepositoryRecord {
pub(crate) id: String,
pub(crate) owner: String,
pub(crate) slug: String,
pub(crate) visibility: String,
pub(crate) state: String,
pub(crate) object_format: String,
pub(crate) created_at: i64,
pub(crate) archived_at: Option<i64>,
}
#[derive(Debug, Eq, PartialEq)]
pub(crate) struct HomeRepositoryRecord {
pub(crate) owner: String,
pub(crate) slug: String,
pub(crate) visibility: String,
pub(crate) state: String,
pub(crate) description: String,
pub(crate) updated_at: i64,
}
pub(crate) struct RepositorySettings {
pub(crate) repository: RepositoryRecord,
pub(crate) description: String,
pub(crate) collaborators: Vec<RepositoryCollaboratorRecord>,
pub(crate) default_branch: String,
pub(crate) branches: Vec<String>,
}
pub(crate) struct RepositoryCollaboratorRecord {
pub(crate) username: String,
pub(crate) role: String,
}
#[derive(Debug, Eq, PartialEq)]
pub(crate) struct PublicProfile {
pub(crate) username: String,
pub(crate) bio: String,
pub(crate) contact_email: String,
pub(crate) repositories: Vec<HomeRepositoryRecord>,
pub(crate) page: usize,
pub(crate) has_next: bool,
}
#[derive(Debug, Serialize)]
pub(crate) struct AccountInspection {
#[serde(rename = "type")]
pub(crate) record_type: &'static str,
pub(crate) username: String,
pub(crate) administrator: bool,
pub(crate) state: String,
pub(crate) created_at: i64,
pub(crate) keys: Vec<KeyInspection>,
}
#[derive(Debug, Serialize)]
pub(crate) struct KeyInspection {
pub(crate) label: String,
pub(crate) fingerprint: String,
pub(crate) created_at: i64,
pub(crate) last_used_at: Option<i64>,
pub(crate) revoked_at: Option<i64>,
}
#[derive(Debug, Serialize)]
pub(crate) struct GitIntentInspection {
#[serde(rename = "type")]
pub(crate) record_type: &'static str,
pub(crate) id: String,
pub(crate) repository_path: String,
pub(crate) actor: String,
pub(crate) quarantine_path: String,
pub(crate) state: String,
pub(crate) pack_name: Option<String>,
pub(crate) created_at: i64,
}
#[derive(Debug, Serialize)]
pub(crate) struct RepositoryInspection {
#[serde(rename = "type")]
pub(crate) record_type: &'static str,
pub(crate) id: String,
pub(crate) owner: String,
pub(crate) slug: String,
pub(crate) visibility: String,
pub(crate) state: String,
pub(crate) object_format: String,
pub(crate) created_at: i64,
pub(crate) archived_at: Option<i64>,
}
impl From<RepositoryRecord> for RepositoryInspection {
fn from(repository: RepositoryRecord) -> Self {
Self {
record_type: "repository",
id: repository.id,
owner: repository.owner,
slug: repository.slug,
visibility: repository.visibility,
state: repository.state,
object_format: repository.object_format,
created_at: repository.created_at,
archived_at: repository.archived_at,
}
}
}
#[derive(Debug, Serialize)]
pub(crate) struct DumpRow {
#[serde(rename = "type")]
pub(crate) record_type: &'static str,
pub(crate) table: String,
pub(crate) columns: Vec<DumpColumn>,
}
#[derive(Debug, Serialize)]
pub(crate) struct DumpColumn {
pub(crate) name: String,
pub(crate) value: DumpValue,
}
#[derive(Debug, Serialize)]
#[serde(tag = "type", content = "value", rename_all = "kebab-case")]
pub(crate) enum DumpValue {
Null,
Integer(i64),
Real(String),
TextUtf8(String),
TextHex(String),
BlobHex(String),
}
struct TableColumn {
name: String,
primary_key: i64,
}
#[allow(
dead_code,
reason = "some integration tests compile storage without authorization"
)]
pub(crate) struct RepositoryAuthorizationRecord {
pub(crate) repository: RepositoryRecord,
pub(crate) role: Option<String>,
}
#[allow(
dead_code,
reason = "some integration tests compile storage without the production SSH server"
)]
pub(crate) struct ActiveSshIdentity {
pub(crate) username: String,
pub(crate) canonical_key: String,
pub(crate) fingerprint: String,
}
#[allow(
dead_code,
reason = "some integration tests import storage without public event pages"
)]
pub(crate) struct RepositoryEventRecord {
pub(crate) event_id: String,
pub(crate) sequence: i64,
pub(crate) kind: String,
pub(crate) actor: String,
pub(crate) ref_name: Option<Vec<u8>>,
pub(crate) old_target: Option<String>,
pub(crate) new_target: Option<String>,
pub(crate) payload_version: i64,
pub(crate) payload: String,
pub(crate) created_at: i64,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub(crate) struct IssueRecord {
pub(crate) id: String,
pub(crate) number: i64,
pub(crate) title: String,
pub(crate) body: String,
pub(crate) state: String,
pub(crate) author: String,
pub(crate) created_at: i64,
pub(crate) updated_at: i64,
pub(crate) closed_at: Option<i64>,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub(crate) struct IssueCommentRecord {
pub(crate) id: String,
pub(crate) author: String,
pub(crate) body: String,
pub(crate) created_at: i64,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub(crate) struct IssueTimelineRecord {
pub(crate) sequence: i64,
pub(crate) kind: String,
pub(crate) actor: String,
pub(crate) payload: String,
pub(crate) created_at: i64,
}
pub(crate) struct IssueDetail {
pub(crate) repository: RepositoryRecord,
pub(crate) issue: IssueRecord,
pub(crate) comments: Vec<IssueCommentRecord>,
pub(crate) timeline: Vec<IssueTimelineRecord>,
pub(crate) can_comment: bool,
pub(crate) can_edit: bool,
pub(crate) comments_page: usize,
pub(crate) comments_has_next: bool,
pub(crate) timeline_page: usize,
pub(crate) timeline_has_next: bool,
}
pub(crate) struct NewIssue<'a> {
pub(crate) owner: &'a str,
pub(crate) repository: &'a str,
pub(crate) actor: &'a str,
pub(crate) title: &'a str,
pub(crate) body: &'a str,
pub(crate) created_at: i64,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub(crate) struct PullRequestRecord {
pub(crate) id: String,
pub(crate) number: i64,
pub(crate) title: String,
pub(crate) body: String,
pub(crate) state: String,
pub(crate) author: String,
pub(crate) base_ref: String,
pub(crate) head_ref: String,
pub(crate) base_object_id: String,
pub(crate) head_object_id: String,
pub(crate) created_at: i64,
pub(crate) updated_at: i64,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub(crate) struct PullRequestRevisionRecord {
pub(crate) id: String,
pub(crate) number: i64,
pub(crate) author: String,
pub(crate) base_object_id: String,
pub(crate) head_object_id: String,
pub(crate) created_at: i64,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub(crate) struct PullRequestReviewRecord {
pub(crate) id: String,
pub(crate) revision: i64,
pub(crate) author: String,
pub(crate) kind: String,
pub(crate) body: String,
pub(crate) commit_object_id: Option<String>,
pub(crate) path: Option<Vec<u8>>,
pub(crate) side: Option<String>,
pub(crate) line: Option<i64>,
pub(crate) created_at: i64,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub(crate) struct PullRequestTimelineRecord {
pub(crate) sequence: i64,
pub(crate) kind: String,
pub(crate) actor: String,
pub(crate) payload: String,
pub(crate) created_at: i64,
}
pub(crate) struct PullRequestDetail {
pub(crate) repository: RepositoryRecord,
pub(crate) pull_request: PullRequestRecord,
pub(crate) revisions: Vec<PullRequestRevisionRecord>,
pub(crate) reviews: Vec<PullRequestReviewRecord>,
pub(crate) timeline: Vec<PullRequestTimelineRecord>,
pub(crate) reviews_page: usize,
pub(crate) reviews_has_next: bool,
pub(crate) timeline_page: usize,
pub(crate) timeline_has_next: bool,
pub(crate) can_edit: bool,
pub(crate) can_change_state: bool,
pub(crate) can_revise: bool,
pub(crate) can_review: bool,
pub(crate) can_merge: bool,
}
pub(crate) struct NewPullRequestReview<'a> {
pub(crate) owner: &'a str,
pub(crate) repository: &'a str,
pub(crate) number: i64,
pub(crate) revision: i64,
pub(crate) actor: &'a str,
pub(crate) kind: &'a str,
pub(crate) body: &'a str,
pub(crate) commit_object_id: Option<&'a str>,
pub(crate) path: Option<&'a [u8]>,
pub(crate) side: Option<&'a str>,
pub(crate) line: Option<i64>,
pub(crate) created_at: i64,
}
pub(crate) struct NewPullRequestRefIntent<'a> {
pub(crate) id: &'a str,
pub(crate) pull_request_id: &'a str,
pub(crate) owner: &'a str,
pub(crate) repository: &'a str,
pub(crate) actor: &'a str,
pub(crate) title: &'a str,
pub(crate) body: &'a str,
pub(crate) base_ref: &'a str,
pub(crate) head_ref: &'a str,
pub(crate) base_object_id: &'a str,
pub(crate) head_object_id: &'a str,
pub(crate) created_at: i64,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub(crate) struct PullRequestRefIntentRecord {
pub(crate) id: String,
pub(crate) repository_id: String,
pub(crate) pull_request_id: String,
pub(crate) pull_request_number: i64,
pub(crate) revision_number: i64,
pub(crate) operation: String,
pub(crate) title: String,
pub(crate) body: String,
pub(crate) author_account_id: i64,
pub(crate) actor: String,
pub(crate) base_ref: String,
pub(crate) head_ref: String,
pub(crate) base_object_id: String,
pub(crate) old_head_object_id: Option<String>,
pub(crate) head_object_id: String,
pub(crate) state: String,
pub(crate) created_at: i64,
}
impl PullRequestRefIntentRecord {
#[allow(clippy::too_many_arguments)]
fn from_new(
intent: &NewPullRequestRefIntent<'_>,
repository_id: String,
author_account_id: i64,
pull_request_number: i64,
revision_number: i64,
old_head_object_id: Option<String>,
operation: &str,
) -> Self {
Self {
id: intent.id.to_owned(),
repository_id,
pull_request_id: intent.pull_request_id.to_owned(),
pull_request_number,
revision_number,
operation: operation.to_owned(),
title: intent.title.to_owned(),
body: intent.body.to_owned(),
author_account_id,
actor: intent.actor.to_owned(),
base_ref: intent.base_ref.to_owned(),
head_ref: intent.head_ref.to_owned(),
base_object_id: intent.base_object_id.to_owned(),
old_head_object_id,
head_object_id: intent.head_object_id.to_owned(),
state: "pending".to_owned(),
created_at: intent.created_at,
}
}
}
pub(crate) struct IssueChange<'a> {
pub(crate) owner: &'a str,
pub(crate) repository: &'a str,
pub(crate) number: i64,
pub(crate) actor: &'a str,
pub(crate) changed_at: i64,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub(crate) struct WatchRecord {
pub(crate) id: String,
pub(crate) created_at: i64,
pub(crate) updated_at: i64,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub(crate) struct FeedTokenRecord {
pub(crate) id: String,
pub(crate) scope: String,
pub(crate) created_at: i64,
pub(crate) revoked_at: Option<i64>,
}
pub(crate) struct ActivityEventRecord {
pub(crate) repository: RepositoryRecord,
pub(crate) event: RepositoryEventRecord,
}
pub(crate) struct ActivityCursor {
pub(crate) created_at: i64,
pub(crate) event_id: String,
}
pub(crate) struct ActivityPage {
pub(crate) events: Vec<ActivityEventRecord>,
pub(crate) next_before: Option<ActivityCursor>,
}
pub(crate) struct TokenFeedPage {
pub(crate) username: String,
pub(crate) events: Vec<ActivityEventRecord>,
}
pub(crate) struct MetadataSearchCandidate {
pub(crate) kind: String,
pub(crate) record_id: String,
pub(crate) owner: String,
pub(crate) repository: String,
pub(crate) title: Vec<u8>,
pub(crate) body: Vec<u8>,
}
pub(crate) struct GitOperationIntent<'a> {
pub(crate) id: &'a str,
pub(crate) repository_path: &'a str,
pub(crate) actor: &'a str,
pub(crate) initial_refs: &'a [u8],
pub(crate) proposed_refs: &'a [u8],
pub(crate) event_payload: &'a [u8],
pub(crate) quarantine_path: &'a str,
pub(crate) created_at: i64,
}
pub(crate) struct NewPullRequestMerge<'a> {
pub(crate) owner: &'a str,
pub(crate) repository: &'a str,
pub(crate) number: i64,
pub(crate) revision: i64,
pub(crate) actor: &'a str,
pub(crate) method: &'a str,
pub(crate) base_ref: &'a str,
pub(crate) old_target: &'a str,
pub(crate) head_target: &'a str,
pub(crate) new_target: &'a str,
pub(crate) created_at: i64,
}
pub(crate) struct NewAuditEvent<'a> {
pub(crate) action: &'a str,
pub(crate) actor: &'a str,
pub(crate) target: &'a str,
pub(crate) outcome: &'a str,
pub(crate) correlation_id: &'a str,
pub(crate) created_at: i64,
}
pub(crate) struct AuditContext<'a> {
pub(crate) actor: &'a str,
pub(crate) correlation_id: &'a str,
pub(crate) created_at: i64,
}
#[allow(
dead_code,
reason = "some integration tests compile storage without the audit CLI"
)]
pub(crate) struct AuditEventRecord {
pub(crate) id: i64,
pub(crate) action: String,
pub(crate) actor: String,
pub(crate) target: String,
pub(crate) outcome: String,
pub(crate) correlation_id: String,
pub(crate) created_at: i64,
}
pub(crate) struct GitIntentRecord {
pub(crate) id: String,
pub(crate) repository_path: String,
pub(crate) initial_refs: Vec<u8>,
pub(crate) proposed_refs: Vec<u8>,
pub(crate) quarantine_path: String,
pub(crate) state: String,
pub(crate) pack_name: Option<String>,
}
#[allow(
dead_code,
reason = "some integration tests compile storage without accounts"
)]
fn insert_ssh_key(
transaction: &rusqlite::Transaction<'_>,
account_id: i64,
key: &NewSshKey<'_>,
created_at: i64,
) -> Result<(), StoreError> {
match transaction.execute(
"INSERT INTO ssh_public_key
(account_id, canonical_key, fingerprint, created_at, label, last_used_at, revoked_at)
VALUES (?1, ?2, ?3, ?4, ?5, NULL, NULL)",
rusqlite::params![
account_id,
key.canonical_key,
key.fingerprint,
created_at,
key.label,
],
) {
Ok(1) => Ok(()),
Err(error) if is_unique_constraint(&error) => Err(StoreError::KeyExists),
Err(error) => Err(error.into()),
Ok(_) => unreachable!("an INSERT changes one row"),
}
}
fn consume_account_key_authorization(
transaction: &rusqlite::Transaction<'_>,
authorization: &AccountKeyAuthorization<'_>,
) -> Result<(i64, Option<i64>), StoreError> {
let session = transaction
.query_row(
"SELECT web_session.account_id, web_session.ssh_public_key_id
FROM web_session
JOIN account ON account.id = web_session.account_id
WHERE web_session.session_hash = ?1 AND web_session.csrf_hash = ?2
AND web_session.ended_at IS NULL AND web_session.expires_at >= ?3
AND account.username = ?4 AND account.state = 'active'",
rusqlite::params![
authorization.session_hash,
authorization.csrf_hash,
authorization.changed_at,
authorization.username,
],
|row| Ok((row.get::<_, i64>(0)?, row.get::<_, Option<i64>>(1)?)),
)
.optional()?
.ok_or(StoreError::InvalidSession)?;
let approval = transaction
.query_row(
"SELECT ssh_login_approval.account_id
FROM ssh_login_approval
JOIN ssh_public_key
ON ssh_public_key.id = ssh_login_approval.ssh_public_key_id
WHERE secret_hash = ?1 AND csrf_hash = ?2
AND purpose = 'account-key' AND expected_account_id = ?3
AND ssh_login_approval.account_id = ?3 AND approved_at IS NOT NULL
AND consumed_at IS NULL AND expires_at >= ?4
AND ssh_public_key.revoked_at IS NULL",
rusqlite::params![
authorization.secret_hash,
authorization.csrf_hash,
session.0,
authorization.changed_at,
],
|row| row.get::<_, i64>(0),
)
.optional()?
.ok_or(StoreError::InvalidLoginApproval)?;
if approval != session.0 {
return Err(StoreError::InvalidLoginApproval);
}
let changed = transaction.execute(
"UPDATE ssh_login_approval SET consumed_at = ?2
WHERE secret_hash = ?1 AND consumed_at IS NULL",
rusqlite::params![authorization.secret_hash, authorization.changed_at],
)?;
if changed != 1 {
return Err(StoreError::InvalidLoginApproval);
}
Ok(session)
}
fn insert_audit_event(
connection: &Connection,
event: &NewAuditEvent<'_>,
) -> Result<(), StoreError> {
connection.execute(
"INSERT INTO audit_event
(action, actor, target, outcome, correlation_id, created_at)
VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
rusqlite::params![
event.action,
event.actor,
event.target,
event.outcome,
event.correlation_id,
event.created_at,
],
)?;
Ok(())
}
struct RepositoryIssueAccess {
repository: RepositoryRecord,
active_actor_id: Option<i64>,
role: Option<String>,
}
impl RepositoryIssueAccess {
fn can_read(&self) -> bool {
self.repository.state == "active"
&& (self.repository.visibility == "public" || self.role.is_some())
}
fn can_write_issue(&self, author_account_id: i64) -> bool {
self.can_read()
&& (self.active_actor_id == Some(author_account_id)
|| matches!(
self.role.as_deref(),
Some("owner" | "maintainer" | "writer")
))
}
fn can_maintain(&self) -> bool {
self.can_read() && matches!(self.role.as_deref(), Some("owner" | "maintainer"))
}
fn can_write_repository(&self) -> bool {
self.can_read()
&& matches!(
self.role.as_deref(),
Some("owner" | "maintainer" | "writer")
)
}
}
struct StoredIssue {
issue: IssueRecord,
author_account_id: i64,
}
fn repository_issue_access(
connection: &Connection,
owner: &str,
repository: &str,
actor: Option<&str>,
) -> Result<RepositoryIssueAccess, StoreError> {
let result = connection.query_row(
"SELECT repository.id, owner.username, repository.slug,
repository.visibility, repository.state, repository.object_format,
repository.created_at, repository.archived_at,
CASE WHEN actor.state = 'active' THEN actor.id END,
CASE
WHEN actor.state != 'active' THEN NULL
WHEN actor.id = repository.owner_account_id THEN 'owner'
ELSE repository_collaborator.role
END
FROM repository
JOIN account AS owner ON owner.id = repository.owner_account_id
LEFT JOIN account AS actor ON actor.username = ?3
LEFT JOIN repository_collaborator
ON repository_collaborator.repository_id = repository.id
AND repository_collaborator.account_id = actor.id
WHERE owner.username = ?1 AND repository.slug = ?2",
rusqlite::params![owner, repository, actor],
|row| {
Ok(RepositoryIssueAccess {
repository: repository_from_row(row)?,
active_actor_id: row.get(8)?,
role: row.get(9)?,
})
},
);
match result {
Ok(access) => Ok(access),
Err(rusqlite::Error::QueryReturnedNoRows) => Err(StoreError::RepositoryNotFound(
owner.to_owned(),
repository.to_owned(),
)),
Err(error) => Err(error.into()),
}
}
#[allow(clippy::too_many_arguments)]
fn insert_pull_request_intent(
transaction: &rusqlite::Transaction<'_>,
intent: &NewPullRequestRefIntent<'_>,
repository_id: &str,
author_account_id: i64,
pull_request_number: i64,
revision_number: i64,
old_head_object_id: Option<&str>,
operation: &str,
) -> Result<(), StoreError> {
transaction.execute(
"INSERT INTO pull_request_ref_intent
(id, repository_id, pull_request_id, pull_request_number, revision_number,
operation, title, body, author_account_id, actor, base_ref, head_ref,
base_object_id, old_head_object_id, head_object_id, state, created_at)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12,
?13, ?14, ?15, 'pending', ?16)",
rusqlite::params![
intent.id,
repository_id,
intent.pull_request_id,
pull_request_number,
revision_number,
operation,
intent.title,
intent.body,
author_account_id,
intent.actor,
intent.base_ref,
intent.head_ref,
intent.base_object_id,
old_head_object_id,
intent.head_object_id,
intent.created_at,
],
)?;
Ok(())
}
fn pull_request_intent(
connection: &Connection,
id: &str,
) -> Result<PullRequestRefIntentRecord, StoreError> {
connection
.query_row(
"SELECT id, repository_id, pull_request_id, pull_request_number,
revision_number, operation, title, body, author_account_id, actor,
base_ref, head_ref, base_object_id, old_head_object_id,
head_object_id, state, created_at
FROM pull_request_ref_intent WHERE id = ?1",
[id],
pull_request_intent_from_row,
)
.optional()?
.ok_or_else(|| StoreError::PullRequestIntentState(id.to_owned()))
}
fn pull_request_intent_from_row(
row: &rusqlite::Row<'_>,
) -> rusqlite::Result<PullRequestRefIntentRecord> {
Ok(PullRequestRefIntentRecord {
id: row.get(0)?,
repository_id: row.get(1)?,
pull_request_id: row.get(2)?,
pull_request_number: row.get(3)?,
revision_number: row.get(4)?,
operation: row.get(5)?,
title: row.get(6)?,
body: row.get(7)?,
author_account_id: row.get(8)?,
actor: row.get(9)?,
base_ref: row.get(10)?,
head_ref: row.get(11)?,
base_object_id: row.get(12)?,
old_head_object_id: row.get(13)?,
head_object_id: row.get(14)?,
state: row.get(15)?,
created_at: row.get(16)?,
})
}
fn pull_request_from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<PullRequestRecord> {
Ok(PullRequestRecord {
id: row.get(0)?,
number: row.get(1)?,
title: row.get(2)?,
body: row.get(3)?,
state: row.get(4)?,
author: row.get(5)?,
base_ref: row.get(6)?,
head_ref: row.get(7)?,
base_object_id: row.get(8)?,
head_object_id: row.get(9)?,
created_at: row.get(10)?,
updated_at: row.get(11)?,
})
}
fn pull_request_review_from_row(
row: &rusqlite::Row<'_>,
) -> rusqlite::Result<PullRequestReviewRecord> {
Ok(PullRequestReviewRecord {
id: row.get(0)?,
revision: row.get(1)?,
author: row.get(2)?,
kind: row.get(3)?,
body: row.get(4)?,
commit_object_id: row.get(5)?,
path: row.get(6)?,
side: row.get(7)?,
line: row.get(8)?,
created_at: row.get(9)?,
})
}
fn feed_token_from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<FeedTokenRecord> {
Ok(FeedTokenRecord {
id: row.get(0)?,
scope: row.get(1)?,
created_at: row.get(2)?,
revoked_at: row.get(3)?,
})
}
fn activity_from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<ActivityEventRecord> {
Ok(ActivityEventRecord {
repository: RepositoryRecord {
id: row.get(0)?,
owner: row.get(1)?,
slug: row.get(2)?,
visibility: row.get(3)?,
state: row.get(4)?,
object_format: row.get(5)?,
created_at: row.get(6)?,
archived_at: row.get(7)?,
},
event: RepositoryEventRecord {
event_id: row.get(8)?,
sequence: row.get(9)?,
kind: row.get(10)?,
actor: row.get(11)?,
ref_name: row.get(12)?,
old_target: row.get(13)?,
new_target: row.get(14)?,
payload_version: row.get(15)?,
payload: row.get(16)?,
created_at: row.get(17)?,
},
})
}
const ACTIVITY_SELECT: &str = "SELECT repository.id, owner.username, repository.slug,
repository.visibility, repository.state, repository.object_format,
repository.created_at, repository.archived_at,
repository_event.event_id, repository_event.sequence,
repository_event.kind, repository_event.actor, repository_event.ref_name,
repository_event.old_target, repository_event.new_target,
repository_event.payload_version, repository_event.payload,
repository_event.created_at
FROM repository_event
JOIN repository ON repository.id = repository_event.repository_id
JOIN account AS owner ON owner.id = repository.owner_account_id";
fn watched_activity_events(
connection: &Connection,
account_id: i64,
before: Option<&ActivityCursor>,
limit: i64,
) -> Result<Vec<ActivityEventRecord>, StoreError> {
let visibility = "
JOIN watch
ON watch.repository_id = repository.id AND watch.account_id = ?1
WHERE repository.state = 'active'
AND (repository.visibility = 'public'
OR repository.owner_account_id = ?1
OR EXISTS (
SELECT 1 FROM repository_collaborator
WHERE repository_collaborator.repository_id = repository.id
AND repository_collaborator.account_id = ?1
))";
match before {
Some(before) => {
let sql = format!(
"{ACTIVITY_SELECT}{visibility}
AND (repository_event.created_at < ?2
OR (repository_event.created_at = ?2
AND repository_event.event_id < ?3))
ORDER BY repository_event.created_at DESC, repository_event.event_id DESC
LIMIT ?4"
);
let mut statement = connection.prepare(&sql)?;
statement
.query_map(
rusqlite::params![account_id, before.created_at, before.event_id, limit],
activity_from_row,
)?
.collect::<Result<Vec<_>, _>>()
.map_err(Into::into)
}
None => {
let sql = format!(
"{ACTIVITY_SELECT}{visibility}
ORDER BY repository_event.created_at DESC, repository_event.event_id DESC
LIMIT ?2"
);
let mut statement = connection.prepare(&sql)?;
statement
.query_map(rusqlite::params![account_id, limit], activity_from_row)?
.collect::<Result<Vec<_>, _>>()
.map_err(Into::into)
}
}
}
fn find_issue(
connection: &Connection,
repository: &RepositoryRecord,
owner: &str,
slug: &str,
number: i64,
) -> Result<StoredIssue, StoreError> {
let result = connection.query_row(
"SELECT issue.id, issue.number, issue.title, issue.body, issue.state,
account.username, issue.created_at, issue.updated_at, issue.closed_at,
issue.author_account_id
FROM issue
JOIN account ON account.id = issue.author_account_id
WHERE issue.repository_id = ?1 AND issue.number = ?2",
rusqlite::params![repository.id, number],
|row| {
Ok(StoredIssue {
issue: issue_from_row(row)?,
author_account_id: row.get(9)?,
})
},
);
match result {
Ok(issue) => Ok(issue),
Err(rusqlite::Error::QueryReturnedNoRows) => Err(StoreError::IssueNotFound(
owner.to_owned(),
slug.to_owned(),
number,
)),
Err(error) => Err(error.into()),
}
}
fn issue_mutation_context(
connection: &Connection,
owner: &str,
repository: &str,
number: i64,
actor: &str,
) -> Result<(RepositoryIssueAccess, StoredIssue), StoreError> {
let access = repository_issue_access(connection, owner, repository, Some(actor))?;
if !access.can_read() {
return Err(StoreError::IssueHidden);
}
if access.active_actor_id.is_none() {
return Err(StoreError::IssueDenied);
}
let issue = find_issue(connection, &access.repository, owner, repository, number)?;
Ok((access, issue))
}
fn issue_from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<IssueRecord> {
Ok(IssueRecord {
id: row.get(0)?,
number: row.get(1)?,
title: row.get(2)?,
body: row.get(3)?,
state: row.get(4)?,
author: row.get(5)?,
created_at: row.get(6)?,
updated_at: row.get(7)?,
closed_at: row.get(8)?,
})
}
fn insert_issue_event(
transaction: &rusqlite::Transaction<'_>,
repository_id: &str,
issue_id: &str,
actor: &str,
created_at: i64,
event: &event::VersionedEvent,
) -> Result<(), StoreError> {
insert_domain_event(
transaction,
&NewDomainEvent {
repository_id,
source_intent_id: None,
source_ordinal: None,
issue_id: Some(issue_id),
pull_request_id: None,
event,
actor,
ref_name: None,
old_target: None,
new_target: None,
created_at,
},
)
}
fn insert_pull_request_event(
transaction: &rusqlite::Transaction<'_>,
repository_id: &str,
pull_request_id: &str,
actor: &str,
created_at: i64,
event: &event::VersionedEvent,
) -> Result<(), StoreError> {
insert_domain_event(
transaction,
&NewDomainEvent {
repository_id,
source_intent_id: None,
source_ordinal: None,
issue_id: None,
pull_request_id: Some(pull_request_id),
event,
actor,
ref_name: None,
old_target: None,
new_target: None,
created_at,
},
)
}
struct NewDomainEvent<'a> {
repository_id: &'a str,
source_intent_id: Option<&'a str>,
source_ordinal: Option<i64>,
issue_id: Option<&'a str>,
pull_request_id: Option<&'a str>,
event: &'a event::VersionedEvent,
actor: &'a str,
ref_name: Option<&'a [u8]>,
old_target: Option<&'a str>,
new_target: Option<&'a str>,
created_at: i64,
}
fn insert_domain_event(
transaction: &rusqlite::Transaction<'_>,
event: &NewDomainEvent<'_>,
) -> Result<(), StoreError> {
transaction.execute(
"INSERT INTO repository_event
(event_id, repository_id, sequence, source_intent_id, source_ordinal,
issue_id, pull_request_id, kind, actor, ref_name, old_target, new_target,
payload_version, payload,
created_at)
VALUES (
lower(hex(randomblob(16))),
?1,
(SELECT COALESCE(MAX(sequence), 0) + 1
FROM repository_event WHERE repository_id = ?1),
?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13
)",
rusqlite::params![
event.repository_id,
event.source_intent_id,
event.source_ordinal,
event.issue_id,
event.pull_request_id,
event.event.kind.as_str(),
event.actor,
event.ref_name,
event.old_target,
event.new_target,
event::PAYLOAD_VERSION,
event.event.payload,
event.created_at,
],
)?;
Ok(())
}
fn complete_pull_request_merge(
transaction: &rusqlite::Transaction<'_>,
intent_id: &str,
actor: &str,
completed_at: i64,
) -> Result<(), StoreError> {
let merge = transaction
.query_row(
"SELECT merge.repository_id, merge.pull_request_id, pull_request.number,
merge.revision_id, merge.revision_number, merge.method,
merge.base_ref, merge.old_target, merge.new_target,
pull_request.state, pull_request.base_ref,
pull_request.base_object_id, pull_request.head_object_id,
revision.base_object_id, revision.head_object_id,
(SELECT MAX(number) FROM pull_request_revision AS latest
WHERE latest.pull_request_id = pull_request.id)
FROM pull_request_merge_intent AS merge
JOIN pull_request ON pull_request.id = merge.pull_request_id
JOIN pull_request_revision AS revision ON revision.id = merge.revision_id
WHERE merge.intent_id = ?1",
[intent_id],
|row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, i64>(2)?,
row.get::<_, String>(3)?,
row.get::<_, i64>(4)?,
row.get::<_, String>(5)?,
row.get::<_, String>(6)?,
row.get::<_, String>(7)?,
row.get::<_, String>(8)?,
row.get::<_, String>(9)?,
row.get::<_, String>(10)?,
row.get::<_, String>(11)?,
row.get::<_, String>(12)?,
row.get::<_, String>(13)?,
row.get::<_, String>(14)?,
row.get::<_, i64>(15)?,
))
},
)
.optional()?;
let Some(merge) = merge else {
return Ok(());
};
if merge.9 != "open"
|| merge.10 != merge.6
|| merge.11 != merge.7
|| merge.12 != merge.14
|| merge.13 != merge.7
|| merge.15 != merge.4
{
return Err(StoreError::PullRequestState);
}
let changed = transaction.execute(
"UPDATE pull_request SET state = 'merged', updated_at = ?2
WHERE id = ?1 AND state = 'open'",
rusqlite::params![merge.1, completed_at],
)?;
if changed != 1 {
return Err(StoreError::PullRequestState);
}
let event = event::pull_request_merge(
&merge.1, merge.2, merge.4, &merge.5, &merge.6, &merge.7, &merge.8,
);
insert_domain_event(
transaction,
&NewDomainEvent {
repository_id: &merge.0,
source_intent_id: Some(intent_id),
source_ordinal: Some(2),
issue_id: None,
pull_request_id: Some(&merge.1),
event: &event,
actor,
ref_name: None,
old_target: None,
new_target: None,
created_at: completed_at,
},
)
}
fn end_sessions(
transaction: &rusqlite::Transaction<'_>,
account_id: i64,
ended_at: i64,
) -> Result<(), StoreError> {
transaction.execute(
"UPDATE web_session SET ended_at = ?2
WHERE account_id = ?1 AND ended_at IS NULL",
rusqlite::params![account_id, ended_at],
)?;
Ok(())
}
fn revoke_feed_tokens(
transaction: &rusqlite::Transaction<'_>,
account_id: i64,
revoked_at: i64,
) -> Result<(), StoreError> {
transaction.execute(
"UPDATE feed_token SET revoked_at = ?2
WHERE account_id = ?1 AND revoked_at IS NULL",
rusqlite::params![account_id, revoked_at],
)?;
Ok(())
}
fn insert_push_events(
transaction: &rusqlite::Transaction<'_>,
intent_id: &str,
repository_path: &str,
actor: &str,
initial_bytes: &[u8],
proposed_bytes: &[u8],
created_at: i64,
) -> Result<(), StoreError> {
let Some(repository_id) = managed_repository_id(repository_path) else {
return Ok(());
};
let exists = transaction
.query_row(
"SELECT id FROM repository WHERE id = ?1",
[repository_id],
|row| row.get::<_, String>(0),
)
.optional()?;
if exists.is_none() {
return Ok(());
}
let initial = parse_event_refs(initial_bytes)?;
let proposed = parse_event_refs(proposed_bytes)?;
if initial.len() != proposed.len() {
return Err(StoreError::EventPayload);
}
let push = event::push(intent_id);
insert_domain_event(
transaction,
&NewDomainEvent {
repository_id,
source_intent_id: Some(intent_id),
source_ordinal: Some(0),
issue_id: None,
pull_request_id: None,
event: &push,
actor,
ref_name: None,
old_target: None,
new_target: None,
created_at,
},
)?;
for (index, ((old, old_name), (new, new_name))) in initial.into_iter().zip(proposed).enumerate()
{
if old_name != new_name || (is_null_id(&old) && is_null_id(&new)) {
return Err(StoreError::EventPayload);
}
let tag = if old_name.starts_with(b"refs/tags/") {
true
} else if old_name.starts_with(b"refs/heads/") {
false
} else {
return Err(StoreError::EventPayload);
};
let kind = if is_null_id(&old) {
if tag {
event::EventKind::TagCreated
} else {
event::EventKind::RefCreated
}
} else if is_null_id(&new) {
if tag {
event::EventKind::TagDeleted
} else {
event::EventKind::RefDeleted
}
} else if tag {
event::EventKind::TagUpdated
} else {
event::EventKind::RefUpdated
};
let old_target = (!is_null_id(&old)).then_some(old);
let new_target = (!is_null_id(&new)).then_some(new);
let ordinal = i64::try_from(index + 1).map_err(|_| StoreError::EventPayload)?;
let event = event::reference(
kind,
&old_name,
old_target.as_deref(),
new_target.as_deref(),
);
insert_domain_event(
transaction,
&NewDomainEvent {
repository_id,
source_intent_id: Some(intent_id),
source_ordinal: Some(ordinal),
issue_id: None,
pull_request_id: None,
event: &event,
actor,
ref_name: Some(&old_name),
old_target: old_target.as_deref(),
new_target: new_target.as_deref(),
created_at,
},
)?;
}
insert_audit_event(
transaction,
&NewAuditEvent {
action: "ref.update",
actor,
target: repository_id,
outcome: "success",
correlation_id: intent_id,
created_at,
},
)?;
Ok(())
}
fn managed_repository_id(repository_path: &str) -> Option<&str> {
let name = Path::new(repository_path).file_name()?.to_str()?;
let id = name.strip_suffix(".git")?;
(id.len() == 32 && id.bytes().all(|byte| byte.is_ascii_hexdigit())).then_some(id)
}
fn parse_event_refs(bytes: &[u8]) -> Result<Vec<(String, Vec<u8>)>, StoreError> {
let mut references = Vec::new();
for line in bytes.split(|byte| *byte == b'\n') {
if line.is_empty() {
continue;
}
let Some(space) = line.iter().position(|byte| *byte == b' ') else {
return Err(StoreError::EventPayload);
};
let id = std::str::from_utf8(&line[..space]).map_err(|_| StoreError::EventPayload)?;
if !matches!(id.len(), 40 | 64) || !id.bytes().all(|byte| byte.is_ascii_hexdigit()) {
return Err(StoreError::EventPayload);
}
let name = line[space + 1..].to_vec();
if name.is_empty() {
return Err(StoreError::EventPayload);
}
references.push((id.to_owned(), name));
}
Ok(references)
}
fn is_null_id(id: &str) -> bool {
id.bytes().all(|byte| byte == b'0')
}
fn require_one_intent(id: &str, changed: usize) -> Result<(), StoreError> {
if changed == 1 {
Ok(())
} else {
Err(StoreError::IntentState(id.to_owned()))
}
}
fn active_account_id(connection: &Connection, username: &str) -> Result<i64, StoreError> {
let result = connection.query_row(
"SELECT id FROM account WHERE username = ?1 AND state = 'active'",
[username],
|row| row.get(0),
);
match result {
Ok(id) => Ok(id),
Err(rusqlite::Error::QueryReturnedNoRows) => {
Err(StoreError::AccountNotFound(username.to_owned()))
}
Err(error) => Err(error.into()),
}
}
fn repository_state_error(
transaction: &rusqlite::Transaction<'_>,
owner_id: i64,
owner: &str,
slug: &str,
) -> Result<StoreError, StoreError> {
let state = transaction.query_row(
"SELECT state FROM repository WHERE owner_account_id = ?1 AND slug = ?2",
rusqlite::params![owner_id, slug],
|row| row.get::<_, String>(0),
);
match state {
Ok(state) if state == "archived" => Ok(StoreError::RepositoryArchived(
owner.to_owned(),
slug.to_owned(),
)),
Ok(_) => unreachable!("repository state has a database constraint"),
Err(rusqlite::Error::QueryReturnedNoRows) => Ok(StoreError::RepositoryNotFound(
owner.to_owned(),
slug.to_owned(),
)),
Err(error) => Err(error.into()),
}
}
fn repository_from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<RepositoryRecord> {
Ok(RepositoryRecord {
id: row.get(0)?,
owner: row.get(1)?,
slug: row.get(2)?,
visibility: row.get(3)?,
state: row.get(4)?,
object_format: row.get(5)?,
created_at: row.get(6)?,
archived_at: row.get(7)?,
})
}
fn quote_identifier(identifier: &str) -> String {
format!("\"{}\"", identifier.replace('"', "\"\""))
}
fn dump_value(value: ValueRef<'_>) -> DumpValue {
match value {
ValueRef::Null => DumpValue::Null,
ValueRef::Integer(value) => DumpValue::Integer(value),
ValueRef::Real(value) => DumpValue::Real(format!("{value:.17e}")),
ValueRef::Text(value) => match std::str::from_utf8(value) {
Ok(value) => DumpValue::TextUtf8(value.to_owned()),
Err(_) => DumpValue::TextHex(hex(value)),
},
ValueRef::Blob(value) => DumpValue::BlobHex(hex(value)),
}
}
fn hex(bytes: &[u8]) -> String {
const HEX: &[u8; 16] = b"0123456789abcdef";
let mut output = String::with_capacity(bytes.len() * 2);
for byte in bytes {
output.push(char::from(HEX[(byte >> 4) as usize]));
output.push(char::from(HEX[(byte & 0x0f) as usize]));
}
output
}
fn is_unique_constraint(error: &rusqlite::Error) -> bool {
matches!(
error,
rusqlite::Error::SqliteFailure(code, _)
if code.extended_code == rusqlite::ffi::SQLITE_CONSTRAINT_UNIQUE
|| code.extended_code == rusqlite::ffi::SQLITE_CONSTRAINT_PRIMARYKEY
)
}
fn page_offset(page: usize, page_size: usize) -> Result<i64, StoreError> {
page.checked_sub(1)
.and_then(|page| page.checked_mul(page_size))
.and_then(|offset| i64::try_from(offset).ok())
.ok_or(StoreError::EventLimit)
}
#[allow(
dead_code,
reason = "the integration test imports this module without the CLI operation"
)]
pub(crate) fn doctor(instance_dir: &Path) -> Result<(), StoreError> {
let path = instance_dir.join(DATABASE_FILE);
let store = Store::open_read_only(&path)?;
let actual = store.schema_version()?;
if actual != SCHEMA_VERSION {
return Err(StoreError::SchemaVersion {
expected: SCHEMA_VERSION,
actual,
});
}
store.integrity_check()
}
#[allow(
dead_code,
reason = "M1A proves migrations before the M2 server calls them"
)]
fn migration_backup_path(path: &Path, version: i64) -> PathBuf {
let mut backup = OsString::from(path.as_os_str());
backup.push(format!(".v{version}.backup"));
PathBuf::from(backup)
}
fn configure(connection: &Connection) -> Result<(), StoreError> {
connection.busy_timeout(BUSY_TIMEOUT)?;
connection.pragma_update(None, "journal_mode", "WAL")?;
connection.pragma_update(None, "synchronous", "FULL")?;
connection.pragma_update(None, "foreign_keys", true)?;
verify_text_setting(connection, "journal_mode", "wal")?;
verify_integer_setting(connection, "synchronous", 2, "2")?;
verify_integer_setting(connection, "foreign_keys", 1, "1")?;
verify_integer_setting(
connection,
"busy_timeout",
BUSY_TIMEOUT_MILLISECONDS,
"5000",
)?;
Ok(())
}
fn verify_text_setting(
connection: &Connection,
name: &'static str,
expected: &'static str,
) -> Result<(), StoreError> {
let actual: String = connection.pragma_query_value(None, name, |row| row.get(0))?;
if actual.eq_ignore_ascii_case(expected) {
return Ok(());
}
Err(StoreError::Setting {
name,
expected,
actual,
})
}
fn verify_integer_setting(
connection: &Connection,
name: &'static str,
expected: i64,
expected_text: &'static str,
) -> Result<(), StoreError> {
let actual: i64 = connection.pragma_query_value(None, name, |row| row.get(0))?;
if actual == expected {
return Ok(());
}
Err(StoreError::Setting {
name,
expected: expected_text,
actual: actual.to_string(),
})
}