#[allow(dead_code, reason = "the storage test exercises selected store APIs")]
#[path = "../src/store/mod.rs"]
mod store;

use std::process::{Child, Command};
use std::sync::mpsc;
use std::sync::{Arc, atomic::AtomicBool, atomic::Ordering};
use std::thread;
use std::time::{Duration, Instant};
use std::{env, ffi::OsString, fs};

use rusqlite::{Connection, ErrorCode, TransactionBehavior, params};
use store::{
    AuditContext, GitOperationIntent, InitialAdministrator, IssueChange, NewAuditEvent, NewIssue,
    NewRepository, NewRepositoryReference, RepositoryOrigin, Store, StoreError,
};
use tempfile::TempDir;

const V1_FIXTURE: &str = include_str!("fixtures/sqlite/v1.sql");
const V2_FIXTURE: &str = include_str!("fixtures/sqlite/v2.sql");
const V3_FIXTURE: &str = include_str!("fixtures/sqlite/v3.sql");
const V4_FIXTURE: &str = include_str!("fixtures/sqlite/v4.sql");
const V5_FIXTURE: &str = include_str!("fixtures/sqlite/v5.sql");
const V6_FIXTURE: &str = include_str!("fixtures/sqlite/v6.sql");
const V7_FIXTURE: &str = include_str!("fixtures/sqlite/v7.sql");
const V8_FIXTURE: &str = concat!(
    include_str!("fixtures/sqlite/v7.sql"),
    include_str!("../src/store/migrations/008_web_sessions.sql"),
    "PRAGMA user_version = 8;\n",
);
const V9_FIXTURE: &str = concat!(
    include_str!("fixtures/sqlite/v7.sql"),
    include_str!("../src/store/migrations/008_web_sessions.sql"),
    include_str!("../src/store/migrations/009_repository_authorization.sql"),
    "PRAGMA user_version = 9;\n",
);
const V10_FIXTURE: &str = concat!(
    include_str!("fixtures/sqlite/v7.sql"),
    include_str!("../src/store/migrations/008_web_sessions.sql"),
    include_str!("../src/store/migrations/009_repository_authorization.sql"),
    include_str!("../src/store/migrations/010_audit_history.sql"),
    "PRAGMA user_version = 10;\n",
);
const V11_FIXTURE: &str = concat!(
    include_str!("fixtures/sqlite/v7.sql"),
    include_str!("../src/store/migrations/008_web_sessions.sql"),
    include_str!("../src/store/migrations/009_repository_authorization.sql"),
    include_str!("../src/store/migrations/010_audit_history.sql"),
    include_str!("../src/store/migrations/011_domain_events.sql"),
    "PRAGMA user_version = 11;\n",
);
const V12_FIXTURE: &str = concat!(
    include_str!("fixtures/sqlite/v7.sql"),
    include_str!("../src/store/migrations/008_web_sessions.sql"),
    include_str!("../src/store/migrations/009_repository_authorization.sql"),
    include_str!("../src/store/migrations/010_audit_history.sql"),
    include_str!("../src/store/migrations/011_domain_events.sql"),
    include_str!("../src/store/migrations/012_issues.sql"),
    "PRAGMA user_version = 12;\n",
);
const V13_FIXTURE: &str = concat!(
    include_str!("fixtures/sqlite/v7.sql"),
    include_str!("../src/store/migrations/008_web_sessions.sql"),
    include_str!("../src/store/migrations/009_repository_authorization.sql"),
    include_str!("../src/store/migrations/010_audit_history.sql"),
    include_str!("../src/store/migrations/011_domain_events.sql"),
    include_str!("../src/store/migrations/012_issues.sql"),
    include_str!("../src/store/migrations/013_watches.sql"),
    "PRAGMA user_version = 13;\n",
);
const V14_FIXTURE: &str = concat!(
    include_str!("fixtures/sqlite/v7.sql"),
    include_str!("../src/store/migrations/008_web_sessions.sql"),
    include_str!("../src/store/migrations/009_repository_authorization.sql"),
    include_str!("../src/store/migrations/010_audit_history.sql"),
    include_str!("../src/store/migrations/011_domain_events.sql"),
    include_str!("../src/store/migrations/012_issues.sql"),
    include_str!("../src/store/migrations/013_watches.sql"),
    include_str!("../src/store/migrations/014_feed_tokens.sql"),
    "PRAGMA user_version = 14;\n",
);
const V15_FIXTURE: &str = concat!(
    include_str!("fixtures/sqlite/v7.sql"),
    include_str!("../src/store/migrations/008_web_sessions.sql"),
    include_str!("../src/store/migrations/009_repository_authorization.sql"),
    include_str!("../src/store/migrations/010_audit_history.sql"),
    include_str!("../src/store/migrations/011_domain_events.sql"),
    include_str!("../src/store/migrations/012_issues.sql"),
    include_str!("../src/store/migrations/013_watches.sql"),
    include_str!("../src/store/migrations/014_feed_tokens.sql"),
    include_str!("../src/store/migrations/015_pull_requests.sql"),
    "PRAGMA user_version = 15;\n",
);
const V16_FIXTURE: &str = concat!(
    include_str!("fixtures/sqlite/v7.sql"),
    include_str!("../src/store/migrations/008_web_sessions.sql"),
    include_str!("../src/store/migrations/009_repository_authorization.sql"),
    include_str!("../src/store/migrations/010_audit_history.sql"),
    include_str!("../src/store/migrations/011_domain_events.sql"),
    include_str!("../src/store/migrations/012_issues.sql"),
    include_str!("../src/store/migrations/013_watches.sql"),
    include_str!("../src/store/migrations/014_feed_tokens.sql"),
    include_str!("../src/store/migrations/015_pull_requests.sql"),
    include_str!("../src/store/migrations/016_pull_request_reviews.sql"),
    "PRAGMA user_version = 16;\n",
);
const V17_FIXTURE: &str = concat!(
    include_str!("fixtures/sqlite/v7.sql"),
    include_str!("../src/store/migrations/008_web_sessions.sql"),
    include_str!("../src/store/migrations/009_repository_authorization.sql"),
    include_str!("../src/store/migrations/010_audit_history.sql"),
    include_str!("../src/store/migrations/011_domain_events.sql"),
    include_str!("../src/store/migrations/012_issues.sql"),
    include_str!("../src/store/migrations/013_watches.sql"),
    include_str!("../src/store/migrations/014_feed_tokens.sql"),
    include_str!("../src/store/migrations/015_pull_requests.sql"),
    include_str!("../src/store/migrations/016_pull_request_reviews.sql"),
    include_str!("../src/store/migrations/017_pull_request_merges.sql"),
    "PRAGMA user_version = 17;\n",
);

fn database(directory: &TempDir, name: &str) -> std::path::PathBuf {
    directory.path().join(name)
}

fn migration_backup(path: &std::path::Path, version: i64) -> std::path::PathBuf {
    let mut backup = OsString::from(path.as_os_str());
    backup.push(format!(".v{version}.backup"));
    backup.into()
}

fn create_fixture(path: &std::path::Path, fixture: &str) {
    let connection = Connection::open(path).expect("open a fixture database");
    connection
        .execute_batch(fixture)
        .expect("create the fixture");
}

fn spawn_crash_child(
    mode: &str,
    database_path: &std::path::Path,
    ready_path: &std::path::Path,
) -> Child {
    Command::new(env::current_exe().expect("find the integration test executable"))
        .args(["--exact", "crash_child", "--nocapture", "--test-threads=1"])
        .env("TIT_M1A_CHILD_MODE", mode)
        .env("TIT_M1A_DATABASE", database_path)
        .env("TIT_M1A_READY", ready_path)
        .spawn()
        .expect("start the crash-test child")
}

fn wait_for_child(child: &mut Child, ready_path: &std::path::Path) {
    let deadline = Instant::now() + Duration::from_secs(10);
    loop {
        if ready_path.exists() {
            return;
        }
        if let Some(status) = child.try_wait().expect("inspect the crash-test child") {
            panic!("crash-test child stopped before it was ready: {status}");
        }
        assert!(Instant::now() < deadline, "crash-test child was not ready");
        thread::sleep(Duration::from_millis(10));
    }
}

