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,
            &current.0,
            number,
            title,
            body,
            &current.2,
        );
        insert_pull_request_event(
            &transaction,
            &access.repository.id,
            &current.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, &current.0, number, &current.2, &current.3, state);
        insert_pull_request_event(
            &transaction,
            &access.repository.id,
            &current.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,
            &current.issue.id,
            change.number,
            title,
            body,
        );
        insert_issue_event(
            &transaction,
            &access.repository.id,
            &current.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(&current.issue.id, number, &comment_id, actor, body);
        insert_issue_event(
            &transaction,
            &access.repository.id,
            &current.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, &current.issue.id, number, state);
        insert_issue_event(
            &transaction,
            &access.repository.id,
            &current.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(),
    })
}