fn kill_child(mut child: Child, ready_path: &std::path::Path) {
    wait_for_child(&mut child, ready_path);
    child.kill().expect("kill the crash-test child");
    child.wait().expect("wait for the crash-test child");
}

fn signal_ready_and_wait() {
    let ready_path = env::var_os("TIT_M1A_READY").expect("read the ready-file path");
    fs::write(ready_path, b"ready").expect("write the ready file");
    loop {
        thread::park();
    }
}

#[test]
fn crash_child() {
    let Some(mode) = env::var_os("TIT_M1A_CHILD_MODE") else {
        return;
    };
    let database_path = env::var_os("TIT_M1A_DATABASE").expect("read the database path");
    let database_path = std::path::Path::new(&database_path);

    match mode.to_str().expect("read the crash-test mode") {
        "write-uncommitted" => {
            let mut store = Store::open(database_path).expect("open the child store");
            let transaction = store
                .connection_mut()
                .transaction_with_behavior(TransactionBehavior::Immediate)
                .expect("start the child transaction");
            transaction
                .execute(
                    "INSERT INTO m1a_parent (id, name, created_at) VALUES (1, 'uncommitted', 1)",
                    [],
                )
                .expect("insert an uncommitted row");
            signal_ready_and_wait();
        }
        "write-committed" => {
            let store = Store::open(database_path).expect("open the child store");
            store
                .connection()
                .execute(
                    "INSERT INTO m1a_parent (id, name, created_at) VALUES (1, 'committed', 1)",
                    [],
                )
                .expect("insert a committed row");
            signal_ready_and_wait();
        }
        "migration-uncommitted" => {
            let mut store = Store::open_unmigrated(database_path).expect("open the child fixture");
            store
                .migrate_with_hook(|version| {
                    if version == 2 {
                        signal_ready_and_wait();
                    }
                })
                .expect("migrate the child fixture");
        }
        "migration-committed" => {
            Store::open(database_path).expect("migrate the child fixture");
            signal_ready_and_wait();
        }
        unexpected => panic!("unknown crash-test mode: {unexpected}"),
    }
}

#[test]
fn configures_connections_and_creates_the_current_schema() {
    let directory = TempDir::new().expect("create a temporary directory");
    let store = Store::open(&database(&directory, "store.sqlite")).expect("open the store");

    assert_eq!(store.schema_version().expect("read the schema version"), 23);
    assert_eq!(
        store
            .connection()
            .pragma_query_value::<String, _>(None, "journal_mode", |row| row.get(0))
            .expect("read the journal mode"),
        "wal"
    );
    assert_eq!(
        store
            .connection()
            .pragma_query_value::<i64, _>(None, "synchronous", |row| row.get(0))
            .expect("read the synchronization mode"),
        2
    );
    assert_eq!(
        store
            .connection()
            .pragma_query_value::<i64, _>(None, "foreign_keys", |row| row.get(0))
            .expect("read the foreign-key mode"),
        1
    );
    assert_eq!(
        store
            .connection()
            .pragma_query_value::<i64, _>(None, "busy_timeout", |row| row.get(0))
            .expect("read the busy timeout"),
        5_000
    );
    store.integrity_check().expect("check database integrity");
}

#[test]
fn persists_and_completes_git_operation_intents_with_their_events() {
    let directory = TempDir::new().expect("create a temporary directory");
    let mut store = Store::open(&database(&directory, "store.sqlite")).expect("open the store");
    let intent = GitOperationIntent {
        id: "00112233445566778899aabbccddeeff",
        repository_path: "/srv/tit/repositories/alice/example.git",
        actor: "SHA256:test-key",
        initial_refs: b"0000000000000000000000000000000000000000 refs/heads/main\n",
        proposed_refs: b"1111111111111111111111111111111111111111 refs/heads/main\n",
        event_payload: b"push event",
        quarantine_path: "/srv/tit/repositories/alice/example.git/objects/tit-quarantine/test",
        created_at: 1,
    };
    store.begin_git_intent(&intent).expect("begin an intent");
    let pending = store
        .incomplete_git_intents()
        .expect("list pending intents");
    assert_eq!(pending.len(), 1);
    assert_eq!(pending[0].id, intent.id);
    assert_eq!(pending[0].repository_path, intent.repository_path);
    assert_eq!(pending[0].initial_refs, intent.initial_refs);
    assert_eq!(pending[0].proposed_refs, intent.proposed_refs);
    assert_eq!(pending[0].quarantine_path, intent.quarantine_path);
    assert_eq!(pending[0].state, "pending");
    assert_eq!(pending[0].pack_name, None);

    store
        .mark_git_objects_promoted(intent.id, Some("pack-test.pack"))
        .expect("mark objects as promoted");
    let promoted = store
        .incomplete_git_intents()
        .expect("list promoted intents");
    assert_eq!(promoted[0].state, "promoted");
    assert_eq!(promoted[0].pack_name.as_deref(), Some("pack-test.pack"));
    store
        .complete_git_intent(intent.id)
        .expect("complete the intent");
    assert!(
        store
            .incomplete_git_intents()
            .expect("list incomplete intents")
            .is_empty()
    );
    assert_eq!(
        store
            .connection()
            .query_row("SELECT count(*) FROM git_operation_event", [], |row| row
                .get::<_, i64>(0))
            .expect("count Git operation events"),
        1
    );

    let abandoned = GitOperationIntent {
        id: "ffeeddccbbaa99887766554433221100",
        ..intent
    };
    store
        .begin_git_intent(&abandoned)
        .expect("begin an abandoned intent");
    store
        .abandon_git_intent(abandoned.id)
        .expect("abandon the intent");
    assert!(
        store
            .incomplete_git_intents()
            .expect("list incomplete intents")
            .is_empty()
    );
}

#[test]
fn creates_only_one_initial_administrator_in_one_transaction() {
    let directory = TempDir::new().expect("create a temporary directory");
    let mut store = Store::open(&database(&directory, "store.sqlite")).expect("open the store");
    let recovery_hash = [7_u8; 32];
    let administrator = InitialAdministrator {
        username: "alice",
        canonical_key: "ssh-ed25519 AAAAexample",
        fingerprint: "SHA256:example",
        recovery_hash: &recovery_hash,
        created_at: 10,
    };
    store
        .create_initial_administrator(&administrator)
        .expect("create the initial administrator");
    let record: (String, i64, String, String, Vec<u8>) = store
        .connection()
        .query_row(
            "SELECT account.username, account.is_administrator, account.state,
                    ssh_public_key.fingerprint, recovery_credential.credential_hash
             FROM account
             JOIN ssh_public_key ON ssh_public_key.account_id = account.id
             JOIN recovery_credential ON recovery_credential.account_id = account.id",
            [],
            |row| {
                Ok((
                    row.get(0)?,
                    row.get(1)?,
                    row.get(2)?,
                    row.get(3)?,
                    row.get(4)?,
                ))
            },
        )
        .expect("read the initial administrator");
    assert_eq!(record.0, "alice");
    assert_eq!(record.1, 1);
    assert_eq!(record.2, "active");
    assert_eq!(record.3, "SHA256:example");
    assert_eq!(record.4, recovery_hash);
    assert!(matches!(
        store.create_initial_administrator(&administrator),
        Err(StoreError::AlreadyInitialized)
    ));
    assert_eq!(
        store
            .connection()
            .query_row("SELECT count(*) FROM account", [], |row| row
                .get::<_, i64>(0))
            .expect("count accounts"),
        1
    );
}

#[test]
fn creates_renames_archives_and_reads_owned_repositories() {
    let directory = TempDir::new().expect("create a temporary directory");
    let mut store = Store::open(&database(&directory, "store.sqlite")).expect("open the store");
    let recovery_hash = [7_u8; 32];
    store
        .create_initial_administrator(&InitialAdministrator {
            username: "alice",
            canonical_key: "ssh-ed25519 AAAAexample",
            fingerprint: "SHA256:example",
            recovery_hash: &recovery_hash,
            created_at: 10,
        })
        .expect("create an account");
    let initial_references = [
        NewRepositoryReference {
            name: b"refs/heads/main".to_vec(),
            target: "1".repeat(64),
        },
        NewRepositoryReference {
            name: b"refs/tags/v1".to_vec(),
            target: "2".repeat(64),
        },
    ];
    let repository = NewRepository {
        id: "00112233445566778899aabbccddeeff",
        owner: "alice",
        slug: "project",
        object_format: "sha256",
        default_branch: "refs/heads/main",
        created_at: 20,
        origin: RepositoryOrigin::Imported,
        initial_references: &initial_references,
        actor: "admin-cli",
        correlation_id: "test-create",
    };
    store
        .create_repository(&repository)
        .expect("create a repository");
    let created = store
        .repository("alice", "project")
        .expect("read a repository");
    assert_eq!(created.id, repository.id);
    assert_eq!(created.owner, "alice");
    assert_eq!(created.slug, "project");
    assert_eq!(created.visibility, "public");
    assert_eq!(created.state, "active");
    assert_eq!(created.object_format, "sha256");
    assert_eq!(created.created_at, 20);
    assert_eq!(created.archived_at, None);
    let (_, imported_events) = store
        .public_repository_events("alice", "project", None, 10)
        .expect("read imported repository events");
    assert_eq!(imported_events.len(), 3);
    assert_eq!(imported_events[0].kind, "tag-created");
    assert_eq!(imported_events[1].kind, "ref-created");
    assert_eq!(imported_events[2].kind, "repository-imported");
    assert_eq!(
        imported_events
            .iter()
            .map(|event| event.sequence)
            .collect::<Vec<_>>(),
        vec![3, 2, 1]
    );
    for event in &imported_events {
        assert_eq!(event.event_id.len(), 32);
        assert!(
            event
                .event_id
                .bytes()
                .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
        );
        assert_eq!(event.payload_version, 1);
        let payload: serde_json::Value =
            serde_json::from_str(&event.payload).expect("parse a versioned event payload");
        assert_eq!(payload["version"], 1);
    }
    let imported_payload: serde_json::Value = serde_json::from_str(&imported_events[2].payload)
        .expect("parse the repository import payload");
    assert_eq!(imported_payload["owner"], "alice");
    assert_eq!(imported_payload["repository"], "project");
    assert_eq!(imported_payload["object_format"], "sha256");
    let event_plan: String = store
        .connection()
        .query_row(
            "EXPLAIN QUERY PLAN
             SELECT event_id FROM repository_event
             WHERE repository_id = ?1 AND sequence < ?2
             ORDER BY sequence DESC LIMIT ?3",
            rusqlite::params![repository.id, 100, 20],
            |row| row.get(3),
        )
        .expect("read the repository event query plan");
    assert!(
        event_plan.contains("repository_event_feed"),
        "query plan: {event_plan}"
    );
    let duplicate_sequence = store
        .connection()
        .execute(
            "INSERT INTO repository_event
             (event_id, repository_id, sequence, kind, actor, payload_version,
              payload, created_at)
             VALUES ('aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', ?1, 1, 'push', 'alice',
                     1, '{\"version\":1}', 20)",
            [repository.id],
        )
        .expect_err("reject a duplicate repository event sequence");
    assert_eq!(
        duplicate_sequence.sqlite_error_code(),
        Some(ErrorCode::ConstraintViolation)
    );
    let unversioned_payload = store
        .connection()
        .execute(
            "INSERT INTO repository_event
             (event_id, repository_id, sequence, kind, actor, payload_version,
              payload, created_at)
             VALUES ('bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb', ?1, 4, 'push', 'alice',
                     1, '{}', 20)",
            [repository.id],
        )
        .expect_err("reject an unversioned repository event payload");
    assert_eq!(
        unversioned_payload.sqlite_error_code(),
        Some(ErrorCode::ConstraintViolation)
    );
    store
        .connection()
        .execute_batch(
            "CREATE TEMP TRIGGER reject_repository_event
             BEFORE INSERT ON repository_event
             BEGIN
                 SELECT RAISE(ABORT, 'injected event failure');
             END;",
        )
        .expect("install an event failure trigger");
    let rejected_repository = NewRepository {
        id: "11111111111111111111111111111111",
        owner: "alice",
        slug: "event-failure",
        object_format: "sha1",
        default_branch: "refs/heads/main",
        created_at: 20,
        origin: RepositoryOrigin::Created,
        initial_references: &[],
        actor: "alice",
        correlation_id: "test-event-failure",
    };
    assert!(matches!(
        store.create_repository(&rejected_repository),
        Err(StoreError::Sqlite(_))
    ));
    assert!(matches!(
        store.repository("alice", "event-failure"),
        Err(StoreError::RepositoryNotFound(_, _))
    ));
    store
        .connection()
        .execute_batch("DROP TRIGGER reject_repository_event;")
        .expect("remove the event failure trigger");

    let initial = format!("{} refs/heads/main\n", "0".repeat(64));
    let proposed = format!("{} refs/heads/main\n", "a".repeat(64));
    let push = GitOperationIntent {
        id: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
        repository_path: "/srv/tit/repositories/00112233445566778899aabbccddeeff.git",
        actor: "alice",
        initial_refs: initial.as_bytes(),
        proposed_refs: proposed.as_bytes(),
        event_payload: proposed.as_bytes(),
        quarantine_path: "/srv/tit/quarantine/push",
        created_at: 21,
    };
    store.begin_git_intent(&push).expect("begin a managed push");
    let (_, pending_events) = store
        .public_repository_events("alice", "project", None, 10)
        .expect("read events while a push is pending");
    assert_eq!(pending_events.len(), 3);
    store
        .mark_git_objects_promoted(push.id, None)
        .expect("promote a managed push");
    let (_, promoted_events) = store
        .public_repository_events("alice", "project", None, 10)
        .expect("read events while a push is promoted");
    assert_eq!(promoted_events.len(), 3);
    store
        .connection()
        .execute_batch(
            "CREATE TEMP TRIGGER reject_push_event
             BEFORE INSERT ON repository_event
             BEGIN
                 SELECT RAISE(ABORT, 'injected push event failure');
             END;",
        )
        .expect("install a push event failure trigger");
    assert!(matches!(
        store.complete_git_intent(push.id),
        Err(StoreError::Sqlite(_))
    ));
    assert!(
        !store
            .git_intent_completed(push.id)
            .expect("read the rolled-back intent state")
    );
    let (_, rolled_back_events) = store
        .public_repository_events("alice", "project", None, 10)
        .expect("read events after the failed completion");
    assert_eq!(rolled_back_events.len(), 3);
    store
        .connection()
        .execute_batch("DROP TRIGGER reject_push_event;")
        .expect("remove the push event failure trigger");
    store
        .complete_git_intent(push.id)
        .expect("complete a managed push");
    assert!(
        store
            .git_intent_completed(push.id)
            .expect("read the completed intent")
    );
    let (_, pushed_events) = store
        .public_repository_events("alice", "project", None, 10)
        .expect("read pushed repository events");
    assert_eq!(pushed_events.len(), 5);
    assert_eq!(
        pushed_events
            .iter()
            .map(|event| event.sequence)
            .collect::<Vec<_>>(),
        vec![5, 4, 3, 2, 1]
    );
    assert_eq!(pushed_events[0].kind, "ref-created");
    assert_eq!(
        pushed_events[0].new_target.as_deref(),
        Some("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa")
    );
    assert_eq!(pushed_events[1].kind, "push");
    let push_payload: serde_json::Value =
        serde_json::from_str(&pushed_events[1].payload).expect("parse the push payload");
    assert_eq!(push_payload["operation_id"], push.id);
    let (_, older_events) = store
        .public_repository_events("alice", "project", Some(4), 10)
        .expect("read events before a repository sequence");
    assert_eq!(
        older_events
            .iter()
            .map(|event| event.sequence)
            .collect::<Vec<_>>(),
        vec![3, 2, 1]
    );
    let event_ids = pushed_events
        .iter()
        .map(|event| event.event_id.clone())
        .collect::<Vec<_>>();
    let reopened = Store::open(&database(&directory, "store.sqlite"))
        .expect("reopen the repository event store");
    let (_, reopened_events) = reopened
        .public_repository_events("alice", "project", None, 10)
        .expect("read repository events after a reopen");
    assert_eq!(
        reopened_events
            .iter()
            .map(|event| event.event_id.clone())
            .collect::<Vec<_>>(),
        event_ids
    );
    drop(reopened);
    let audits = store.audit_events(10).expect("read audit history");
    assert_eq!(audits.len(), 2);
    assert_eq!(audits[0].action, "ref.update");
    assert_eq!(audits[0].actor, "alice");
    assert_eq!(audits[0].target, repository.id);
    assert_eq!(audits[0].outcome, "success");
    assert_eq!(audits[0].correlation_id, push.id);
    assert_eq!(audits[1].action, "repository.import");
    assert_eq!(audits[1].created_at, 20);
    assert!(matches!(store.audit_events(0), Err(StoreError::AuditLimit)));
    store
        .record_audit_event(&NewAuditEvent {
            action: "repository.rename",
            actor: "admin-cli",
            target: "alice/missing",
            outcome: "failure",
            correlation_id: "test-failure",
            created_at: 22,
        })
        .expect("record an audit failure");
    assert_eq!(
        store.audit_events(1).expect("read the newest audit event")[0].outcome,
        "failure"
    );
    store
        .connection()
        .execute(
            "INSERT INTO account
             (id, username, is_administrator, state, created_at)
             VALUES (2, 'bob', 0, 'active', 20)",
            [],
        )
        .expect("create a collaborator");
    store
        .update_repository_settings(
            "alice",
            "project",
            "alice",
            "A small public repository.",
            "private",
            23,
            "settings",
        )
        .expect("update repository settings");
    store
        .update_repository_collaborator(
            "alice",
            "project",
            "alice",
            "bob",
            Some("reader"),
            24,
            "collaborator",
        )
        .expect("add a repository collaborator");
    let settings = store
        .repository_settings("alice", "project", "alice")
        .expect("read repository settings");
    assert_eq!(settings.description, "A small public repository.");
    assert_eq!(settings.repository.visibility, "private");
    assert_eq!(settings.collaborators.len(), 1);
    assert_eq!(settings.collaborators[0].username, "bob");
    assert!(matches!(
        store.repository_settings("alice", "project", "bob"),
        Err(StoreError::PullRequestDenied)
    ));
    store
        .update_repository_settings(
            "alice",
            "project",
            "alice",
            "A small public repository.",
            "public",
            25,
            "settings-public",
        )
        .expect("restore public visibility");

    assert!(matches!(
        store.create_repository(&repository),
        Err(StoreError::RepositoryIdentifierCollision)
    ));
    let duplicate_name = NewRepository {
        id: "ffeeddccbbaa99887766554433221100",
        ..repository
    };
    assert!(matches!(
        store.create_repository(&duplicate_name),
        Err(StoreError::RepositoryExists(owner, slug))
            if owner == "alice" && slug == "project"
    ));
    let missing_owner = NewRepository {
        id: "1234567890abcdef1234567890abcdef",
        owner: "charlie",
        slug: "project",
        object_format: "sha1",
        default_branch: "refs/heads/main",
        created_at: 20,
        origin: RepositoryOrigin::Created,
        initial_references: &[],
        actor: "admin-cli",
        correlation_id: "test-missing",
    };
    assert!(matches!(
        store.create_repository(&missing_owner),
        Err(StoreError::AccountNotFound(owner)) if owner == "charlie"
    ));
    let (_, unchanged_events) = store
        .public_repository_events("alice", "project", None, 10)
        .expect("read events after rejected repository mutations");
    assert_eq!(unchanged_events.len(), 5);

    assert!(matches!(
        store.rename_repository_for_owner(
            "alice",
            "project",
            "reader-rename",
            "bob",
            25,
            "test-reader-rename",
        ),
        Err(StoreError::PullRequestDenied)
    ));
    store
        .rename_repository_for_owner("alice", "project", "renamed", "alice", 25, "test-rename")
        .expect("rename a repository");
    assert!(matches!(
        store.repository("alice", "project"),
        Err(StoreError::RepositoryNotFound(_, _))
    ));
    store
        .archive_repository_for_actor("alice", "renamed", "alice", 30, "test-archive")
        .expect("archive a repository");
    let archived = store
        .repository("alice", "renamed")
        .expect("read the archived repository");
    assert_eq!(archived.state, "archived");
    assert_eq!(archived.archived_at, Some(30));
    assert!(matches!(
        store.archive_repository("alice", "renamed", 31, "admin-cli", "test-archive-fail"),
        Err(StoreError::RepositoryArchived(_, _))
    ));
    assert!(matches!(
        store.rename_repository(
            "alice",
            "renamed",
            "again",
            32,
            "admin-cli",
            "test-rename-fail"
        ),
        Err(StoreError::RepositoryArchived(_, _))
    ));
    assert!(matches!(
        store.unarchive_repository_for_owner(
            "alice",
            "renamed",
            "bob",
            33,
            "test-reader-unarchive",
        ),
        Err(StoreError::PullRequestDenied)
    ));
    store
        .unarchive_repository_for_owner("alice", "renamed", "alice", 34, "test-unarchive")
        .expect("unarchive a repository");
    let unarchived = Store::open(&database(&directory, "store.sqlite"))
        .expect("reopen the unarchived repository store")
        .repository_settings("alice", "renamed", "alice")
        .expect("read the unarchived repository after restart");
    assert_eq!(unarchived.repository.state, "active");
    assert_eq!(unarchived.repository.archived_at, None);
    assert_eq!(unarchived.repository.visibility, "public");
    assert_eq!(unarchived.collaborators.len(), 1);
}

#[test]
fn runs_the_issue_workflow_with_atomic_events_and_repository_roles() {
    let directory = TempDir::new().expect("create an issue fixture directory");
    let mut store = Store::open(&database(&directory, "issues.sqlite")).expect("open the store");
    store
        .create_initial_administrator(&InitialAdministrator {
            username: "alice",
            canonical_key: "ssh-ed25519 AAAAalice",
            fingerprint: "SHA256:alice",
            recovery_hash: &[7; 32],
            created_at: 1,
        })
        .expect("create the repository owner");
    store
        .connection()
        .execute_batch(
            "INSERT INTO account (id, username, is_administrator, state, created_at) VALUES
                 (2, 'bob', 0, 'active', 1),
                 (3, 'carol', 0, 'active', 1),
                 (4, 'maintainer', 0, 'active', 1),
                 (5, 'stranger', 0, 'active', 1),
                 (6, 'suspended', 0, 'suspended', 1);",
        )
        .expect("create issue actors");
    store
        .create_repository(&NewRepository {
            id: "00112233445566778899aabbccddeeff",
            owner: "alice",
            slug: "project",
            object_format: "sha1",
            default_branch: "refs/heads/main",
            created_at: 2,
            origin: RepositoryOrigin::Created,
            initial_references: &[],
            actor: "alice",
            correlation_id: "issue-repository",
        })
        .expect("create an issue repository");
    store
        .connection()
        .execute_batch(
            "UPDATE repository SET visibility = 'private';
             INSERT INTO repository_collaborator
                 (repository_id, account_id, role, created_at)
             VALUES
                 ('00112233445566778899aabbccddeeff', 2, 'reader', 3),
                 ('00112233445566778899aabbccddeeff', 3, 'writer', 3),
                 ('00112233445566778899aabbccddeeff', 4, 'maintainer', 3);",
        )
        .expect("configure issue roles");

    let source = "Use **the supplied Markdown**.\n\n<script>do not run</script>";
    let issue = store
        .create_issue(&NewIssue {
            owner: "alice",
            repository: "project",
            actor: "bob",
            title: "Preserve Markdown",
            body: source,
            created_at: 4,
        })
        .expect("create an issue as a reader");
    assert_eq!(issue.number, 1);
    assert_eq!(issue.body, source);
    let second = store
        .create_issue(&NewIssue {
            owner: "alice",
            repository: "project",
            actor: "carol",
            title: "Second issue",
            body: "",
            created_at: 5,
        })
        .expect("allocate the next issue number");
    assert_eq!(second.number, 2);
    assert!(matches!(
        store.issues("alice", "project", None),
        Err(StoreError::IssueHidden)
    ));
    assert_eq!(
        store
            .issues("alice", "project", Some("bob"))
            .expect("list private issues as a reader")
            .1
            .len(),
        2
    );
    let change = |actor, changed_at| IssueChange {
        owner: "alice",
        repository: "project",
        number: 1,
        actor,
        changed_at,
    };
    assert!(matches!(
        store.edit_issue(
            &IssueChange {
                number: 2,
                ..change("bob", 6)
            },
            "Reader edit",
            "denied",
        ),
        Err(StoreError::IssueDenied)
    ));
    store
        .edit_issue(&change("bob", 7), "Preserve the source", source)
        .expect("edit an authored issue");
    store
        .edit_issue(&change("carol", 8), "Writer edit", source)
        .expect("edit an issue as a writer");
    let comment_id = store
        .comment_issue(
            "alice",
            "project",
            1,
            "bob",
            "A comment with [text](https://example.com).",
            9,
        )
        .expect("comment as a reader");
    assert_eq!(comment_id.len(), 32);
    store
        .set_issue_state("alice", "project", 1, "bob", "closed", 10)
        .expect("close an authored issue");
    store
        .set_issue_state("alice", "project", 1, "carol", "open", 11)
        .expect("reopen an issue as a writer");

    let detail = store
        .issue_detail("alice", "project", 1, Some("maintainer"), 1, 1, 50)
        .expect("read the issue timeline");
    assert_eq!(detail.repository.slug, "project");
    assert_eq!(detail.issue.title, "Writer edit");
    assert_eq!(detail.issue.body, source);
    assert_eq!(detail.issue.state, "open");
    assert_eq!(detail.comments.len(), 1);
    assert!(detail.can_comment);
    assert!(detail.can_edit);
    assert_eq!(
        detail
            .timeline
            .iter()
            .map(|event| event.kind.as_str())
            .collect::<Vec<_>>(),
        [
            "issue-created",
            "issue-edited",
            "issue-edited",
            "issue-commented",
            "issue-closed",
            "issue-reopened",
        ]
    );
    assert!(
        detail
            .timeline
            .windows(2)
            .all(|events| events[0].sequence < events[1].sequence)
    );
    let first_activity_page = store
        .issue_detail("alice", "project", 1, Some("maintainer"), 1, 1, 1)
        .expect("read the first bounded activity page");
    assert_eq!(first_activity_page.comments.len(), 1);
    assert!(!first_activity_page.comments_has_next);
    assert_eq!(first_activity_page.timeline.len(), 1);
    assert!(first_activity_page.timeline_has_next);
    let second_timeline_page = store
        .issue_detail("alice", "project", 1, Some("maintainer"), 1, 2, 1)
        .expect("read the second timeline page");
    assert_eq!(second_timeline_page.timeline_page, 2);
    assert_eq!(second_timeline_page.timeline.len(), 1);
    for event in &detail.timeline {
        let payload: serde_json::Value =
            serde_json::from_str(&event.payload).expect("parse an issue event payload");
        assert_eq!(payload["version"], 1);
        assert_eq!(payload["issue_id"], issue.id);
        assert!(event.created_at >= 4);
        assert!(!event.actor.is_empty());
    }

    let events_before_watch: i64 = store
        .connection()
        .query_row("SELECT count(*) FROM repository_event", [], |row| {
            row.get(0)
        })
        .expect("count events before watch changes");
    let watch = store
        .set_watch("alice", "project", "bob", true, 16)
        .expect("watch all repository activity")
        .expect("read the created watch");
    assert_eq!(watch.id.len(), 32);
    assert_eq!(watch.created_at, 16);
    assert_eq!(watch.updated_at, 16);
    let updated = store
        .set_watch("alice", "project", "bob", true, 17)
        .expect("keep watching")
        .expect("read the updated watch");
    assert_eq!(updated.id, watch.id);
    assert_eq!(updated.created_at, 16);
    assert_eq!(updated.updated_at, 17);
    let (watched_repository, stored_watch) = store
        .watch("alice", "project", Some("bob"))
        .expect("read watch preferences");
    assert_eq!(watched_repository.slug, "project");
    assert_eq!(stored_watch.expect("find the watch"), updated);
    assert!(matches!(
        store.watch("alice", "project", Some("stranger")),
        Err(StoreError::WatchDenied)
    ));
    assert!(matches!(
        store.set_watch("alice", "project", "suspended", true, 18),
        Err(StoreError::WatchDenied)
    ));
    assert_eq!(
        store
            .connection()
            .query_row("SELECT count(*) FROM repository_event", [], |row| row
                .get::<_, i64>(0))
            .expect("count events after watch changes"),
        events_before_watch
    );
    assert_eq!(
        store
            .set_watch("alice", "project", "bob", false, 19)
            .expect("stop watching the repository"),
        None
    );
    assert_eq!(
        store
            .watch("alice", "project", Some("bob"))
            .expect("read the cleared watch")
            .1,
        None
    );
    store
        .set_watch("alice", "project", "bob", true, 20)
        .expect("restore the watch");
    store
        .comment_issue(
            "alice",
            "project",
            1,
            "carol",
            "Please review this, @bob. Do not match @bobby.",
            21,
        )
        .expect("create a mention event");
    let first_web_activity = store
        .watched_activity_page("bob", None, 2)
        .expect("read the first watched activity page");
    assert_eq!(first_web_activity.events.len(), 2);
    let before = first_web_activity
        .next_before
        .as_ref()
        .expect("read the stable activity cursor");
    store
        .comment_issue("alice", "project", 1, "carol", "Newer activity.", 22)
        .expect("create activity after the first page");
    let second_web_activity = store
        .watched_activity_page("bob", Some(before), 2)
        .expect("read the second watched activity page");
    assert!(second_web_activity.events.iter().all(|second| {
        first_web_activity
            .events
            .iter()
            .all(|first| first.event.event_id != second.event.event_id)
    }));
    store
        .connection()
        .execute_batch(
            "UPDATE repository SET visibility = 'private' WHERE slug = 'project';
             DELETE FROM repository_collaborator
             WHERE repository_id = (SELECT id FROM repository WHERE slug = 'project')
               AND account_id = (SELECT id FROM account WHERE username = 'bob');",
        )
        .expect("remove watched private repository access");
    assert!(
        store
            .watched_activity_page("bob", None, 20)
            .expect("apply activity access immediately")
            .events
            .is_empty()
    );
    store
        .set_repository_collaborator(
            "alice",
            "project",
            "bob",
            "maintainer",
            &AuditContext {
                actor: "alice",
                correlation_id: "restore-activity-access",
                created_at: 22,
            },
        )
        .expect("restore watched repository access");
    store
        .connection()
        .execute(
            "UPDATE repository
             SET state = 'archived', archived_at = 22 WHERE slug = 'project'",
            [],
        )
        .expect("archive the watched repository");
    assert!(
        store
            .watched_activity_page("bob", None, 20)
            .expect("apply archived state immediately")
            .events
            .is_empty()
    );
    store
        .connection()
        .execute(
            "UPDATE repository
             SET state = 'active', archived_at = NULL WHERE slug = 'project'",
            [],
        )
        .expect("restore the watched repository");
    let watched_token = store
        .create_feed_token("bob", &[2; 32], 23)
        .expect("create a watched feed token");
    assert!(matches!(
        store.create_feed_token("bob", &[3; 32], 24),
        Err(StoreError::FeedTokenLimit)
    ));
    let watched_page = store
        .token_feed_events(&[2; 32], 100)
        .expect("read watched issue activity");
    assert!(
        watched_page
            .events
            .iter()
            .any(|event| event.event.kind == "issue-commented")
    );
    assert!(!watched_page.events.is_empty());
    assert_eq!(store.feed_tokens("bob").expect("list feed tokens").len(), 1);
    store
        .connection()
        .execute_batch(
            "CREATE TEMP TRIGGER reject_feed_token
             BEFORE INSERT ON feed_token
             BEGIN
                 SELECT RAISE(ABORT, 'injected feed token failure');
             END;",
        )
        .expect("inject a feed token failure");
    assert!(matches!(
        store.rotate_feed_token("bob", &watched_token.id, &[7; 32], 27),
        Err(StoreError::Sqlite(_))
    ));
    store
        .token_feed_events(&[2; 32], 100)
        .expect("retain the old token after a failed rotation");
    store
        .connection()
        .execute_batch("DROP TRIGGER reject_feed_token;")
        .expect("remove the feed token failure");
    let rotated = store
        .rotate_feed_token("bob", &watched_token.id, &[6; 32], 28)
        .expect("rotate the watched feed token");
    assert_eq!(rotated.scope, watched_token.scope);
    assert!(matches!(
        store.token_feed_events(&[2; 32], 100),
        Err(StoreError::FeedTokenNotFound)
    ));
    store
        .revoke_feed_token("bob", &rotated.id, 29)
        .expect("revoke the watched feed token");
    assert!(matches!(
        store.token_feed_events(&[6; 32], 100),
        Err(StoreError::FeedTokenNotFound)
    ));
    assert!(
        store
            .repository_issue_events("alice", "project", None, 100)
            .expect("read issue-only events")
            .1
            .iter()
            .all(|event| event.kind.starts_with("issue-"))
    );
    assert_eq!(watched_token.scope, "watched");
    let mut metadata = Vec::new();
    let metadata_truncated = store
        .visit_metadata_search_candidates(Some("bob"), 100, |candidate| {
            metadata.push(candidate);
            true
        })
        .expect("scan authorized metadata");
    assert!(!metadata_truncated);
    assert!(metadata.iter().any(|candidate| {
        candidate.kind == "repository"
            && candidate.record_id == "00112233445566778899aabbccddeeff"
            && candidate.owner == "alice"
            && candidate.repository == "project"
            && candidate.title == b"alice/project"
            && candidate.body.is_empty()
    }));
    let mut anonymous_metadata = Vec::new();
    let anonymous_truncated = store
        .visit_metadata_search_candidates(None, 100, |candidate| {
            anonymous_metadata.push(candidate);
            true
        })
        .expect("scan anonymous metadata");
    assert!(!anonymous_truncated);
    assert!(anonymous_metadata.is_empty());

    let comments_before: i64 = store
        .connection()
        .query_row("SELECT count(*) FROM issue_comment", [], |row| row.get(0))
        .expect("count issue comments");
    store
        .connection()
        .execute_batch(
            "CREATE TEMP TRIGGER reject_issue_event
             BEFORE INSERT ON repository_event
             BEGIN
                 SELECT RAISE(ABORT, 'injected issue event failure');
             END;",
        )
        .expect("inject an issue event failure");
    assert!(matches!(
        store.comment_issue("alice", "project", 1, "bob", "must roll back", 21),
        Err(StoreError::Sqlite(_))
    ));
    assert_eq!(
        store
            .connection()
            .query_row("SELECT count(*) FROM issue_comment", [], |row| row
                .get::<_, i64>(0))
            .expect("count comments after rollback"),
        comments_before
    );
    store
        .create_feed_token("bob", &[10; 32], 30)
        .expect("create a replacement watched feed token");
    assert!(matches!(
        store.create_feed_token("bob", &[11; 32], 31),
        Err(StoreError::FeedTokenLimit)
    ));
}

#[test]
fn enforces_constraints_and_supports_crud_and_indexed_scans() {
    let directory = TempDir::new().expect("create a temporary directory");
    let mut store = Store::open(&database(&directory, "store.sqlite")).expect("open the store");
    let connection = store.connection_mut();

    connection
        .execute(
            "INSERT INTO m1a_parent (id, name, created_at) VALUES (?1, ?2, ?3)",
            params![1, "parent", 10],
        )
        .expect("insert a parent");
    connection
        .execute(
            "INSERT INTO m1a_child (id, parent_id, sequence, body) VALUES (?1, ?2, ?3, ?4)",
            params![1, 1, 1, "body"],
        )
        .expect("insert a child");

    let duplicate = connection
        .execute(
            "INSERT INTO m1a_parent (id, name, created_at) VALUES (?1, ?2, ?3)",
            params![2, "parent", 11],
        )
        .expect_err("reject a duplicate name");
    assert_eq!(
        duplicate.sqlite_error_code(),
        Some(ErrorCode::ConstraintViolation)
    );

    let orphan = connection
        .execute(
            "INSERT INTO m1a_child (id, parent_id, sequence, body) VALUES (?1, ?2, ?3, ?4)",
            params![2, 999, 1, "orphan"],
        )
        .expect_err("reject an orphan");
    assert_eq!(
        orphan.sqlite_error_code(),
        Some(ErrorCode::ConstraintViolation)
    );

    let deletion = connection
        .execute("DELETE FROM m1a_parent WHERE id = ?1", [1])
        .expect_err("restrict deletion of a referenced parent");
    assert_eq!(
        deletion.sqlite_error_code(),
        Some(ErrorCode::ConstraintViolation)
    );

    connection
        .execute(
            "UPDATE m1a_child SET body = ?1, state = ?2 WHERE id = ?3",
            params!["changed", "closed", 1],
        )
        .expect("update a child");
    let child: (String, String) = connection
        .query_row(
            "SELECT body, state FROM m1a_child WHERE id = ?1",
            [1],
            |row| Ok((row.get(0)?, row.get(1)?)),
        )
        .expect("read the child");
    assert_eq!(child, ("changed".to_owned(), "closed".to_owned()));

    let plan: String = connection
        .query_row(
            "EXPLAIN QUERY PLAN SELECT id FROM m1a_parent WHERE created_at >= ?1 ORDER BY created_at, id",
            [0],
            |row| row.get(3),
        )
        .expect("read the query plan");
    assert!(plan.contains("m1a_parent_created_at"), "query plan: {plan}");

    connection
        .execute("DELETE FROM m1a_child WHERE id = ?1", [1])
        .expect("delete the child");
    connection
        .execute("DELETE FROM m1a_parent WHERE id = ?1", [1])
        .expect("delete the parent");
    assert_eq!(
        connection
            .query_row("SELECT count(*) FROM m1a_parent", [], |row| row
                .get::<_, i64>(0))
            .expect("count parents"),
        0
    );
}

#[test]
fn rolls_back_a_failed_unit_of_work() {
    let directory = TempDir::new().expect("create a temporary directory");
    let mut store = Store::open(&database(&directory, "store.sqlite")).expect("open the store");

    let transaction = store
        .connection_mut()
        .transaction()
        .expect("start a transaction");
    transaction
        .execute(
            "INSERT INTO m1a_parent (id, name, created_at) VALUES (?1, ?2, ?3)",
            params![1, "rolled-back", 1],
        )
        .expect("insert a parent");
    transaction.rollback().expect("roll back the transaction");

    assert_eq!(
        store
            .connection()
            .query_row("SELECT count(*) FROM m1a_parent", [], |row| row
                .get::<_, i64>(0))
            .expect("count parents"),
        0
    );
}

#[test]
fn recovers_committed_state_and_discards_uncommitted_state_after_a_process_kill() {
    for (mode, expected_rows) in [("write-uncommitted", 0), ("write-committed", 1)] {
        let directory = TempDir::new().expect("create a temporary directory");
        let path = database(&directory, "store.sqlite");
        Store::open(&path).expect("create the store");
        let ready_path = database(&directory, "ready");

        let child = spawn_crash_child(mode, &path, &ready_path);
        kill_child(child, &ready_path);

        let store = Store::open(&path).expect("recover the store");
        store.integrity_check().expect("check recovered integrity");
        assert_eq!(
            store
                .connection()
                .query_row("SELECT count(*) FROM m1a_parent", [], |row| row
                    .get::<_, i64>(0))
                .expect("count recovered rows"),
            expected_rows,
            "recovered row count for {mode}"
        );
    }
}

#[test]
fn permits_reads_during_a_write_and_serializes_writers() {
    let directory = TempDir::new().expect("create a temporary directory");
    let path = database(&directory, "store.sqlite");
    let store = Store::open(&path).expect("open the store");
    store
        .connection()
        .execute(
            "INSERT INTO m1a_parent (id, name, created_at) VALUES (1, 'initial', 1)",
            [],
        )
        .expect("insert the initial parent");
    drop(store);

    let (locked_sender, locked_receiver) = mpsc::channel();
    let writer_path = path.clone();
    let writer = thread::spawn(move || {
        let mut store = Store::open(&writer_path).expect("open the first writer");
        let transaction = store
            .connection_mut()
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .expect("start the first write");
        transaction
            .execute(
                "INSERT INTO m1a_parent (id, name, created_at) VALUES (2, 'first-writer', 2)",
                [],
            )
            .expect("write with the first writer");
        locked_sender.send(()).expect("signal the write lock");
        thread::sleep(Duration::from_millis(200));
        transaction.commit().expect("commit the first writer");
    });

    locked_receiver.recv().expect("wait for the write lock");
    let reader = Store::open(&path).expect("open a reader");
    assert_eq!(
        reader
            .connection()
            .query_row("SELECT count(*) FROM m1a_parent", [], |row| row
                .get::<_, i64>(0))
            .expect("read during a write"),
        1
    );
    drop(reader);

    let started = Instant::now();
    let second_writer = Store::open(&path).expect("open the second writer");
    second_writer
        .connection()
        .execute(
            "INSERT INTO m1a_parent (id, name, created_at) VALUES (3, 'second-writer', 3)",
            [],
        )
        .expect("write with the second writer");
    assert!(started.elapsed() >= Duration::from_millis(100));
    writer.join().expect("join the first writer");

    assert_eq!(
        second_writer
            .connection()
            .query_row("SELECT count(*) FROM m1a_parent", [], |row| row
                .get::<_, i64>(0))
            .expect("count committed writes"),
        3
    );
}

#[test]
fn returns_busy_after_the_configured_timeout() {
    let directory = TempDir::new().expect("create a temporary directory");
    let path = database(&directory, "store.sqlite");
    let mut first = Store::open(&path).expect("open the first store");
    let second = Store::open(&path).expect("open the second store");
    second
        .connection()
        .busy_timeout(Duration::from_millis(50))
        .expect("set a short busy timeout");

    let transaction = first
        .connection_mut()
        .transaction_with_behavior(TransactionBehavior::Immediate)
        .expect("start a write transaction");
    transaction
        .execute(
            "INSERT INTO m1a_parent (id, name, created_at) VALUES (1, 'locked', 1)",
            [],
        )
        .expect("hold the write lock");

    let started = Instant::now();
    let error = second
        .connection()
        .execute(
            "INSERT INTO m1a_parent (id, name, created_at) VALUES (2, 'busy', 2)",
            [],
        )
        .expect_err("return a busy error");
    assert_eq!(error.sqlite_error_code(), Some(ErrorCode::DatabaseBusy));
    assert!(started.elapsed() >= Duration::from_millis(40));
    transaction.rollback().expect("release the write lock");
}

#[test]
fn backs_up_a_consistent_database_while_writes_continue() {
    let directory = TempDir::new().expect("create a temporary directory");
    let source_path = database(&directory, "source.sqlite");
    let backup_path = database(&directory, "backup.sqlite");
    let source = Store::open(&source_path).expect("open the source store");
    source
        .connection()
        .execute(
            "INSERT INTO m1a_parent (id, name, created_at) VALUES (1, 'parent', 1)",
            [],
        )
        .expect("insert the parent");
    for sequence in 0..1_000 {
        source
            .connection()
            .execute(
                "INSERT INTO m1a_child (id, parent_id, sequence, body) VALUES (?1, 1, ?2, 'initial')",
                params![sequence + 1, sequence],
            )
            .expect("insert an initial child");
    }

    let running = Arc::new(AtomicBool::new(true));
    let writer_running = Arc::clone(&running);
    let writer_path = source_path.clone();
    let writer = thread::spawn(move || {
        let store = Store::open(&writer_path).expect("open the backup writer");
        let mut sequence = 1_000;
        while writer_running.load(Ordering::Relaxed) {
            store
                .connection()
                .execute(
                    "INSERT INTO m1a_child (id, parent_id, sequence, body) VALUES (?1, 1, ?2, 'concurrent')",
                    params![sequence + 1, sequence],
                )
                .expect("insert a concurrent child");
            sequence += 1;
        }
        sequence
    });

    let reader_running = Arc::clone(&running);
    let reader_path = source_path.clone();
    let (reader_ready_sender, reader_ready_receiver) = mpsc::channel();
    let reader = thread::spawn(move || {
        let store = Store::open(&reader_path).expect("open the backup reader");
        let mut read_count = 0;
        while reader_running.load(Ordering::Relaxed) {
            let _: i64 = store
                .connection()
                .query_row("SELECT count(*) FROM m1a_child", [], |row| row.get(0))
                .expect("read while the backup runs");
            if read_count == 0 {
                reader_ready_sender
                    .send(())
                    .expect("signal that the reader is ready");
            }
            read_count += 1;
        }
        read_count
    });
    reader_ready_receiver
        .recv_timeout(Duration::from_secs(5))
        .expect("wait for the backup reader");

    source
        .backup(&backup_path)
        .expect("create an online backup");
    running.store(false, Ordering::Relaxed);
    let final_sequence = writer.join().expect("join the backup writer");
    let read_count = reader.join().expect("join the backup reader");
    assert!(read_count > 0);
    let backup = Store::open(&backup_path).expect("open the backup");
    backup.integrity_check().expect("check backup integrity");
    let backup_count: i64 = backup
        .connection()
        .query_row("SELECT count(*) FROM m1a_child", [], |row| row.get(0))
        .expect("count backup children");
    assert!(backup_count >= 1_000);
    assert!(backup_count <= i64::from(final_sequence));
}

#[test]
fn migrates_each_committed_historical_fixture() {
    for (fixture, initial_version) in [
        (V1_FIXTURE, 1),
        (V2_FIXTURE, 2),
        (V3_FIXTURE, 3),
        (V4_FIXTURE, 4),
        (V5_FIXTURE, 5),
        (V6_FIXTURE, 6),
        (V7_FIXTURE, 7),
        (V8_FIXTURE, 8),
        (V9_FIXTURE, 9),
        (V10_FIXTURE, 10),
        (V11_FIXTURE, 11),
        (V12_FIXTURE, 12),
        (V13_FIXTURE, 13),
        (V14_FIXTURE, 14),
        (V15_FIXTURE, 15),
        (V16_FIXTURE, 16),
        (V17_FIXTURE, 17),
    ] {
        let directory = TempDir::new().expect("create a temporary directory");
        let path = database(&directory, "tit.sqlite3");
        create_fixture(&path, fixture);

        let store = Store::open(&path).expect("migrate the fixture");
        assert_eq!(store.schema_version().expect("read the schema version"), 23);
        store.integrity_check().expect("check migrated integrity");
        let state: String = store
            .connection()
            .query_row("SELECT state FROM m1a_child WHERE id = 1", [], |row| {
                row.get(0)
            })
            .expect("read the migrated state");
        let expected = if initial_version == 1 {
            "open"
        } else {
            "closed"
        };
        assert_eq!(state, expected);
        store::doctor(directory.path()).expect("check the migrated fixture with doctor");

        let backup_path = migration_backup(&path, initial_version);
        assert!(backup_path.exists());
        let backup = Store::open_unmigrated(&backup_path).expect("open the migration backup");
        assert_eq!(
            backup.schema_version().expect("read the backup version"),
            initial_version
        );
        backup.integrity_check().expect("check backup integrity");
    }
}

#[test]
fn migration_uses_an_existing_repository_symbolic_head() {
    let directory = TempDir::new().expect("create a migration directory");
    let path = database(&directory, "tit.sqlite3");
    let mut store = Store::open(&path).expect("create the current database");
    store
        .connection()
        .execute(
            "INSERT INTO account
             (id, username, is_administrator, state, created_at)
             VALUES (1, 'alice', 1, 'active', 1)",
            [],
        )
        .expect("create the migration account");
    store
        .create_repository(&NewRepository {
            id: "00112233445566778899aabbccddeeff",
            owner: "alice",
            slug: "project",
            object_format: "sha1",
            default_branch: "refs/heads/main",
            created_at: 2,
            origin: RepositoryOrigin::Created,
            initial_references: &[],
            actor: "alice",
            correlation_id: "migration-default",
        })
        .expect("create the migration repository");
    store
        .connection()
        .execute("DROP TABLE repository_default_branch", [])
        .expect("remove the new table from the historical fixture");
    store
        .connection()
        .pragma_update(None, "user_version", 22)
        .expect("set the historical schema version");
    drop(store);
    let repository = directory
        .path()
        .join("repositories")
        .join("00112233445566778899aabbccddeeff.git");
    fs::create_dir_all(&repository).expect("create the historical repository directory");
    fs::write(repository.join("HEAD"), "ref: refs/heads/trunk\n")
        .expect("write the historical symbolic HEAD");

    let migrated = Store::open(&path).expect("migrate the historical database");
    assert_eq!(
        migrated
            .repository_default_branch("alice", "project")
            .expect("read the migrated default branch"),
        "refs/heads/trunk"
    );
}

#[test]
fn backfills_repository_events_when_version_five_is_migrated() {
    let directory = TempDir::new().expect("create a temporary directory");
    let path = database(&directory, "tit.sqlite3");
    create_fixture(&path, V5_FIXTURE);
    let connection = Connection::open(&path).expect("open the version-five fixture");
    connection
        .execute(
            "INSERT INTO account
             (id, username, is_administrator, state, created_at)
             VALUES (1, 'alice', 1, 'active', 1)",
            [],
        )
        .expect("insert a historical account");
    connection
        .execute(
            "INSERT INTO repository
             (id, owner_account_id, slug, visibility, state, object_format, created_at)
             VALUES ('00112233445566778899aabbccddeeff', 1, 'project', 'public',
                     'active', 'sha1', 2)",
            [],
        )
        .expect("insert a historical repository");
    connection
        .pragma_update(None, "user_version", 5)
        .expect("set the historical schema version");
    drop(connection);

    let store = Store::open(&path).expect("migrate the version-five fixture");
    let (_, events) = store
        .public_repository_events("alice", "project", None, 10)
        .expect("read the backfilled event");
    assert_eq!(events.len(), 1);
    assert_eq!(events[0].kind, "repository-created");
    assert_eq!(events[0].sequence, 1);
    assert_eq!(events[0].payload_version, 1);
    assert_eq!(
        serde_json::from_str::<serde_json::Value>(&events[0].payload)
            .expect("parse the backfilled event payload")["version"],
        1
    );
    assert_eq!(events[0].created_at, 2);
}

#[test]
fn recovers_complete_schema_versions_after_a_process_kill_during_migration() {
    for (mode, expected_version) in [("migration-uncommitted", 1), ("migration-committed", 23)] {
        let directory = TempDir::new().expect("create a temporary directory");
        let path = database(&directory, "fixture.sqlite");
        create_fixture(&path, V1_FIXTURE);
        let ready_path = database(&directory, "ready");

        let child = spawn_crash_child(mode, &path, &ready_path);
        kill_child(child, &ready_path);

        let store = Store::open_unmigrated(&path).expect("recover the fixture");
        assert_eq!(
            store.schema_version().expect("read the recovered version"),
            expected_version,
            "recovered schema version for {mode}"
        );
        store.integrity_check().expect("check recovered integrity");

        let has_state: i64 = store
            .connection()
            .query_row(
                "SELECT count(*) FROM pragma_table_info('m1a_child') WHERE name = 'state'",
                [],
                |row| row.get(0),
            )
            .expect("inspect the recovered schema");
        assert_eq!(has_state, i64::from(expected_version >= 2));
    }
}

#[test]
fn rejects_a_newer_schema() {
    let directory = TempDir::new().expect("create a temporary directory");
    let path = database(&directory, "newer.sqlite");
    let connection = Connection::open(&path).expect("open a fixture database");
    connection
        .pragma_update(None, "user_version", 999)
        .expect("set a newer version");
    drop(connection);

    assert!(matches!(
        Store::open(&path),
        Err(StoreError::NewerSchema(999))
    ));
}

#[test]
fn checkpoints_and_vacuums_without_losing_data() {
    let directory = TempDir::new().expect("create a temporary directory");
    let path = database(&directory, "store.sqlite");
    let store = Store::open(&path).expect("open the store");
    store
        .connection()
        .execute(
            "INSERT INTO m1a_parent (id, name, created_at) VALUES (1, 'parent', 1)",
            [],
        )
        .expect("insert a parent");

    store.checkpoint().expect("checkpoint the WAL");
    store.vacuum().expect("vacuum the database");
    store.integrity_check().expect("check database integrity");
    assert_eq!(
        store
            .connection()
            .query_row("SELECT count(*) FROM m1a_parent", [], |row| row
                .get::<_, i64>(0))
            .expect("count parents"),
        1
    );
}

#[test]
fn maintenance_prunes_only_terminal_records_before_the_cutoff() {
    let directory = TempDir::new().expect("create a maintenance fixture directory");
    let mut store =
        Store::open(&database(&directory, "maintenance.sqlite")).expect("open the store");
    store
        .create_initial_administrator(&InitialAdministrator {
            username: "alice",
            canonical_key: "ssh-ed25519 AAAAmaintenance",
            fingerprint: "SHA256:maintenance",
            recovery_hash: &[7; 32],
            created_at: 1,
        })
        .expect("create the account");
    store
        .create_repository(&NewRepository {
            id: "00112233445566778899aabbccddeeff",
            owner: "alice",
            slug: "project",
            object_format: "sha1",
            default_branch: "refs/heads/main",
            created_at: 1,
            origin: RepositoryOrigin::Created,
            initial_references: &[],
            actor: "alice",
            correlation_id: "maintenance-repository",
        })
        .expect("create the repository");
    store
        .connection()
        .execute_batch(
            "DELETE FROM audit_event;
             INSERT INTO signup_invitation
                 (code_hash, created_at, expires_at, consumed_at)
             VALUES
                 (randomblob(32), 1, 2, NULL),
                 (randomblob(32), 99, 100, NULL);
             INSERT INTO login_nonce
                 (nonce_hash, csrf_hash, account_id, ssh_public_key_id,
                  created_at, expires_at, consumed_at)
             VALUES
                 (randomblob(32), randomblob(32), 1, 1, 1, 2, NULL),
                 (randomblob(32), randomblob(32), 1, 1, 99, 100, NULL);
             INSERT INTO ssh_login_approval
                 (secret_hash, csrf_hash, account_id, ssh_public_key_id,
                  created_at, expires_at, approved_at, consumed_at)
             VALUES
                 (randomblob(32), randomblob(32), NULL, NULL, 1, 2, NULL, NULL),
                 (randomblob(32), randomblob(32), NULL, NULL, 99, 100, NULL, NULL);
             INSERT INTO web_session
                 (session_hash, csrf_hash, account_id, created_at, expires_at, ended_at)
             VALUES
                 (randomblob(32), randomblob(32), 1, 1, 2, NULL),
                 (randomblob(32), randomblob(32), 1, 99, 200, NULL);
             INSERT INTO feed_token
                 (id, token_hash, account_id, scope, repository_id, created_at, revoked_at)
             VALUES
                 ('11111111111111111111111111111111', randomblob(32), 1,
                  'repository', '00112233445566778899aabbccddeeff', 1, 2),
                 ('22222222222222222222222222222222', randomblob(32), 1,
                  'repository', '00112233445566778899aabbccddeeff', 99, 100);
             INSERT INTO audit_event
                 (action, actor, target, outcome, correlation_id, created_at)
             VALUES
                 ('old', 'test', 'old', 'success', 'old', 1),
                 ('current', 'test', 'current', 'success', 'current', 100);",
        )
        .expect("create maintenance records");

    let result = store.maintain(100).expect("run maintenance");
    assert_eq!(result.deleted, 6);
    for table in [
        "signup_invitation",
        "login_nonce",
        "ssh_login_approval",
        "web_session",
        "feed_token",
        "audit_event",
    ] {
        let count: i64 = store
            .connection()
            .query_row(&format!("SELECT count(*) FROM {table}"), [], |row| {
                row.get(0)
            })
            .expect("count retained maintenance records");
        assert_eq!(count, 1, "unexpected retained count for {table}");
    }
}
