use crate::model::{Datum, DatumKey, PromptGuess, RatingValueError, RecallFeedback, Review};
use crate::scheduler::{CardMemoryState, CardSchedule};
use rusqlite::{Connection, OptionalExtension, params};
use std::collections::{HashMap, HashSet};
use std::fmt;
use std::path::Path;
use std::time::{Duration, SystemTime, SystemTimeError, UNIX_EPOCH};

pub const CURRENT_SCHEMA_VERSION: i64 = 10;
const DAY_IN_SECONDS: i64 = 24 * 60 * 60;
const DEFERRED_STATE_TYPE: &str = "deferred";
const FLAG_STATE_TYPE: &str = "flag";
const SUSPENDED_STATE_TYPE: &str = "suspended";

pub struct SqliteReviewStore {
    connection: Connection,
}

#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct ScrubReviewDatabaseSummary {
    removed_review_passes: usize,
    removed_review_pass_items: usize,
    removed_review_events: usize,
    removed_card_schedules: usize,
    removed_card_states: usize,
    removed_card_reversals: usize,
    removed_legacy_card_reversal_exclusions: usize,
    removed_card_reversal_batches: usize,
}

impl ScrubReviewDatabaseSummary {
    pub fn total_removed_rows(self) -> usize {
        self.removed_review_passes
            + self.removed_review_pass_items
            + self.removed_review_events
            + self.removed_card_schedules
            + self.removed_card_states
            + self.removed_card_reversals
            + self.removed_legacy_card_reversal_exclusions
            + self.removed_card_reversal_batches
    }
}

impl SqliteReviewStore {
    pub fn open(path: impl AsRef<Path>) -> Result<Self, ReviewStoreError> {
        let connection = Connection::open(path)?;
        connection.busy_timeout(Duration::from_secs(5))?;
        let store = Self { connection };
        store.verify_integrity()?;
        store.initialize()?;
        Ok(store)
    }

    pub fn open_in_memory() -> Result<Self, ReviewStoreError> {
        let connection = Connection::open_in_memory()?;
        connection.busy_timeout(Duration::from_secs(5))?;
        let store = Self { connection };
        store.verify_integrity()?;
        store.initialize()?;
        Ok(store)
    }

    pub fn verify_integrity(&self) -> Result<(), ReviewStoreError> {
        let mut statement = self
            .connection
            .prepare("PRAGMA quick_check")
            .map_err(|error| ReviewStoreError::IntegrityCheckFailed(error.to_string()))?;
        let rows = statement
            .query_map([], |row| row.get::<_, String>(0))
            .map_err(|error| ReviewStoreError::IntegrityCheckFailed(error.to_string()))?;

        for row in rows {
            let result =
                row.map_err(|error| ReviewStoreError::IntegrityCheckFailed(error.to_string()))?;
            if result != "ok" {
                return Err(ReviewStoreError::IntegrityCheckFailed(result));
            }
        }

        Ok(())
    }

    pub fn verify_deck_consistency(&self, data: &[Datum]) -> Result<(), ReviewStoreError> {
        self.verify_review_events_are_typed()?;
        self.verify_card_schedules_are_typed()?;
        self.verify_card_states_are_well_formed()?;
        self.verify_foreign_keys()?;
        self.verify_deck_keys_are_unique(data)?;
        self.verify_active_reversals(data)?;
        Ok(())
    }

    pub fn backup_to_path(&self, path: impl AsRef<Path>) -> Result<(), ReviewStoreError> {
        let path = path.as_ref().display().to_string();
        self.connection.execute("VACUUM main INTO ?1", [path])?;
        Ok(())
    }

    pub fn reclaim_space(&self) -> Result<(), ReviewStoreError> {
        self.connection.execute_batch("VACUUM")?;
        Ok(())
    }

    fn verify_review_events_are_typed(&self) -> Result<(), ReviewStoreError> {
        let mut statement = self.connection.prepare(
            "
            SELECT id, datum_key, reviewed_at_unix_seconds, prompt_guess, recall_feedback
            FROM review_events
            ORDER BY id
            ",
        )?;
        let rows = statement.query_map([], |row| {
            Ok(StoredReviewEvent {
                id: row.get(0)?,
                datum_key: row.get(1)?,
                reviewed_at_unix_seconds: row.get(2)?,
                prompt_guess: row.get(3)?,
                recall_feedback: row.get(4)?,
            })
        })?;

        for row in rows {
            let event: ReviewEvent = row?.try_into()?;
            verify_datum_key_shape("review_events", event.datum_key())?;
        }
        Ok(())
    }

    fn verify_card_schedules_are_typed(&self) -> Result<(), ReviewStoreError> {
        let mut statement = self.connection.prepare(
            "
            SELECT
                datum_key,
                memory_state,
                due_at_unix_seconds,
                stability_days,
                difficulty,
                elapsed_days,
                scheduled_days,
                interval_seconds,
                repetitions,
                lapses,
                last_reviewed_at_unix_seconds,
                last_recall_feedback
            FROM card_schedules
            ORDER BY datum_key
            ",
        )?;
        let rows = statement.query_map([], |row| {
            Ok(StoredCardSchedule {
                datum_key: row.get(0)?,
                memory_state: row.get(1)?,
                due_at_unix_seconds: row.get(2)?,
                stability_days: row.get(3)?,
                difficulty: row.get(4)?,
                elapsed_days: row.get(5)?,
                scheduled_days: row.get(6)?,
                interval_seconds: row.get(7)?,
                repetitions: row.get(8)?,
                lapses: row.get(9)?,
                last_reviewed_at_unix_seconds: row.get(10)?,
                last_recall_feedback: row.get(11)?,
            })
        })?;

        for row in rows {
            let schedule: CardSchedule = row?.try_into()?;
            verify_datum_key_shape("card_schedules", schedule.datum_key())?;
            if schedule.last_reviewed_at_unix_seconds().is_some()
                != schedule.last_recall_feedback().is_some()
            {
                return Err(ReviewStoreError::ConsistencyCheckFailed(format!(
                    "card schedule {} has incomplete last-review metadata",
                    schedule.datum_key()
                )));
            }
        }
        Ok(())
    }

    fn verify_card_states_are_well_formed(&self) -> Result<(), ReviewStoreError> {
        let mut statement = self.connection.prepare(
            "
            SELECT datum_key, state_type, ended_at_unix_seconds
            FROM card_states
            ORDER BY datum_key, state_type
            ",
        )?;
        let rows = statement.query_map([], |row| {
            Ok((
                row.get::<_, String>(0)?,
                row.get::<_, String>(1)?,
                row.get::<_, Option<i64>>(2)?,
            ))
        })?;

        for row in rows {
            let (datum_key, state_type, ended_at_unix_seconds) = row?;
            let datum_key = DatumKey::from_storage_key(&datum_key);
            verify_datum_key_shape("card_states", &datum_key)?;
            if state_type != DEFERRED_STATE_TYPE
                && state_type != FLAG_STATE_TYPE
                && state_type != SUSPENDED_STATE_TYPE
            {
                return Err(ReviewStoreError::ConsistencyCheckFailed(format!(
                    "card_states contains unknown state type {state_type}"
                )));
            }
            if ended_at_unix_seconds.is_some_and(|ended_at| ended_at < 0) {
                return Err(ReviewStoreError::ConsistencyCheckFailed(format!(
                    "card_states contains negative end time for {datum_key}"
                )));
            }
        }
        Ok(())
    }

    fn verify_foreign_keys(&self) -> Result<(), ReviewStoreError> {
        let mut statement = self.connection.prepare(
            "
            PRAGMA foreign_key_check
            ",
        )?;
        let mut rows = statement.query([])?;
        if let Some(row) = rows.next()? {
            let table_name: String = row.get(0)?;
            let row_id: i64 = row.get(1)?;
            let parent_table: String = row.get(2)?;
            return Err(ReviewStoreError::ConsistencyCheckFailed(format!(
                "{table_name} row {row_id} refers to missing {parent_table}"
            )));
        }
        Ok(())
    }

    fn verify_deck_keys_are_unique(&self, data: &[Datum]) -> Result<(), ReviewStoreError> {
        let mut deck_keys = HashSet::new();
        for datum in data {
            let datum_key = DatumKey::for_datum(datum);
            if !deck_keys.insert(datum_key.clone()) {
                return Err(ReviewStoreError::ConsistencyCheckFailed(format!(
                    "deck contains duplicate datum key {datum_key}"
                )));
            }
        }
        Ok(())
    }

    fn verify_active_reversals(&self, data: &[Datum]) -> Result<(), ReviewStoreError> {
        let reversible_keys = self.reversible_datum_keys()?;
        let response_counts = normalized_response_counts(data);
        let deck_keys: HashSet<DatumKey> = data.iter().map(DatumKey::for_datum).collect();

        for datum_key in &reversible_keys {
            verify_datum_key_shape("card_reversals", datum_key)?;
            if datum_key.is_reverse() {
                return Err(ReviewStoreError::ConsistencyCheckFailed(format!(
                    "card_reversals contains generated reverse key {datum_key}; expected parent card key"
                )));
            }
        }

        for datum in data {
            let datum_key = DatumKey::for_datum(datum);
            if !deck_keys.contains(&datum_key) {
                continue;
            }
            if reversible_keys.contains(&datum_key) {
                let response = normalize_reverse_prompt(datum.response());
                if response_counts.get(&response).copied().unwrap_or(0) != 1 {
                    return Err(ReviewStoreError::ConsistencyCheckFailed(format!(
                        "reversible card {datum_key} has a duplicate response"
                    )));
                }
            }
        }
        Ok(())
    }

    pub fn initialize(&self) -> Result<(), ReviewStoreError> {
        self.connection.execute_batch(
            "
            PRAGMA foreign_keys = ON;

            CREATE TABLE IF NOT EXISTS cosimo_schema_version (
                version INTEGER PRIMARY KEY
            );
            ",
        )?;
        self.migrate_schema()?;

        self.connection.execute_batch(
            "
            CREATE TABLE IF NOT EXISTS review_events (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                datum_key TEXT NOT NULL,
                reviewed_at_unix_seconds INTEGER NOT NULL,
                prompt_guess INTEGER NOT NULL CHECK (prompt_guess BETWEEN 0 AND 4),
                recall_feedback INTEGER NOT NULL CHECK (recall_feedback BETWEEN 1 AND 4)
            );

            CREATE INDEX IF NOT EXISTS review_events_datum_time
                ON review_events (datum_key, reviewed_at_unix_seconds, id);

            CREATE TABLE IF NOT EXISTS card_schedules (
                datum_key TEXT PRIMARY KEY,
                memory_state TEXT NOT NULL CHECK (memory_state IN ('new', 'learning', 'review', 'relearning')),
                due_at_unix_seconds INTEGER NOT NULL,
                stability_days REAL NOT NULL CHECK (stability_days >= 0),
                difficulty REAL NOT NULL CHECK (difficulty >= 0),
                elapsed_days INTEGER NOT NULL CHECK (elapsed_days >= 0),
                scheduled_days INTEGER NOT NULL CHECK (scheduled_days >= 0),
                interval_seconds INTEGER NOT NULL CHECK (interval_seconds >= 0),
                repetitions INTEGER NOT NULL CHECK (repetitions >= 0),
                lapses INTEGER NOT NULL CHECK (lapses >= 0),
                last_reviewed_at_unix_seconds INTEGER,
                last_recall_feedback INTEGER CHECK (
                    last_recall_feedback IS NULL OR last_recall_feedback BETWEEN 1 AND 4
                )
            );

            CREATE TABLE IF NOT EXISTS card_states (
                datum_key TEXT NOT NULL,
                state_type TEXT NOT NULL CHECK (state_type IN ('deferred', 'flag', 'suspended')),
                ended_at_unix_seconds INTEGER CHECK (
                    ended_at_unix_seconds IS NULL OR ended_at_unix_seconds >= 0
                ),
                PRIMARY KEY (datum_key, state_type)
            );

            CREATE INDEX IF NOT EXISTS card_states_active_type
                ON card_states (state_type, ended_at_unix_seconds, datum_key);

            CREATE TABLE IF NOT EXISTS card_reversals (
                datum_key TEXT PRIMARY KEY,
                created_by TEXT NOT NULL DEFAULT 'manual' CHECK (created_by IN ('manual', 'bulk')),
                bulk_batch_id INTEGER,
                created_at_unix_seconds INTEGER,
                CHECK ((created_by = 'bulk') = (bulk_batch_id IS NOT NULL))
            );

            CREATE TABLE IF NOT EXISTS card_reversal_batches (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                created_at_unix_seconds INTEGER NOT NULL
            );

            CREATE TABLE IF NOT EXISTS review_passes (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                started_at_unix_seconds INTEGER NOT NULL,
                finished_at_unix_seconds INTEGER NOT NULL,
                was_scheduled INTEGER NOT NULL CHECK (was_scheduled IN (0, 1)),
                ended_early INTEGER NOT NULL CHECK (ended_early IN (0, 1)),
                initial_questions INTEGER NOT NULL CHECK (initial_questions >= 0),
                total_questions INTEGER NOT NULL CHECK (total_questions >= 0),
                unique_cards_reviewed INTEGER NOT NULL CHECK (unique_cards_reviewed >= 0),
                CHECK (finished_at_unix_seconds >= started_at_unix_seconds),
                CHECK (unique_cards_reviewed <= total_questions)
            );

            CREATE INDEX IF NOT EXISTS review_passes_started_at
                ON review_passes (started_at_unix_seconds, id);

            CREATE TABLE IF NOT EXISTS review_pass_items (
                pass_id INTEGER NOT NULL,
                review_event_id INTEGER NOT NULL,
                position INTEGER NOT NULL CHECK (position > 0),
                was_repeat INTEGER NOT NULL CHECK (was_repeat IN (0, 1)),
                PRIMARY KEY (pass_id, position),
                FOREIGN KEY (pass_id) REFERENCES review_passes(id) ON DELETE CASCADE,
                FOREIGN KEY (review_event_id) REFERENCES review_events(id) ON DELETE CASCADE,
                UNIQUE (pass_id, review_event_id)
            );

            CREATE INDEX IF NOT EXISTS review_pass_items_event
                ON review_pass_items (review_event_id);
            ",
        )?;
        self.migrate_card_suspensions_to_card_states()?;
        self.allow_unanswered_prompt_guess()?;

        Ok(())
    }

    fn migrate_schema(&self) -> Result<(), ReviewStoreError> {
        let schema_state = self.schema_state()?;
        match schema_state {
            SchemaState::Uninitialized => {
                self.set_schema_version(CURRENT_SCHEMA_VERSION)?;
            }
            SchemaState::Current => {}
            SchemaState::NeedsMigration(mut version) => {
                self.normalize_schema_version_table()?;
                while version < CURRENT_SCHEMA_VERSION {
                    version = self.apply_schema_migration(version)?;
                    self.set_schema_version(version)?;
                }
            }
            SchemaState::Unsupported(versions) => {
                return Err(ReviewStoreError::UnsupportedSchemaVersion {
                    stored_versions: versions,
                    supported_version: CURRENT_SCHEMA_VERSION,
                });
            }
        }

        self.normalize_schema_version_table()
    }

    fn schema_state(&self) -> Result<SchemaState, ReviewStoreError> {
        let versions = self.schema_versions()?;
        match versions.as_slice() {
            [] => Ok(SchemaState::Uninitialized),
            [version] if *version == CURRENT_SCHEMA_VERSION => Ok(SchemaState::Current),
            [version] if *version < CURRENT_SCHEMA_VERSION => {
                Ok(SchemaState::NeedsMigration(*version))
            }
            _ => Ok(SchemaState::Unsupported(versions)),
        }
    }

    fn apply_schema_migration(&self, version: i64) -> Result<i64, ReviewStoreError> {
        match version {
            1 => {
                self.connection.execute_batch(
                    "
                    CREATE TABLE IF NOT EXISTS card_suspensions (
                        datum_key TEXT PRIMARY KEY
                    );
                    ",
                )?;
                Ok(2)
            }
            2 => {
                self.create_review_pass_tables()?;
                Ok(3)
            }
            3 => {
                self.add_schedule_interval_seconds()?;
                Ok(4)
            }
            4 => {
                self.create_card_reversals_table()?;
                Ok(5)
            }
            5 => {
                self.add_card_reversal_batch_metadata()?;
                Ok(6)
            }
            6 => {
                self.create_card_reversal_exclusions_table()?;
                Ok(7)
            }
            7 => {
                self.create_card_states_table()?;
                Ok(8)
            }
            8 => {
                self.allow_deferred_card_state()?;
                Ok(9)
            }
            9 => {
                self.retire_card_reversal_exclusions_table()?;
                Ok(10)
            }
            _ => Err(ReviewStoreError::UnsupportedSchemaVersion {
                stored_versions: vec![version],
                supported_version: CURRENT_SCHEMA_VERSION,
            }),
        }
    }

    fn create_review_pass_tables(&self) -> Result<(), ReviewStoreError> {
        self.connection.execute_batch(
            "
            CREATE TABLE IF NOT EXISTS review_passes (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                started_at_unix_seconds INTEGER NOT NULL,
                finished_at_unix_seconds INTEGER NOT NULL,
                was_scheduled INTEGER NOT NULL CHECK (was_scheduled IN (0, 1)),
                ended_early INTEGER NOT NULL CHECK (ended_early IN (0, 1)),
                initial_questions INTEGER NOT NULL CHECK (initial_questions >= 0),
                total_questions INTEGER NOT NULL CHECK (total_questions >= 0),
                unique_cards_reviewed INTEGER NOT NULL CHECK (unique_cards_reviewed >= 0),
                CHECK (finished_at_unix_seconds >= started_at_unix_seconds),
                CHECK (unique_cards_reviewed <= total_questions)
            );

            CREATE INDEX IF NOT EXISTS review_passes_started_at
                ON review_passes (started_at_unix_seconds, id);

            CREATE TABLE IF NOT EXISTS review_pass_items (
                pass_id INTEGER NOT NULL,
                review_event_id INTEGER NOT NULL,
                position INTEGER NOT NULL CHECK (position > 0),
                was_repeat INTEGER NOT NULL CHECK (was_repeat IN (0, 1)),
                PRIMARY KEY (pass_id, position),
                FOREIGN KEY (pass_id) REFERENCES review_passes(id) ON DELETE CASCADE,
                FOREIGN KEY (review_event_id) REFERENCES review_events(id) ON DELETE CASCADE,
                UNIQUE (pass_id, review_event_id)
            );

            CREATE INDEX IF NOT EXISTS review_pass_items_event
                ON review_pass_items (review_event_id);
            ",
        )?;
        Ok(())
    }

    fn add_schedule_interval_seconds(&self) -> Result<(), ReviewStoreError> {
        if !self.table_exists("card_schedules")? {
            return Ok(());
        }

        self.connection.execute_batch(
            "
            ALTER TABLE card_schedules
                ADD COLUMN interval_seconds INTEGER NOT NULL DEFAULT 0 CHECK (interval_seconds >= 0);
            ",
        )?;
        self.connection.execute(
            "
            UPDATE card_schedules
            SET interval_seconds = scheduled_days * ?1
            ",
            [DAY_IN_SECONDS],
        )?;
        Ok(())
    }

    fn create_card_reversals_table(&self) -> Result<(), ReviewStoreError> {
        self.connection.execute_batch(
            "
            CREATE TABLE IF NOT EXISTS card_reversals (
                datum_key TEXT PRIMARY KEY,
                created_by TEXT NOT NULL DEFAULT 'manual' CHECK (created_by IN ('manual', 'bulk')),
                bulk_batch_id INTEGER,
                created_at_unix_seconds INTEGER,
                CHECK ((created_by = 'bulk') = (bulk_batch_id IS NOT NULL))
            );

            CREATE TABLE IF NOT EXISTS card_reversal_batches (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                created_at_unix_seconds INTEGER NOT NULL
            );
            ",
        )?;
        Ok(())
    }

    fn add_card_reversal_batch_metadata(&self) -> Result<(), ReviewStoreError> {
        self.connection.execute_batch(
            "
            CREATE TABLE IF NOT EXISTS card_reversal_batches (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                created_at_unix_seconds INTEGER NOT NULL
            );
            ",
        )?;

        if !self.column_exists("card_reversals", "created_by")? {
            self.connection.execute_batch(
                "
                ALTER TABLE card_reversals
                    ADD COLUMN created_by TEXT NOT NULL DEFAULT 'manual' CHECK (created_by IN ('manual', 'bulk'));
                ",
            )?;
        }
        if !self.column_exists("card_reversals", "bulk_batch_id")? {
            self.connection.execute_batch(
                "
                ALTER TABLE card_reversals
                    ADD COLUMN bulk_batch_id INTEGER;
                ",
            )?;
        }
        if !self.column_exists("card_reversals", "created_at_unix_seconds")? {
            self.connection.execute_batch(
                "
                ALTER TABLE card_reversals
                    ADD COLUMN created_at_unix_seconds INTEGER;
                ",
            )?;
        }
        Ok(())
    }

    fn create_card_reversal_exclusions_table(&self) -> Result<(), ReviewStoreError> {
        self.connection.execute_batch(
            "
            CREATE TABLE IF NOT EXISTS card_reversal_exclusions (
                datum_key TEXT PRIMARY KEY
            );
            ",
        )?;
        Ok(())
    }

    fn create_card_states_table(&self) -> Result<(), ReviewStoreError> {
        self.connection.execute_batch(
            "
            CREATE TABLE IF NOT EXISTS card_states (
                datum_key TEXT NOT NULL,
                state_type TEXT NOT NULL CHECK (state_type IN ('deferred', 'flag', 'suspended')),
                ended_at_unix_seconds INTEGER CHECK (
                    ended_at_unix_seconds IS NULL OR ended_at_unix_seconds >= 0
                ),
                PRIMARY KEY (datum_key, state_type)
            );

            CREATE INDEX IF NOT EXISTS card_states_active_type
                ON card_states (state_type, ended_at_unix_seconds, datum_key);
            ",
        )?;
        self.migrate_card_suspensions_to_card_states()?;
        Ok(())
    }

    fn allow_deferred_card_state(&self) -> Result<(), ReviewStoreError> {
        self.connection.execute_batch(
            "
            CREATE TABLE card_states_new (
                datum_key TEXT NOT NULL,
                state_type TEXT NOT NULL CHECK (state_type IN ('deferred', 'flag', 'suspended')),
                ended_at_unix_seconds INTEGER CHECK (
                    ended_at_unix_seconds IS NULL OR ended_at_unix_seconds >= 0
                ),
                PRIMARY KEY (datum_key, state_type)
            );

            INSERT INTO card_states_new (
                datum_key,
                state_type,
                ended_at_unix_seconds
            )
            SELECT
                datum_key,
                state_type,
                ended_at_unix_seconds
            FROM card_states;

            DROP TABLE card_states;
            ALTER TABLE card_states_new RENAME TO card_states;

            CREATE INDEX IF NOT EXISTS card_states_active_type
                ON card_states (state_type, ended_at_unix_seconds, datum_key);
            ",
        )?;
        Ok(())
    }

    fn retire_card_reversal_exclusions_table(&self) -> Result<(), ReviewStoreError> {
        if !self.table_exists("card_reversal_exclusions")? {
            return Ok(());
        }
        if self.table_exists("legacy_card_reversal_exclusions")? {
            self.connection
                .execute_batch("DROP TABLE card_reversal_exclusions;")?;
            return Ok(());
        }

        self.connection.execute_batch(
            "
            ALTER TABLE card_reversal_exclusions
                RENAME TO legacy_card_reversal_exclusions;
            ",
        )?;
        Ok(())
    }

    fn migrate_card_suspensions_to_card_states(&self) -> Result<(), ReviewStoreError> {
        if !self.table_exists("card_suspensions")? {
            return Ok(());
        }

        self.connection.execute(
            "
            INSERT INTO card_states (
                datum_key,
                state_type,
                ended_at_unix_seconds
            )
            SELECT datum_key, ?1, NULL
            FROM card_suspensions
            WHERE 1
            ON CONFLICT(datum_key, state_type) DO UPDATE SET
                ended_at_unix_seconds = NULL
            ",
            [SUSPENDED_STATE_TYPE],
        )?;
        self.connection
            .execute_batch("DROP TABLE card_suspensions;")?;
        Ok(())
    }

    fn table_exists(&self, table_name: &str) -> Result<bool, ReviewStoreError> {
        Ok(self
            .connection
            .query_row(
                "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?1",
                [table_name],
                |_| Ok(()),
            )
            .optional()?
            .is_some())
    }

    fn column_exists(&self, table_name: &str, column_name: &str) -> Result<bool, ReviewStoreError> {
        let mut statement = self
            .connection
            .prepare("SELECT name FROM pragma_table_info(?1)")?;
        let rows = statement.query_map([table_name], |row| row.get::<_, String>(0))?;
        for row in rows {
            if row? == column_name {
                return Ok(true);
            }
        }
        Ok(false)
    }

    fn set_schema_version(&self, version: i64) -> Result<(), ReviewStoreError> {
        self.connection.execute_batch("BEGIN IMMEDIATE")?;
        let result: Result<(), ReviewStoreError> = (|| {
            self.connection
                .execute("DELETE FROM cosimo_schema_version", [])?;
            self.connection.execute(
                "INSERT INTO cosimo_schema_version (version) VALUES (?1)",
                [version],
            )?;
            Ok(())
        })();

        match result {
            Ok(()) => {
                self.connection.execute_batch("COMMIT")?;
                Ok(())
            }
            Err(error) => {
                let _ = self.connection.execute_batch("ROLLBACK");
                Err(error)
            }
        }
    }

    fn normalize_schema_version_table(&self) -> Result<(), ReviewStoreError> {
        let Some(create_sql) = self.schema_version_table_sql()? else {
            return Ok(());
        };
        if !create_sql.contains("CHECK") {
            return Ok(());
        }

        self.connection.execute_batch("BEGIN IMMEDIATE")?;
        let result = self.connection.execute_batch(
            "
                CREATE TABLE cosimo_schema_version_new (
                    version INTEGER PRIMARY KEY
                );

                INSERT INTO cosimo_schema_version_new (version)
                SELECT version FROM cosimo_schema_version;

                DROP TABLE cosimo_schema_version;
                ALTER TABLE cosimo_schema_version_new RENAME TO cosimo_schema_version;
            ",
        );

        match result {
            Ok(()) => {
                self.connection.execute_batch("COMMIT")?;
                Ok(())
            }
            Err(error) => {
                let _ = self.connection.execute_batch("ROLLBACK");
                Err(error.into())
            }
        }
    }

    fn schema_version_table_sql(&self) -> Result<Option<String>, ReviewStoreError> {
        Ok(self
            .connection
            .query_row(
                "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'cosimo_schema_version'",
                [],
                |row| row.get(0),
            )
            .optional()?)
    }

    fn schema_versions(&self) -> Result<Vec<i64>, ReviewStoreError> {
        let mut statement = self
            .connection
            .prepare("SELECT version FROM cosimo_schema_version ORDER BY version")?;
        let rows = statement.query_map([], |row| row.get::<_, i64>(0))?;

        let mut versions = Vec::new();
        for row in rows {
            versions.push(row?);
        }
        Ok(versions)
    }

    fn allow_unanswered_prompt_guess(&self) -> Result<(), ReviewStoreError> {
        let create_sql: Option<String> = self.connection.query_row(
            "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'review_events'",
            [],
            |row| row.get(0),
        )?;
        if create_sql
            .as_deref()
            .is_some_and(|sql| sql.contains("prompt_guess BETWEEN 0 AND 4"))
        {
            return Ok(());
        }

        self.connection.execute_batch(
            "
            BEGIN IMMEDIATE;

            CREATE TABLE review_events_new (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                datum_key TEXT NOT NULL,
                reviewed_at_unix_seconds INTEGER NOT NULL,
                prompt_guess INTEGER NOT NULL CHECK (prompt_guess BETWEEN 0 AND 4),
                recall_feedback INTEGER NOT NULL CHECK (recall_feedback BETWEEN 1 AND 4)
            );

            INSERT INTO review_events_new (
                id,
                datum_key,
                reviewed_at_unix_seconds,
                prompt_guess,
                recall_feedback
            )
            SELECT
                id,
                datum_key,
                reviewed_at_unix_seconds,
                prompt_guess,
                recall_feedback
            FROM review_events;

            DROP TABLE review_events;
            ALTER TABLE review_events_new RENAME TO review_events;

            CREATE INDEX review_events_datum_time
                ON review_events (datum_key, reviewed_at_unix_seconds, id);

            COMMIT;
            ",
        )?;

        Ok(())
    }

    pub fn record_review_now(
        &self,
        datum: &Datum,
        review: Review,
    ) -> Result<ReviewEvent, ReviewStoreError> {
        self.record_review_at(datum, review, unix_seconds_now()?)
    }

    pub fn record_review_at(
        &self,
        datum: &Datum,
        review: Review,
        reviewed_at_unix_seconds: i64,
    ) -> Result<ReviewEvent, ReviewStoreError> {
        let datum_key = DatumKey::for_datum(datum);
        self.record_review_for_datum_key_at(&datum_key, review, reviewed_at_unix_seconds)
    }

    pub fn record_review_for_datum_key_at(
        &self,
        datum_key: &DatumKey,
        review: Review,
        reviewed_at_unix_seconds: i64,
    ) -> Result<ReviewEvent, ReviewStoreError> {
        let current_schedule = self
            .schedule_for_datum_key(datum_key)?
            .unwrap_or_else(|| CardSchedule::new(datum_key.clone()));
        let next_schedule = current_schedule.reviewed(&review, reviewed_at_unix_seconds);

        self.connection.execute_batch("BEGIN IMMEDIATE")?;
        let result = (|| {
            self.connection.execute(
                "
                INSERT INTO review_events (
                    datum_key,
                    reviewed_at_unix_seconds,
                    prompt_guess,
                    recall_feedback
                ) VALUES (?1, ?2, ?3, ?4)
                ",
                params![
                    datum_key.as_str(),
                    reviewed_at_unix_seconds,
                    review.prompt_guess().value(),
                    review.recall_feedback().value(),
                ],
            )?;
            let review_id = self.connection.last_insert_rowid();

            self.save_card_schedule_in_current_transaction(&next_schedule)?;

            Ok(review_id)
        })();

        let review_id = match result {
            Ok(review_id) => {
                self.connection.execute_batch("COMMIT")?;
                review_id
            }
            Err(error) => {
                let _ = self.connection.execute_batch("ROLLBACK");
                return Err(error);
            }
        };

        Ok(ReviewEvent::new(
            review_id,
            datum_key.clone(),
            reviewed_at_unix_seconds,
            review,
        ))
    }

    pub fn record_review_for_datum_key_now(
        &self,
        datum_key: &DatumKey,
        review: Review,
    ) -> Result<ReviewEvent, ReviewStoreError> {
        self.record_review_for_datum_key_at(datum_key, review, unix_seconds_now()?)
    }

    pub fn record_review_pass(&self, pass: &ReviewPass) -> Result<i64, ReviewStoreError> {
        self.connection.execute_batch("BEGIN IMMEDIATE")?;
        let result: Result<i64, ReviewStoreError> = (|| {
            self.connection.execute(
                "
                INSERT INTO review_passes (
                    started_at_unix_seconds,
                    finished_at_unix_seconds,
                    was_scheduled,
                    ended_early,
                    initial_questions,
                    total_questions,
                    unique_cards_reviewed
                ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)
                ",
                params![
                    pass.started_at_unix_seconds(),
                    pass.finished_at_unix_seconds(),
                    bool_to_storage(pass.was_scheduled()),
                    bool_to_storage(pass.ended_early()),
                    pass.initial_questions() as i64,
                    pass.total_questions() as i64,
                    pass.unique_cards_reviewed() as i64,
                ],
            )?;
            let pass_id = self.connection.last_insert_rowid();

            for item in pass.items() {
                self.connection.execute(
                    "
                    INSERT INTO review_pass_items (
                        pass_id,
                        review_event_id,
                        position,
                        was_repeat
                    ) VALUES (?1, ?2, ?3, ?4)
                    ",
                    params![
                        pass_id,
                        item.review_event_id(),
                        item.position() as i64,
                        bool_to_storage(item.was_repeat()),
                    ],
                )?;
            }

            Ok(pass_id)
        })();

        match result {
            Ok(pass_id) => {
                self.connection.execute_batch("COMMIT")?;
                Ok(pass_id)
            }
            Err(error) => {
                let _ = self.connection.execute_batch("ROLLBACK");
                Err(error)
            }
        }
    }

    pub fn review_pass_summaries_for_datum_keys(
        &self,
        datum_keys: &HashSet<DatumKey>,
    ) -> Result<Vec<ReviewPassSummary>, ReviewStoreError> {
        if datum_keys.is_empty() {
            return Ok(Vec::new());
        }

        let mut pass_statement = self.connection.prepare(
            "
            SELECT
                id,
                started_at_unix_seconds,
                finished_at_unix_seconds,
                was_scheduled,
                ended_early,
                initial_questions,
                total_questions,
                unique_cards_reviewed
            FROM review_passes
            ORDER BY started_at_unix_seconds, id
            ",
        )?;
        let pass_rows = pass_statement.query_map([], |row| {
            Ok((
                row.get::<_, i64>(0)?,
                ReviewPassSummary::new(
                    row.get::<_, i64>(1)?,
                    row.get::<_, i64>(2)?,
                    storage_to_bool(row.get::<_, i64>(3)?),
                    storage_to_bool(row.get::<_, i64>(4)?),
                    row.get::<_, i64>(5)? as usize,
                    row.get::<_, i64>(6)? as usize,
                    row.get::<_, i64>(7)? as usize,
                ),
            ))
        })?;
        let mut passes = Vec::new();
        for row in pass_rows {
            passes.push(row?);
        }

        let mut matching_pass_ids = HashSet::new();
        let mut item_statement = self.connection.prepare(
            "
            SELECT review_pass_items.pass_id, review_events.datum_key
            FROM review_pass_items
            JOIN review_events
              ON review_events.id = review_pass_items.review_event_id
            ",
        )?;
        let item_rows = item_statement.query_map([], |row| {
            Ok((
                row.get::<_, i64>(0)?,
                DatumKey::from_storage_key(row.get::<_, String>(1)?),
            ))
        })?;
        for row in item_rows {
            let (pass_id, datum_key) = row?;
            if datum_keys.contains(&datum_key) {
                matching_pass_ids.insert(pass_id);
            }
        }

        Ok(passes
            .into_iter()
            .filter_map(|(pass_id, summary)| {
                matching_pass_ids.contains(&pass_id).then_some(summary)
            })
            .collect())
    }

    fn save_card_schedule_in_current_transaction(
        &self,
        schedule: &CardSchedule,
    ) -> Result<(), ReviewStoreError> {
        self.connection.execute(
            "
            INSERT INTO card_schedules (
                datum_key,
                memory_state,
                due_at_unix_seconds,
                stability_days,
                difficulty,
                elapsed_days,
                scheduled_days,
                interval_seconds,
                repetitions,
                lapses,
                last_reviewed_at_unix_seconds,
                last_recall_feedback
            ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)
            ON CONFLICT(datum_key) DO UPDATE SET
                memory_state = excluded.memory_state,
                due_at_unix_seconds = excluded.due_at_unix_seconds,
                stability_days = excluded.stability_days,
                difficulty = excluded.difficulty,
                elapsed_days = excluded.elapsed_days,
                scheduled_days = excluded.scheduled_days,
                interval_seconds = excluded.interval_seconds,
                repetitions = excluded.repetitions,
                lapses = excluded.lapses,
                last_reviewed_at_unix_seconds = excluded.last_reviewed_at_unix_seconds,
                last_recall_feedback = excluded.last_recall_feedback
            ",
            params![
                schedule.datum_key().as_str(),
                schedule.memory_state().as_str(),
                schedule.due_at_unix_seconds(),
                schedule.stability_days(),
                schedule.difficulty(),
                schedule.elapsed_days(),
                schedule.scheduled_days(),
                schedule.interval_seconds(),
                schedule.repetitions(),
                schedule.lapses(),
                schedule.last_reviewed_at_unix_seconds(),
                schedule.last_recall_feedback().map(RecallFeedback::value),
            ],
        )?;

        Ok(())
    }

    pub fn reviews_for_datum(&self, datum: &Datum) -> Result<Vec<ReviewEvent>, ReviewStoreError> {
        self.reviews_for_datum_key(&DatumKey::for_datum(datum))
    }

    pub fn reviews_for_datum_key(
        &self,
        datum_key: &DatumKey,
    ) -> Result<Vec<ReviewEvent>, ReviewStoreError> {
        let mut statement = self.connection.prepare(
            "
            SELECT id, datum_key, reviewed_at_unix_seconds, prompt_guess, recall_feedback
            FROM review_events
            WHERE datum_key = ?1
            ORDER BY reviewed_at_unix_seconds, id
            ",
        )?;

        let rows = statement.query_map(params![datum_key.as_str()], |row| {
            Ok(StoredReviewEvent {
                id: row.get(0)?,
                datum_key: row.get(1)?,
                reviewed_at_unix_seconds: row.get(2)?,
                prompt_guess: row.get(3)?,
                recall_feedback: row.get(4)?,
            })
        })?;

        let mut events = Vec::new();
        for row in rows {
            events.push(row?.try_into()?);
        }

        Ok(events)
    }

    pub fn schedule_for_datum(
        &self,
        datum: &Datum,
    ) -> Result<Option<CardSchedule>, ReviewStoreError> {
        self.schedule_for_datum_key(&DatumKey::for_datum(datum))
    }

    pub fn schedule_for_datum_key(
        &self,
        datum_key: &DatumKey,
    ) -> Result<Option<CardSchedule>, ReviewStoreError> {
        let mut statement = self.connection.prepare(
            "
            SELECT
                datum_key,
                memory_state,
                due_at_unix_seconds,
                stability_days,
                difficulty,
                elapsed_days,
                scheduled_days,
                interval_seconds,
                repetitions,
                lapses,
                last_reviewed_at_unix_seconds,
                last_recall_feedback
            FROM card_schedules
            WHERE datum_key = ?1
            ",
        )?;

        let mut rows = statement.query_map(params![datum_key.as_str()], |row| {
            Ok(StoredCardSchedule {
                datum_key: row.get(0)?,
                memory_state: row.get(1)?,
                due_at_unix_seconds: row.get(2)?,
                stability_days: row.get(3)?,
                difficulty: row.get(4)?,
                elapsed_days: row.get(5)?,
                scheduled_days: row.get(6)?,
                interval_seconds: row.get(7)?,
                repetitions: row.get(8)?,
                lapses: row.get(9)?,
                last_reviewed_at_unix_seconds: row.get(10)?,
                last_recall_feedback: row.get(11)?,
            })
        })?;

        let Some(row) = rows.next() else {
            return Ok(None);
        };

        Ok(Some(row?.try_into()?))
    }

    pub fn save_card_schedule(&self, schedule: &CardSchedule) -> Result<(), ReviewStoreError> {
        self.save_card_schedule_in_current_transaction(schedule)
    }

    pub fn suspended_datum_keys(&self) -> Result<HashSet<DatumKey>, ReviewStoreError> {
        self.active_state_datum_keys(SUSPENDED_STATE_TYPE)
    }

    pub fn set_datum_suspended(
        &self,
        datum_key: &DatumKey,
        suspended: bool,
    ) -> Result<(), ReviewStoreError> {
        self.set_datum_state_active(datum_key, SUSPENDED_STATE_TYPE, suspended)
    }

    pub fn flagged_datum_keys(&self) -> Result<HashSet<DatumKey>, ReviewStoreError> {
        self.active_state_datum_keys(FLAG_STATE_TYPE)
    }

    pub fn deferred_datum_keys(
        &self,
        now_unix_seconds: i64,
    ) -> Result<HashSet<DatumKey>, ReviewStoreError> {
        let mut statement = self.connection.prepare(
            "
            SELECT datum_key
            FROM card_states
            WHERE state_type = ?1
              AND ended_at_unix_seconds IS NOT NULL
              AND ended_at_unix_seconds > ?2
            ",
        )?;
        let rows = statement.query_map(params![DEFERRED_STATE_TYPE, now_unix_seconds], |row| {
            row.get::<_, String>(0)
        })?;

        let mut datum_keys = HashSet::new();
        for row in rows {
            datum_keys.insert(DatumKey::from_storage_key(row?));
        }
        Ok(datum_keys)
    }

    pub fn defer_datum_until(
        &self,
        datum_key: &DatumKey,
        until_unix_seconds: i64,
    ) -> Result<(), ReviewStoreError> {
        self.connection.execute(
            "
            INSERT INTO card_states (
                datum_key,
                state_type,
                ended_at_unix_seconds
            ) VALUES (?1, ?2, ?3)
            ON CONFLICT(datum_key, state_type) DO UPDATE SET
                ended_at_unix_seconds = ?3
            ",
            params![datum_key.as_str(), DEFERRED_STATE_TYPE, until_unix_seconds],
        )?;
        Ok(())
    }

    pub fn clear_datum_deferred(&self, datum_key: &DatumKey) -> Result<(), ReviewStoreError> {
        self.connection.execute(
            "
            UPDATE card_states
            SET ended_at_unix_seconds = ?1
            WHERE datum_key = ?2
              AND state_type = ?3
              AND ended_at_unix_seconds IS NOT NULL
              AND ended_at_unix_seconds > ?1
            ",
            params![unix_seconds_now()?, datum_key.as_str(), DEFERRED_STATE_TYPE],
        )?;
        Ok(())
    }

    pub fn clear_inactive_card_states(
        &self,
        now_unix_seconds: i64,
    ) -> Result<usize, ReviewStoreError> {
        Ok(self.connection.execute(
            "
            DELETE FROM card_states
            WHERE ended_at_unix_seconds IS NOT NULL
              AND (
                    state_type != ?1
                    OR ended_at_unix_seconds <= ?2
              )
            ",
            params![DEFERRED_STATE_TYPE, now_unix_seconds],
        )?)
    }

    fn active_state_datum_keys(
        &self,
        state_type: &str,
    ) -> Result<HashSet<DatumKey>, ReviewStoreError> {
        let mut statement = self.connection.prepare(
            "
            SELECT datum_key
            FROM card_states
            WHERE state_type = ?1
              AND ended_at_unix_seconds IS NULL
            ",
        )?;
        let rows = statement.query_map([state_type], |row| row.get::<_, String>(0))?;

        let mut datum_keys = HashSet::new();
        for row in rows {
            datum_keys.insert(DatumKey::from_storage_key(row?));
        }
        Ok(datum_keys)
    }

    pub fn set_datum_flagged(
        &self,
        datum_key: &DatumKey,
        flagged: bool,
    ) -> Result<(), ReviewStoreError> {
        self.set_datum_state_active(datum_key, FLAG_STATE_TYPE, flagged)
    }

    fn set_datum_state_active(
        &self,
        datum_key: &DatumKey,
        state_type: &str,
        active: bool,
    ) -> Result<(), ReviewStoreError> {
        if active {
            self.connection.execute(
                "
                INSERT INTO card_states (
                    datum_key,
                    state_type,
                    ended_at_unix_seconds
                ) VALUES (?1, ?2, NULL)
                ON CONFLICT(datum_key, state_type) DO UPDATE SET
                    ended_at_unix_seconds = NULL
                ",
                params![datum_key.as_str(), state_type],
            )?;
        } else {
            self.connection.execute(
                "
                UPDATE card_states
                SET ended_at_unix_seconds = ?1
                WHERE datum_key = ?2
                  AND state_type = ?3
                  AND ended_at_unix_seconds IS NULL
                ",
                params![unix_seconds_now()?, datum_key.as_str(), state_type],
            )?;
        }
        Ok(())
    }

    pub fn reversible_datum_keys(&self) -> Result<HashSet<DatumKey>, ReviewStoreError> {
        let mut statement = self
            .connection
            .prepare("SELECT datum_key FROM card_reversals")?;
        let rows = statement.query_map([], |row| row.get::<_, String>(0))?;

        let mut datum_keys = HashSet::new();
        for row in rows {
            datum_keys.insert(DatumKey::from_storage_key(row?));
        }
        Ok(datum_keys)
    }

    pub fn legacy_reversal_excluded_datum_keys(
        &self,
    ) -> Result<HashSet<DatumKey>, ReviewStoreError> {
        if !self.table_exists("legacy_card_reversal_exclusions")? {
            return Ok(HashSet::new());
        }
        let mut statement = self
            .connection
            .prepare("SELECT datum_key FROM legacy_card_reversal_exclusions")?;
        let rows = statement.query_map([], |row| row.get::<_, String>(0))?;

        let mut datum_keys = HashSet::new();
        for row in rows {
            datum_keys.insert(DatumKey::from_storage_key(row?));
        }
        Ok(datum_keys)
    }

    pub fn clear_legacy_reversal_exclusions(&self) -> Result<(), ReviewStoreError> {
        if self.table_exists("legacy_card_reversal_exclusions")? {
            self.connection
                .execute_batch("DROP TABLE legacy_card_reversal_exclusions;")?;
        }
        Ok(())
    }

    pub fn set_datum_reversible(
        &self,
        datum_key: &DatumKey,
        reversible: bool,
    ) -> Result<(), ReviewStoreError> {
        self.connection
            .execute_batch("BEGIN IMMEDIATE TRANSACTION")?;
        let result: Result<(), ReviewStoreError> = (|| {
            if reversible {
                self.connection.execute(
                    "
                    INSERT OR IGNORE INTO card_reversals (
                        datum_key,
                        created_by,
                        bulk_batch_id,
                        created_at_unix_seconds
                    ) VALUES (?1, 'manual', NULL, NULL)
                    ",
                    params![datum_key.as_str()],
                )?;
            } else {
                self.connection.execute(
                    "DELETE FROM card_reversals WHERE datum_key = ?1",
                    params![datum_key.as_str()],
                )?;
            }
            Ok(())
        })();

        match result {
            Ok(()) => {
                self.connection.execute_batch("COMMIT")?;
                Ok(())
            }
            Err(error) => {
                let _ = self.connection.execute_batch("ROLLBACK");
                Err(error)
            }
        }
    }

    pub fn set_datums_reversible<'a>(
        &self,
        datum_keys: impl IntoIterator<Item = &'a DatumKey>,
        reversible: bool,
    ) -> Result<(), ReviewStoreError> {
        self.connection
            .execute_batch("BEGIN IMMEDIATE TRANSACTION")?;
        let result = (|| {
            let sql = if reversible {
                "
                INSERT OR IGNORE INTO card_reversals (
                    datum_key,
                    created_by,
                    bulk_batch_id,
                    created_at_unix_seconds
                ) VALUES (?1, 'manual', NULL, NULL)
                "
            } else {
                "DELETE FROM card_reversals WHERE datum_key = ?1"
            };
            let mut statement = self.connection.prepare(sql)?;
            for datum_key in datum_keys {
                statement.execute(params![datum_key.as_str()])?;
            }
            Ok(())
        })();

        match result {
            Ok(()) => {
                self.connection.execute_batch("COMMIT")?;
                Ok(())
            }
            Err(error) => {
                let _ = self.connection.execute_batch("ROLLBACK");
                Err(error)
            }
        }
    }

    pub fn set_datums_reversible_in_bulk<'a>(
        &self,
        datum_keys: impl IntoIterator<Item = &'a DatumKey>,
    ) -> Result<(), ReviewStoreError> {
        self.connection
            .execute_batch("BEGIN IMMEDIATE TRANSACTION")?;
        let result: Result<(), ReviewStoreError> = (|| {
            let created_at_unix_seconds = unix_seconds_now()?;
            self.connection.execute(
                "
                INSERT INTO card_reversal_batches (created_at_unix_seconds)
                VALUES (?1)
                ",
                [created_at_unix_seconds],
            )?;
            let batch_id = self.connection.last_insert_rowid();
            let mut statement = self.connection.prepare(
                "
                INSERT OR IGNORE INTO card_reversals (
                    datum_key,
                    created_by,
                    bulk_batch_id,
                    created_at_unix_seconds
                )
                VALUES (?1, 'bulk', ?2, ?3)
                ",
            )?;
            for datum_key in datum_keys {
                statement.execute(params![
                    datum_key.as_str(),
                    batch_id,
                    created_at_unix_seconds
                ])?;
            }
            Ok(())
        })();

        match result {
            Ok(()) => {
                self.connection.execute_batch("COMMIT")?;
                Ok(())
            }
            Err(error) => {
                let _ = self.connection.execute_batch("ROLLBACK");
                Err(error)
            }
        }
    }

    pub fn latest_bulk_reversal_count(&self) -> Result<usize, ReviewStoreError> {
        Ok(self.connection.query_row(
            "
            SELECT COUNT(*)
            FROM card_reversals
            WHERE created_by = 'bulk'
              AND bulk_batch_id = (
                  SELECT MAX(bulk_batch_id)
                  FROM card_reversals
                  WHERE created_by = 'bulk'
              )
            ",
            [],
            |row| row.get::<_, i64>(0),
        )? as usize)
    }

    pub fn remove_latest_bulk_reversal(&self) -> Result<usize, ReviewStoreError> {
        self.connection
            .execute_batch("BEGIN IMMEDIATE TRANSACTION")?;
        let result: Result<usize, ReviewStoreError> = (|| {
            let batch_id = self
                .connection
                .query_row(
                    "
                    SELECT MAX(bulk_batch_id)
                    FROM card_reversals
                    WHERE created_by = 'bulk'
                    ",
                    [],
                    |row| row.get::<_, Option<i64>>(0),
                )?
                .ok_or_else(|| {
                    ReviewStoreError::ConsistencyCheckFailed(
                        "no bulk reversal batch is available to undo".to_string(),
                    )
                })?;
            let removed = self.connection.execute(
                "
                DELETE FROM card_reversals
                WHERE created_by = 'bulk'
                  AND bulk_batch_id = ?1
                ",
                [batch_id],
            )?;
            Ok(removed)
        })();

        match result {
            Ok(removed) => {
                self.connection.execute_batch("COMMIT")?;
                Ok(removed)
            }
            Err(error) => {
                let _ = self.connection.execute_batch("ROLLBACK");
                Err(error)
            }
        }
    }

    pub fn scrub_inactive_data<'a>(
        &self,
        active_datum_keys: impl IntoIterator<Item = &'a DatumKey>,
        active_parent_datum_keys: impl IntoIterator<Item = &'a DatumKey>,
    ) -> Result<ScrubReviewDatabaseSummary, ReviewStoreError> {
        self.connection
            .execute_batch("BEGIN IMMEDIATE TRANSACTION")?;
        let result: Result<ScrubReviewDatabaseSummary, ReviewStoreError> = (|| {
            self.connection.execute_batch(
                "
                CREATE TEMP TABLE IF NOT EXISTS cosimo_active_datum_keys (
                    datum_key TEXT PRIMARY KEY
                );
                CREATE TEMP TABLE IF NOT EXISTS cosimo_active_parent_datum_keys (
                    datum_key TEXT PRIMARY KEY
                );
                DELETE FROM cosimo_active_datum_keys;
                DELETE FROM cosimo_active_parent_datum_keys;
                ",
            )?;

            {
                let mut statement = self.connection.prepare(
                    "
                    INSERT OR IGNORE INTO cosimo_active_datum_keys (datum_key)
                    VALUES (?1)
                    ",
                )?;
                for datum_key in active_datum_keys {
                    statement.execute(params![datum_key.as_str()])?;
                }
            }

            {
                let mut statement = self.connection.prepare(
                    "
                    INSERT OR IGNORE INTO cosimo_active_parent_datum_keys (datum_key)
                    VALUES (?1)
                    ",
                )?;
                for datum_key in active_parent_datum_keys {
                    statement.execute(params![datum_key.as_str()])?;
                }
            }

            let removed_review_pass_items = self.connection.query_row(
                "
                SELECT COUNT(*)
                FROM review_pass_items
                WHERE pass_id IN (
                    SELECT DISTINCT review_pass_items.pass_id
                    FROM review_pass_items
                    JOIN review_events
                      ON review_events.id = review_pass_items.review_event_id
                    WHERE NOT EXISTS (
                        SELECT 1
                        FROM cosimo_active_datum_keys
                        WHERE cosimo_active_datum_keys.datum_key = review_events.datum_key
                    )
                )
                ",
                [],
                |row| row.get::<_, i64>(0),
            )? as usize;
            let mut removed_review_passes = self.connection.execute(
                "
                DELETE FROM review_passes
                WHERE id IN (
                    SELECT DISTINCT review_pass_items.pass_id
                    FROM review_pass_items
                    JOIN review_events
                      ON review_events.id = review_pass_items.review_event_id
                    WHERE NOT EXISTS (
                        SELECT 1
                        FROM cosimo_active_datum_keys
                        WHERE cosimo_active_datum_keys.datum_key = review_events.datum_key
                    )
                )
                ",
                [],
            )?;
            let removed_review_events = self.connection.execute(
                "
                DELETE FROM review_events
                WHERE NOT EXISTS (
                    SELECT 1
                    FROM cosimo_active_datum_keys
                    WHERE cosimo_active_datum_keys.datum_key = review_events.datum_key
                )
                ",
                [],
            )?;
            let removed_card_schedules = self.connection.execute(
                "
                DELETE FROM card_schedules
                WHERE NOT EXISTS (
                    SELECT 1
                    FROM cosimo_active_datum_keys
                    WHERE cosimo_active_datum_keys.datum_key = card_schedules.datum_key
                )
                ",
                [],
            )?;
            let removed_card_states = self.connection.execute(
                "
                DELETE FROM card_states
                WHERE (
                        ended_at_unix_seconds IS NOT NULL
                        AND (
                            state_type != 'deferred'
                            OR ended_at_unix_seconds <= CAST(strftime('%s', 'now') AS INTEGER)
                        )
                    )
                   OR NOT EXISTS (
                        SELECT 1
                        FROM cosimo_active_datum_keys
                        WHERE cosimo_active_datum_keys.datum_key = card_states.datum_key
                    )
                ",
                [],
            )?;
            let removed_card_reversals = self.connection.execute(
                "
                DELETE FROM card_reversals
                WHERE NOT EXISTS (
                    SELECT 1
                    FROM cosimo_active_parent_datum_keys
                    WHERE cosimo_active_parent_datum_keys.datum_key = card_reversals.datum_key
                )
                ",
                [],
            )?;
            let removed_legacy_card_reversal_exclusions = if self
                .table_exists("legacy_card_reversal_exclusions")?
            {
                self.connection.execute(
                        "
                        DELETE FROM legacy_card_reversal_exclusions
                        WHERE NOT EXISTS (
                            SELECT 1
                            FROM cosimo_active_parent_datum_keys
                            WHERE cosimo_active_parent_datum_keys.datum_key = legacy_card_reversal_exclusions.datum_key
                        )
                        ",
                        [],
                    )?
            } else {
                0
            };
            let removed_card_reversal_batches = self.connection.execute(
                "
                DELETE FROM card_reversal_batches
                WHERE NOT EXISTS (
                    SELECT 1
                    FROM card_reversals
                    WHERE card_reversals.bulk_batch_id = card_reversal_batches.id
                )
                ",
                [],
            )?;
            removed_review_passes += self.connection.execute(
                "
                DELETE FROM review_passes
                WHERE NOT EXISTS (
                    SELECT 1
                    FROM review_pass_items
                    WHERE review_pass_items.pass_id = review_passes.id
                )
                ",
                [],
            )?;
            self.connection.execute_batch(
                "
                DROP TABLE cosimo_active_datum_keys;
                DROP TABLE cosimo_active_parent_datum_keys;
                ",
            )?;

            Ok(ScrubReviewDatabaseSummary {
                removed_review_passes,
                removed_review_pass_items,
                removed_review_events,
                removed_card_schedules,
                removed_card_states,
                removed_card_reversals,
                removed_legacy_card_reversal_exclusions,
                removed_card_reversal_batches,
            })
        })();

        match result {
            Ok(summary) => {
                self.connection.execute_batch("COMMIT")?;
                self.connection.execute_batch("VACUUM")?;
                Ok(summary)
            }
            Err(error) => {
                let _ = self.connection.execute_batch("ROLLBACK");
                Err(error)
            }
        }
    }

    pub fn used_card_ids(&self) -> Result<Vec<u64>, ReviewStoreError> {
        let mut card_ids = Vec::new();
        for table_name in [
            "review_events",
            "card_schedules",
            "card_states",
            "card_reversals",
            "legacy_card_reversal_exclusions",
        ] {
            if !self.table_exists(table_name)? {
                continue;
            }
            let sql = format!("SELECT datum_key FROM {table_name}");
            let mut statement = self.connection.prepare(&sql)?;
            let rows = statement.query_map([], |row| row.get::<_, String>(0))?;
            for row in rows {
                let datum_key = DatumKey::from_storage_key(row?);
                if let Some(card_id) = datum_key.card_id() {
                    card_ids.push(card_id);
                }
            }
        }

        card_ids.sort_unstable();
        card_ids.dedup();
        Ok(card_ids)
    }

    pub fn migrate_datum_key(
        &self,
        old_key: &DatumKey,
        new_key: &DatumKey,
    ) -> Result<usize, ReviewStoreError> {
        if old_key == new_key {
            return Ok(0);
        }

        self.connection.execute_batch("BEGIN IMMEDIATE")?;
        let result = (|| {
            let updated = self.connection.execute(
                "UPDATE review_events SET datum_key = ?1 WHERE datum_key = ?2",
                params![new_key.as_str(), old_key.as_str()],
            )?;
            self.connection.execute(
                "UPDATE card_schedules SET datum_key = ?1 WHERE datum_key = ?2",
                params![new_key.as_str(), old_key.as_str()],
            )?;
            self.connection.execute(
                "UPDATE card_states SET datum_key = ?1 WHERE datum_key = ?2",
                params![new_key.as_str(), old_key.as_str()],
            )?;
            self.connection.execute(
                "UPDATE card_reversals SET datum_key = ?1 WHERE datum_key = ?2",
                params![new_key.as_str(), old_key.as_str()],
            )?;
            if self.table_exists("legacy_card_reversal_exclusions")? {
                self.connection.execute(
                    "UPDATE legacy_card_reversal_exclusions SET datum_key = ?1 WHERE datum_key = ?2",
                    params![new_key.as_str(), old_key.as_str()],
                )?;
            }
            Ok(updated)
        })();

        let updated = match result {
            Ok(updated) => {
                self.connection.execute_batch("COMMIT")?;
                updated
            }
            Err(error) => {
                let _ = self.connection.execute_batch("ROLLBACK");
                return Err(error);
            }
        };
        Ok(updated)
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ReviewEvent {
    id: i64,
    datum_key: DatumKey,
    reviewed_at_unix_seconds: i64,
    review: Review,
}

impl ReviewEvent {
    pub fn new(
        id: i64,
        datum_key: DatumKey,
        reviewed_at_unix_seconds: i64,
        review: Review,
    ) -> Self {
        Self {
            id,
            datum_key,
            reviewed_at_unix_seconds,
            review,
        }
    }

    pub fn id(&self) -> i64 {
        self.id
    }

    pub fn datum_key(&self) -> &DatumKey {
        &self.datum_key
    }

    pub fn reviewed_at_unix_seconds(&self) -> i64 {
        self.reviewed_at_unix_seconds
    }

    pub fn review(&self) -> &Review {
        &self.review
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ReviewPass {
    started_at_unix_seconds: i64,
    finished_at_unix_seconds: i64,
    was_scheduled: bool,
    ended_early: bool,
    initial_questions: usize,
    total_questions: usize,
    unique_cards_reviewed: usize,
    items: Vec<ReviewPassItem>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ReviewPassSummary {
    started_at_unix_seconds: i64,
    finished_at_unix_seconds: i64,
    was_scheduled: bool,
    ended_early: bool,
    initial_questions: usize,
    total_questions: usize,
    unique_cards_reviewed: usize,
}

impl ReviewPassSummary {
    pub fn new(
        started_at_unix_seconds: i64,
        finished_at_unix_seconds: i64,
        was_scheduled: bool,
        ended_early: bool,
        initial_questions: usize,
        total_questions: usize,
        unique_cards_reviewed: usize,
    ) -> Self {
        Self {
            started_at_unix_seconds,
            finished_at_unix_seconds,
            was_scheduled,
            ended_early,
            initial_questions,
            total_questions,
            unique_cards_reviewed,
        }
    }

    pub fn started_at_unix_seconds(&self) -> i64 {
        self.started_at_unix_seconds
    }

    pub fn finished_at_unix_seconds(&self) -> i64 {
        self.finished_at_unix_seconds
    }

    pub fn was_scheduled(&self) -> bool {
        self.was_scheduled
    }

    pub fn ended_early(&self) -> bool {
        self.ended_early
    }

    pub fn initial_questions(&self) -> usize {
        self.initial_questions
    }

    pub fn total_questions(&self) -> usize {
        self.total_questions
    }

    pub fn unique_cards_reviewed(&self) -> usize {
        self.unique_cards_reviewed
    }
}

impl ReviewPass {
    pub fn new(summary: ReviewPassSummary, items: Vec<ReviewPassItem>) -> Self {
        Self {
            started_at_unix_seconds: summary.started_at_unix_seconds,
            finished_at_unix_seconds: summary.finished_at_unix_seconds,
            was_scheduled: summary.was_scheduled,
            ended_early: summary.ended_early,
            initial_questions: summary.initial_questions,
            total_questions: summary.total_questions,
            unique_cards_reviewed: summary.unique_cards_reviewed,
            items,
        }
    }

    pub fn started_at_unix_seconds(&self) -> i64 {
        self.started_at_unix_seconds
    }

    pub fn finished_at_unix_seconds(&self) -> i64 {
        self.finished_at_unix_seconds
    }

    pub fn was_scheduled(&self) -> bool {
        self.was_scheduled
    }

    pub fn ended_early(&self) -> bool {
        self.ended_early
    }

    pub fn initial_questions(&self) -> usize {
        self.initial_questions
    }

    pub fn total_questions(&self) -> usize {
        self.total_questions
    }

    pub fn unique_cards_reviewed(&self) -> usize {
        self.unique_cards_reviewed
    }

    pub fn items(&self) -> &[ReviewPassItem] {
        &self.items
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ReviewPassItem {
    review_event_id: i64,
    position: usize,
    was_repeat: bool,
}

impl ReviewPassItem {
    pub fn new(review_event_id: i64, position: usize, was_repeat: bool) -> Self {
        Self {
            review_event_id,
            position,
            was_repeat,
        }
    }

    pub fn review_event_id(self) -> i64 {
        self.review_event_id
    }

    pub fn position(self) -> usize {
        self.position
    }

    pub fn was_repeat(self) -> bool {
        self.was_repeat
    }
}

#[derive(Debug)]
pub enum ReviewStoreError {
    Database(rusqlite::Error),
    IntegrityCheckFailed(String),
    ConsistencyCheckFailed(String),
    UnsupportedSchemaVersion {
        stored_versions: Vec<i64>,
        supported_version: i64,
    },
    InvalidStoredPromptGuess(RatingValueError),
    InvalidStoredRecallFeedback(RatingValueError),
    InvalidStoredMemoryState(String),
    TimeBeforeUnixEpoch(SystemTimeError),
}

impl fmt::Display for ReviewStoreError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Database(source) => write!(f, "review database error: {source}"),
            Self::IntegrityCheckFailed(source) => {
                write!(f, "review database integrity check failed: {source}")
            }
            Self::ConsistencyCheckFailed(source) => {
                write!(f, "review database consistency check failed: {source}")
            }
            Self::UnsupportedSchemaVersion {
                stored_versions,
                supported_version,
            } => write!(
                f,
                "unsupported review database schema version(s): {:?}; this version supports {}",
                stored_versions, supported_version
            ),
            Self::InvalidStoredPromptGuess(source) => {
                write!(f, "invalid stored prompt guess: {source}")
            }
            Self::InvalidStoredRecallFeedback(source) => {
                write!(f, "invalid stored recall feedback: {source}")
            }
            Self::InvalidStoredMemoryState(source) => {
                write!(f, "invalid stored memory state: {source}")
            }
            Self::TimeBeforeUnixEpoch(source) => {
                write!(f, "review time is before the Unix epoch: {source}")
            }
        }
    }
}

impl std::error::Error for ReviewStoreError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::Database(source) => Some(source),
            Self::IntegrityCheckFailed(_) => None,
            Self::ConsistencyCheckFailed(_) => None,
            Self::UnsupportedSchemaVersion { .. } => None,
            Self::InvalidStoredPromptGuess(source) => Some(source),
            Self::InvalidStoredRecallFeedback(source) => Some(source),
            Self::InvalidStoredMemoryState(_) => None,
            Self::TimeBeforeUnixEpoch(source) => Some(source),
        }
    }
}

impl From<rusqlite::Error> for ReviewStoreError {
    fn from(source: rusqlite::Error) -> Self {
        Self::Database(source)
    }
}

impl From<SystemTimeError> for ReviewStoreError {
    fn from(source: SystemTimeError) -> Self {
        Self::TimeBeforeUnixEpoch(source)
    }
}

struct StoredReviewEvent {
    id: i64,
    datum_key: String,
    reviewed_at_unix_seconds: i64,
    prompt_guess: u8,
    recall_feedback: u8,
}

#[derive(Debug, Clone, PartialEq, Eq)]
enum SchemaState {
    Uninitialized,
    Current,
    NeedsMigration(i64),
    Unsupported(Vec<i64>),
}

struct StoredCardSchedule {
    datum_key: String,
    memory_state: String,
    due_at_unix_seconds: i64,
    stability_days: f64,
    difficulty: f64,
    elapsed_days: u32,
    scheduled_days: u32,
    interval_seconds: i64,
    repetitions: u32,
    lapses: u32,
    last_reviewed_at_unix_seconds: Option<i64>,
    last_recall_feedback: Option<u8>,
}

impl TryFrom<StoredCardSchedule> for CardSchedule {
    type Error = ReviewStoreError;

    fn try_from(value: StoredCardSchedule) -> Result<Self, Self::Error> {
        let memory_state = CardMemoryState::from_storage(&value.memory_state)
            .ok_or_else(|| ReviewStoreError::InvalidStoredMemoryState(value.memory_state))?;
        let last_recall_feedback = value
            .last_recall_feedback
            .map(RecallFeedback::try_from)
            .transpose()
            .map_err(ReviewStoreError::InvalidStoredRecallFeedback)?;

        Ok(CardSchedule::from_storage(
            DatumKey::from_storage_key(value.datum_key),
            memory_state,
            value.due_at_unix_seconds,
            value.stability_days,
            value.difficulty,
            value.elapsed_days,
            value.scheduled_days,
            value.interval_seconds,
            value.repetitions,
            value.lapses,
            value.last_reviewed_at_unix_seconds,
            last_recall_feedback,
        ))
    }
}

impl TryFrom<StoredReviewEvent> for ReviewEvent {
    type Error = ReviewStoreError;

    fn try_from(value: StoredReviewEvent) -> Result<Self, Self::Error> {
        let prompt_guess = PromptGuess::try_from(value.prompt_guess)
            .map_err(ReviewStoreError::InvalidStoredPromptGuess)?;
        let recall_feedback = RecallFeedback::try_from(value.recall_feedback)
            .map_err(ReviewStoreError::InvalidStoredRecallFeedback)?;

        Ok(Self::new(
            value.id,
            DatumKey::from_storage_key(value.datum_key),
            value.reviewed_at_unix_seconds,
            Review::new(prompt_guess, recall_feedback),
        ))
    }
}

fn unix_seconds_now() -> Result<i64, ReviewStoreError> {
    Ok(SystemTime::now().duration_since(UNIX_EPOCH)?.as_secs() as i64)
}

fn bool_to_storage(value: bool) -> i64 {
    i64::from(value)
}

fn storage_to_bool(value: i64) -> bool {
    value != 0
}

fn normalize_reverse_prompt(value: &str) -> String {
    value
        .replace("\r\n", "\n")
        .replace('\r', "\n")
        .trim()
        .to_string()
}

fn normalized_response_counts(data: &[Datum]) -> HashMap<String, usize> {
    let mut response_counts = HashMap::new();
    for datum in data {
        *response_counts
            .entry(normalize_reverse_prompt(datum.response()))
            .or_default() += 1;
    }
    response_counts
}

fn verify_datum_key_shape(table_name: &str, datum_key: &DatumKey) -> Result<(), ReviewStoreError> {
    if datum_key
        .parent_key_for_reverse()
        .is_some_and(|parent_key| parent_key.is_reverse())
    {
        return Err(ReviewStoreError::ConsistencyCheckFailed(format!(
            "{table_name} contains nested reverse key {datum_key}"
        )));
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use rusqlite::{OptionalExtension, params};
    use std::fs;
    use std::path::PathBuf;
    use std::process;
    use std::time::{SystemTime, UNIX_EPOCH};

    fn unique_temp_dir(prefix: &str) -> PathBuf {
        let unique = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap()
            .as_nanos();
        let path = std::env::temp_dir().join(format!("cosimo-{prefix}-{}-{unique}", process::id()));
        fs::create_dir_all(&path).unwrap();
        path
    }

    #[test]
    fn records_and_reads_review_events_for_one_datum() {
        let store = SqliteReviewStore::open_in_memory().unwrap();
        let datum = Datum::new("Capital of Spain", "Madrid").unwrap();
        let review = Review::new(PromptGuess::ProbablyKnow, RecallFeedback::Easy);

        let recorded = store
            .record_review_at(&datum, review.clone(), 1_800_000_000)
            .unwrap();

        assert_eq!(recorded.id(), 1);
        assert_eq!(recorded.datum_key(), &DatumKey::for_datum(&datum));
        assert_eq!(recorded.reviewed_at_unix_seconds(), 1_800_000_000);
        assert_eq!(recorded.review(), &review);
        assert_eq!(store.reviews_for_datum(&datum).unwrap(), vec![recorded]);
    }

    #[test]
    fn records_unanswered_prompt_guess() {
        let store = SqliteReviewStore::open_in_memory().unwrap();
        let datum = Datum::new("Capital of Spain", "Madrid").unwrap();
        let review = Review::new(PromptGuess::NotAnswered, RecallFeedback::Good);

        let recorded = store
            .record_review_at(&datum, review.clone(), 1_800_000_000)
            .unwrap();

        assert_eq!(recorded.review(), &review);
        assert_eq!(
            store.reviews_for_datum(&datum).unwrap()[0].review(),
            &review
        );
    }

    #[test]
    fn opening_legacy_database_allows_unanswered_prompt_guess() {
        let temp_dir = unique_temp_dir("legacy-prompt-guess");
        let db_path = temp_dir.join("reviews.sqlite3");
        let connection = Connection::open(&db_path).unwrap();
        connection
            .execute_batch(
                "
                CREATE TABLE cosimo_schema_version (
                    version INTEGER PRIMARY KEY CHECK (version = 1)
                );
                INSERT INTO cosimo_schema_version (version) VALUES (1);
                CREATE TABLE review_events (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    datum_key TEXT NOT NULL,
                    reviewed_at_unix_seconds INTEGER NOT NULL,
                    prompt_guess INTEGER NOT NULL CHECK (prompt_guess BETWEEN 1 AND 4),
                    recall_feedback INTEGER NOT NULL CHECK (recall_feedback BETWEEN 1 AND 4)
                );
                CREATE INDEX review_events_datum_time
                    ON review_events (datum_key, reviewed_at_unix_seconds, id);
                CREATE TABLE card_schedules (
                    datum_key TEXT PRIMARY KEY,
                    memory_state TEXT NOT NULL CHECK (memory_state IN ('new', 'learning', 'review', 'relearning')),
                    due_at_unix_seconds INTEGER NOT NULL,
                    stability_days REAL NOT NULL CHECK (stability_days >= 0),
                    difficulty REAL NOT NULL CHECK (difficulty >= 0),
                    elapsed_days INTEGER NOT NULL CHECK (elapsed_days >= 0),
                    scheduled_days INTEGER NOT NULL CHECK (scheduled_days >= 0),
                    repetitions INTEGER NOT NULL CHECK (repetitions >= 0),
                    lapses INTEGER NOT NULL CHECK (lapses >= 0),
                    last_reviewed_at_unix_seconds INTEGER,
                    last_recall_feedback INTEGER CHECK (
                        last_recall_feedback IS NULL OR last_recall_feedback BETWEEN 1 AND 4
                    )
                );
                ",
            )
            .unwrap();
        drop(connection);

        let store = SqliteReviewStore::open(&db_path).unwrap();
        let datum = Datum::new("Capital of Spain", "Madrid").unwrap();
        let review = Review::new(PromptGuess::NotAnswered, RecallFeedback::Easy);

        store
            .record_review_at(&datum, review.clone(), 1_800_000_000)
            .unwrap();

        assert_eq!(
            store.reviews_for_datum(&datum).unwrap()[0].review(),
            &review
        );
    }

    #[test]
    fn recording_review_updates_card_schedule() {
        let store = SqliteReviewStore::open_in_memory().unwrap();
        let datum = Datum::new("Capital of Spain", "Madrid").unwrap();

        store
            .record_review_at(
                &datum,
                Review::new(PromptGuess::ProbablyKnow, RecallFeedback::Easy),
                1_800_000_000,
            )
            .unwrap();

        let schedule = store.schedule_for_datum(&datum).unwrap().unwrap();
        assert_eq!(schedule.datum_key(), &DatumKey::for_datum(&datum));
        assert_eq!(schedule.memory_state(), CardMemoryState::Review);
        assert_eq!(schedule.repetitions(), 1);
        assert_eq!(schedule.lapses(), 0);
        assert_eq!(schedule.last_recall_feedback(), Some(RecallFeedback::Easy));
        assert_eq!(
            schedule.interval_seconds(),
            i64::from(schedule.scheduled_days()) * DAY_IN_SECONDS
        );
        assert!(schedule.due_at_unix_seconds() > 1_800_000_000);
    }

    #[test]
    fn records_review_passes_and_links_review_events() {
        let store = SqliteReviewStore::open_in_memory().unwrap();
        let datum = Datum::new("Capital of Spain", "Madrid").unwrap();
        let first = store
            .record_review_at(
                &datum,
                Review::new(PromptGuess::ProbablyKnow, RecallFeedback::Again),
                1_800_000_010,
            )
            .unwrap();
        let second = store
            .record_review_at(
                &datum,
                Review::new(PromptGuess::ProbablyKnow, RecallFeedback::Good),
                1_800_000_020,
            )
            .unwrap();

        let pass_id = store
            .record_review_pass(&ReviewPass::new(
                ReviewPassSummary::new(1_800_000_000, 1_800_000_020, true, false, 1, 2, 1),
                vec![
                    ReviewPassItem::new(first.id(), 1, false),
                    ReviewPassItem::new(second.id(), 2, true),
                ],
            ))
            .unwrap();

        let pass: (i64, i64, i64, i64, i64, i64, i64) = store
            .connection
            .query_row(
                "
                SELECT
                    started_at_unix_seconds,
                    finished_at_unix_seconds,
                    was_scheduled,
                    ended_early,
                    initial_questions,
                    total_questions,
                    unique_cards_reviewed
                FROM review_passes
                WHERE id = ?1
                ",
                [pass_id],
                |row| {
                    Ok((
                        row.get(0)?,
                        row.get(1)?,
                        row.get(2)?,
                        row.get(3)?,
                        row.get(4)?,
                        row.get(5)?,
                        row.get(6)?,
                    ))
                },
            )
            .unwrap();
        assert_eq!(pass, (1_800_000_000, 1_800_000_020, 1, 0, 1, 2, 1));

        let items: Vec<(i64, i64, i64)> = store
            .connection
            .prepare(
                "
                SELECT review_event_id, position, was_repeat
                FROM review_pass_items
                WHERE pass_id = ?1
                ORDER BY position
                ",
            )
            .unwrap()
            .query_map([pass_id], |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)))
            .unwrap()
            .collect::<Result<_, _>>()
            .unwrap();
        assert_eq!(items, vec![(first.id(), 1, 0), (second.id(), 2, 1)]);
    }

    #[test]
    fn reads_review_pass_summaries_for_current_datum_keys() {
        let store = SqliteReviewStore::open_in_memory().unwrap();
        let current = Datum::new("Current", "Answer").unwrap();
        let removed = Datum::new("Removed", "Answer").unwrap();
        let current_event = store
            .record_review_at(
                &current,
                Review::new(PromptGuess::ProbablyKnow, RecallFeedback::Good),
                1_800_000_010,
            )
            .unwrap();
        let removed_event = store
            .record_review_at(
                &removed,
                Review::new(PromptGuess::ProbablyKnow, RecallFeedback::Good),
                1_800_000_030,
            )
            .unwrap();
        store
            .record_review_pass(&ReviewPass::new(
                ReviewPassSummary::new(1_800_000_000, 1_800_000_020, true, false, 1, 1, 1),
                vec![ReviewPassItem::new(current_event.id(), 1, false)],
            ))
            .unwrap();
        store
            .record_review_pass(&ReviewPass::new(
                ReviewPassSummary::new(1_800_000_020, 1_800_000_040, true, false, 1, 1, 1),
                vec![ReviewPassItem::new(removed_event.id(), 1, false)],
            ))
            .unwrap();

        let summaries = store
            .review_pass_summaries_for_datum_keys(&HashSet::from([DatumKey::for_datum(&current)]))
            .unwrap();

        assert_eq!(summaries.len(), 1);
        assert_eq!(summaries[0].started_at_unix_seconds(), 1_800_000_000);
        assert_eq!(summaries[0].finished_at_unix_seconds(), 1_800_000_020);
        assert_eq!(summaries[0].total_questions(), 1);
    }

    #[test]
    fn initialize_is_idempotent_and_records_schema_version() {
        let store = SqliteReviewStore::open_in_memory().unwrap();

        store.initialize().unwrap();

        let version: i64 = store
            .connection
            .query_row("SELECT version FROM cosimo_schema_version", [], |row| {
                row.get(0)
            })
            .unwrap();

        assert_eq!(version, CURRENT_SCHEMA_VERSION);
        assert!(
            !store
                .schema_version_table_sql()
                .unwrap()
                .unwrap()
                .contains("CHECK")
        );
    }

    #[test]
    fn stores_and_clears_card_suspensions() {
        let store = SqliteReviewStore::open_in_memory().unwrap();
        let datum_key = DatumKey::from_storage_key("card:1".to_string());

        assert!(!store.table_exists("card_suspensions").unwrap());
        assert!(store.suspended_datum_keys().unwrap().is_empty());

        store.set_datum_suspended(&datum_key, true).unwrap();
        assert!(store.suspended_datum_keys().unwrap().contains(&datum_key));

        store.set_datum_suspended(&datum_key, false).unwrap();
        assert!(!store.suspended_datum_keys().unwrap().contains(&datum_key));
    }

    #[test]
    fn stores_and_clears_card_flags() {
        let store = SqliteReviewStore::open_in_memory().unwrap();
        let datum_key = DatumKey::from_storage_key("card:1".to_string());

        assert!(store.flagged_datum_keys().unwrap().is_empty());

        store.set_datum_flagged(&datum_key, true).unwrap();
        assert!(store.flagged_datum_keys().unwrap().contains(&datum_key));

        store.set_datum_flagged(&datum_key, false).unwrap();
        assert!(!store.flagged_datum_keys().unwrap().contains(&datum_key));

        store.set_datum_flagged(&datum_key, true).unwrap();
        assert!(store.flagged_datum_keys().unwrap().contains(&datum_key));
    }

    #[test]
    fn stores_and_clears_card_deferrals() {
        let store = SqliteReviewStore::open_in_memory().unwrap();
        let datum_key = DatumKey::for_card_id(29);

        assert!(store.deferred_datum_keys(1_000).unwrap().is_empty());

        store.defer_datum_until(&datum_key, 2_000).unwrap();
        assert!(
            store
                .deferred_datum_keys(1_999)
                .unwrap()
                .contains(&datum_key)
        );
        assert!(
            !store
                .deferred_datum_keys(2_000)
                .unwrap()
                .contains(&datum_key)
        );

        store.defer_datum_until(&datum_key, 1_900_000_000).unwrap();
        assert!(
            store
                .deferred_datum_keys(1_899_999_999)
                .unwrap()
                .contains(&datum_key)
        );

        store.clear_datum_deferred(&datum_key).unwrap();
        assert!(
            !store
                .deferred_datum_keys(1_899_999_999)
                .unwrap()
                .contains(&datum_key)
        );
    }

    #[test]
    fn clear_inactive_card_states_removes_expired_and_inactive_rows() {
        let store = SqliteReviewStore::open_in_memory().unwrap();
        let suspended_key = DatumKey::for_card_id(1);
        let flagged_key = DatumKey::for_card_id(2);
        let expired_deferred_key = DatumKey::for_card_id(3);
        let future_deferred_key = DatumKey::for_card_id(4);

        store.set_datum_suspended(&suspended_key, true).unwrap();
        store.set_datum_suspended(&suspended_key, false).unwrap();
        store.set_datum_flagged(&flagged_key, true).unwrap();
        store.set_datum_flagged(&flagged_key, false).unwrap();
        store
            .defer_datum_until(&expired_deferred_key, 2_000)
            .unwrap();
        store
            .defer_datum_until(&future_deferred_key, 4_000)
            .unwrap();

        assert_eq!(store.clear_inactive_card_states(3_000).unwrap(), 3);

        assert!(store.suspended_datum_keys().unwrap().is_empty());
        assert!(store.flagged_datum_keys().unwrap().is_empty());
        assert!(
            !store
                .deferred_datum_keys(3_000)
                .unwrap()
                .contains(&expired_deferred_key)
        );
        assert!(
            store
                .deferred_datum_keys(3_000)
                .unwrap()
                .contains(&future_deferred_key)
        );
        assert_eq!(store.clear_inactive_card_states(3_000).unwrap(), 0);
    }

    #[test]
    fn clear_inactive_card_states_keeps_active_rows() {
        let store = SqliteReviewStore::open_in_memory().unwrap();
        let suspended_key = DatumKey::for_card_id(1);
        let flagged_key = DatumKey::for_card_id(2);
        let future_deferred_key = DatumKey::for_card_id(3);

        store.set_datum_suspended(&suspended_key, true).unwrap();
        store.set_datum_flagged(&flagged_key, true).unwrap();
        store
            .defer_datum_until(&future_deferred_key, 4_000)
            .unwrap();

        assert_eq!(store.clear_inactive_card_states(3_000).unwrap(), 0);

        assert!(
            store
                .suspended_datum_keys()
                .unwrap()
                .contains(&suspended_key)
        );
        assert!(store.flagged_datum_keys().unwrap().contains(&flagged_key));
        assert!(
            store
                .deferred_datum_keys(3_000)
                .unwrap()
                .contains(&future_deferred_key)
        );
    }

    #[test]
    fn stores_and_clears_card_reversals() {
        let store = SqliteReviewStore::open_in_memory().unwrap();
        let datum_key = DatumKey::from_storage_key("card:1".to_string());

        assert!(store.reversible_datum_keys().unwrap().is_empty());

        store.set_datum_reversible(&datum_key, true).unwrap();
        assert!(store.reversible_datum_keys().unwrap().contains(&datum_key));

        store.set_datum_reversible(&datum_key, false).unwrap();
        assert!(!store.reversible_datum_keys().unwrap().contains(&datum_key));
    }

    #[test]
    fn stores_and_clears_card_reversals_in_batches() {
        let store = SqliteReviewStore::open_in_memory().unwrap();
        let datum_keys = vec![
            DatumKey::from_storage_key("card:1".to_string()),
            DatumKey::from_storage_key("card:2".to_string()),
            DatumKey::from_storage_key("card:3".to_string()),
        ];

        store.set_datums_reversible(&datum_keys, true).unwrap();

        let reversible = store.reversible_datum_keys().unwrap();
        for datum_key in &datum_keys {
            assert!(reversible.contains(datum_key));
        }

        store
            .set_datums_reversible(&datum_keys[1..], false)
            .unwrap();

        let reversible = store.reversible_datum_keys().unwrap();
        assert!(reversible.contains(&datum_keys[0]));
        assert!(!reversible.contains(&datum_keys[1]));
        assert!(!reversible.contains(&datum_keys[2]));
    }

    #[test]
    fn bulk_reversal_batches_can_be_undone_independently() {
        let store = SqliteReviewStore::open_in_memory().unwrap();
        let manual_key = DatumKey::from_storage_key("card:1".to_string());
        let first_batch_key = DatumKey::from_storage_key("card:2".to_string());
        let second_batch_key = DatumKey::from_storage_key("card:3".to_string());
        store.set_datum_reversible(&manual_key, true).unwrap();
        store
            .set_datums_reversible_in_bulk(std::slice::from_ref(&first_batch_key))
            .unwrap();
        store
            .set_datums_reversible_in_bulk(std::slice::from_ref(&second_batch_key))
            .unwrap();

        assert_eq!(store.latest_bulk_reversal_count().unwrap(), 1);
        assert_eq!(store.remove_latest_bulk_reversal().unwrap(), 1);

        let reversible = store.reversible_datum_keys().unwrap();
        assert!(reversible.contains(&manual_key));
        assert!(reversible.contains(&first_batch_key));
        assert!(!reversible.contains(&second_batch_key));
        assert_eq!(store.latest_bulk_reversal_count().unwrap(), 1);
    }

    #[test]
    fn undoing_bulk_reversal_reports_when_no_bulk_batch_exists() {
        let store = SqliteReviewStore::open_in_memory().unwrap();
        store
            .set_datum_reversible(&DatumKey::from_storage_key("card:1".to_string()), true)
            .unwrap();

        assert_eq!(store.latest_bulk_reversal_count().unwrap(), 0);
        let error = store.remove_latest_bulk_reversal().unwrap_err();
        assert!(matches!(
            error,
            ReviewStoreError::ConsistencyCheckFailed(message)
                if message.contains("no bulk reversal batch")
        ));
    }

    #[test]
    fn deck_consistency_accepts_dormant_removed_card_history() {
        let store = SqliteReviewStore::open_in_memory().unwrap();
        let removed_parent_key = DatumKey::for_card_id(1);
        let removed_reverse_key = DatumKey::from_storage_key("card:1:reverse");
        store
            .set_datum_reversible(&removed_parent_key, true)
            .unwrap();
        store
            .record_review_for_datum_key_at(
                &removed_reverse_key,
                Review::new(PromptGuess::ProbablyKnow, RecallFeedback::Easy),
                1_800_000_000,
            )
            .unwrap();
        let current_deck = vec![Datum::new("two", "dos").unwrap().with_card_id(2)];

        store.verify_deck_consistency(&current_deck).unwrap();
    }

    #[test]
    fn scrub_inactive_data_removes_dormant_rows_and_keeps_active_rows() {
        let store = SqliteReviewStore::open_in_memory().unwrap();
        let active_parent_key = DatumKey::for_card_id(1);
        let active_reverse_key = DatumKey::for_reverse_of(&active_parent_key);
        let removed_standard_key = DatumKey::for_card_id(2);
        let removed_reversible_parent_key = DatumKey::for_card_id(3);
        let removed_reversible_reverse_key =
            DatumKey::for_reverse_of(&removed_reversible_parent_key);
        let inactive_reverse_key = DatumKey::from_storage_key("card:4:reverse");

        store
            .set_datum_reversible(&active_parent_key, true)
            .unwrap();
        store
            .set_datum_reversible(&removed_reversible_parent_key, true)
            .unwrap();
        let active_event = store
            .record_review_for_datum_key_at(
                &active_parent_key,
                Review::new(PromptGuess::ProbablyKnow, RecallFeedback::Good),
                1_800_000_000,
            )
            .unwrap();
        let active_reverse_event = store
            .record_review_for_datum_key_at(
                &active_reverse_key,
                Review::new(PromptGuess::ProbablyKnow, RecallFeedback::Easy),
                1_800_000_010,
            )
            .unwrap();
        let removed_event = store
            .record_review_for_datum_key_at(
                &removed_standard_key,
                Review::new(PromptGuess::DontKnow, RecallFeedback::Again),
                1_800_000_020,
            )
            .unwrap();
        store
            .record_review_for_datum_key_at(
                &removed_reversible_reverse_key,
                Review::new(PromptGuess::ProbablyDontKnow, RecallFeedback::Hard),
                1_800_000_030,
            )
            .unwrap();
        store
            .record_review_for_datum_key_at(
                &inactive_reverse_key,
                Review::new(PromptGuess::ProbablyDontKnow, RecallFeedback::Hard),
                1_800_000_040,
            )
            .unwrap();
        for offset in 0..500 {
            let key = DatumKey::from_storage_key(format!("card:{}", 10_000 + offset));
            store
                .record_review_for_datum_key_at(
                    &key,
                    Review::new(PromptGuess::ProbablyDontKnow, RecallFeedback::Again),
                    1_800_001_000 + offset,
                )
                .unwrap();
        }
        store
            .set_datum_suspended(&inactive_reverse_key, true)
            .unwrap();
        store.set_datum_flagged(&active_parent_key, true).unwrap();
        store
            .defer_datum_until(&active_parent_key, 4_000_000_000)
            .unwrap();
        store.defer_datum_until(&active_reverse_key, 1).unwrap();
        store
            .defer_datum_until(&removed_standard_key, 4_000_000_000)
            .unwrap();
        store
            .set_datum_flagged(&inactive_reverse_key, true)
            .unwrap();
        store
            .set_datum_flagged(&removed_standard_key, true)
            .unwrap();
        store
            .set_datum_flagged(&removed_standard_key, false)
            .unwrap();
        store
            .set_datums_reversible_in_bulk(std::slice::from_ref(&removed_standard_key))
            .unwrap();

        store
            .record_review_pass(&ReviewPass::new(
                ReviewPassSummary::new(1_800_000_000, 1_800_000_010, true, false, 2, 2, 2),
                vec![
                    ReviewPassItem::new(active_event.id(), 1, false),
                    ReviewPassItem::new(active_reverse_event.id(), 2, false),
                ],
            ))
            .unwrap();
        store
            .record_review_pass(&ReviewPass::new(
                ReviewPassSummary::new(1_800_000_020, 1_800_000_030, true, false, 2, 2, 2),
                vec![
                    ReviewPassItem::new(active_event.id(), 1, false),
                    ReviewPassItem::new(removed_event.id(), 2, false),
                ],
            ))
            .unwrap();

        let summary = store
            .scrub_inactive_data(
                [&active_parent_key, &active_reverse_key],
                [&active_parent_key],
            )
            .unwrap();

        assert!(summary.total_removed_rows() > 0);
        assert_eq!(
            store
                .reviews_for_datum_key(&active_parent_key)
                .unwrap()
                .len(),
            1
        );
        assert_eq!(
            store
                .reviews_for_datum_key(&active_reverse_key)
                .unwrap()
                .len(),
            1
        );
        assert!(
            store
                .reviews_for_datum_key(&removed_standard_key)
                .unwrap()
                .is_empty()
        );
        assert!(
            store
                .reviews_for_datum_key(&removed_reversible_reverse_key)
                .unwrap()
                .is_empty()
        );
        assert!(
            store
                .reviews_for_datum_key(&inactive_reverse_key)
                .unwrap()
                .is_empty()
        );
        assert!(
            store
                .schedule_for_datum_key(&active_parent_key)
                .unwrap()
                .is_some()
        );
        assert!(
            store
                .schedule_for_datum_key(&inactive_reverse_key)
                .unwrap()
                .is_none()
        );
        assert_eq!(
            store.reversible_datum_keys().unwrap(),
            HashSet::from([active_parent_key.clone()])
        );
        assert!(
            store
                .legacy_reversal_excluded_datum_keys()
                .unwrap()
                .is_empty()
        );
        assert!(store.suspended_datum_keys().unwrap().is_empty());
        assert_eq!(
            store.flagged_datum_keys().unwrap(),
            HashSet::from([active_parent_key.clone()])
        );
        assert_eq!(
            store.deferred_datum_keys(3_000_000_000).unwrap(),
            HashSet::from([active_parent_key.clone()])
        );
        assert_eq!(
            store
                .connection
                .query_row(
                    "SELECT COUNT(*) FROM card_states WHERE state_type = 'deferred'",
                    [],
                    |row| row.get::<_, i64>(0)
                )
                .unwrap(),
            1
        );
        assert_eq!(
            store
                .connection
                .query_row("SELECT COUNT(*) FROM review_passes", [], |row| {
                    row.get::<_, i64>(0)
                })
                .unwrap(),
            1
        );
        assert_eq!(
            store
                .connection
                .query_row("SELECT COUNT(*) FROM review_pass_items", [], |row| {
                    row.get::<_, i64>(0)
                })
                .unwrap(),
            2
        );
        assert_eq!(
            store
                .connection
                .query_row("SELECT COUNT(*) FROM card_reversal_batches", [], |row| {
                    row.get::<_, i64>(0)
                })
                .unwrap(),
            0
        );
        assert_eq!(
            store
                .connection
                .query_row("PRAGMA freelist_count", [], |row| row.get::<_, i64>(0))
                .unwrap(),
            0
        );
    }

    #[test]
    fn deck_consistency_rejects_generated_reverse_rows_in_card_reversals() {
        let store = SqliteReviewStore::open_in_memory().unwrap();
        let reverse_key = DatumKey::from_storage_key("card:1:reverse");
        store.set_datum_reversible(&reverse_key, true).unwrap();
        let current_deck = vec![Datum::new("one", "uno").unwrap().with_card_id(1)];

        let error = store.verify_deck_consistency(&current_deck).unwrap_err();

        assert!(matches!(
            error,
            ReviewStoreError::ConsistencyCheckFailed(message)
                if message.contains("generated reverse key card:1:reverse")
        ));
    }

    #[test]
    fn deck_consistency_rejects_active_reversible_duplicate_responses() {
        let store = SqliteReviewStore::open_in_memory().unwrap();
        let reversible_key = DatumKey::for_card_id(1);
        store.set_datum_reversible(&reversible_key, true).unwrap();
        let current_deck = vec![
            Datum::new("one", "uno").unwrap().with_card_id(1),
            Datum::new("another one", "uno").unwrap().with_card_id(2),
        ];

        let error = store.verify_deck_consistency(&current_deck).unwrap_err();

        assert!(matches!(
            error,
            ReviewStoreError::ConsistencyCheckFailed(message)
                if message.contains("reversible card card:1 has a duplicate response")
        ));
    }

    #[test]
    fn deck_consistency_rejects_orphan_review_pass_items() {
        let store = SqliteReviewStore::open_in_memory().unwrap();
        store
            .connection
            .execute_batch(
                "
                PRAGMA foreign_keys = OFF;
                INSERT INTO review_passes (
                    started_at_unix_seconds,
                    finished_at_unix_seconds,
                    was_scheduled,
                    ended_early,
                    initial_questions,
                    total_questions,
                    unique_cards_reviewed
                ) VALUES (1, 2, 1, 0, 1, 1, 1);
                INSERT INTO review_pass_items (
                    pass_id,
                    review_event_id,
                    position,
                    was_repeat
                ) VALUES (1, 999, 1, 0);
                PRAGMA foreign_keys = ON;
                ",
            )
            .unwrap();

        let error = store.verify_deck_consistency(&[]).unwrap_err();

        assert!(matches!(
            error,
            ReviewStoreError::ConsistencyCheckFailed(message)
                if message.contains("review_pass_items row")
                    && message.contains("review_events")
        ));
    }

    #[test]
    fn deck_consistency_rejects_malformed_suspension_keys() {
        let store = SqliteReviewStore::open_in_memory().unwrap();
        let nested_reverse_key = DatumKey::from_storage_key("card:1:reverse:reverse");
        store
            .set_datum_suspended(&nested_reverse_key, true)
            .unwrap();

        let error = store.verify_deck_consistency(&[]).unwrap_err();

        assert!(matches!(
            error,
            ReviewStoreError::ConsistencyCheckFailed(message)
                if message.contains("card_states contains nested reverse key")
        ));
    }

    #[test]
    fn opening_schema_version_one_database_migrates_to_card_states_table() {
        let temp_dir = unique_temp_dir("migrate-to-suspensions");
        let db_path = temp_dir.join("reviews.sqlite3");
        let connection = Connection::open(&db_path).unwrap();
        connection
            .execute_batch(
                "
                CREATE TABLE cosimo_schema_version (
                    version INTEGER PRIMARY KEY
                );
                INSERT INTO cosimo_schema_version (version) VALUES (1);
                CREATE TABLE review_events (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    datum_key TEXT NOT NULL,
                    reviewed_at_unix_seconds INTEGER NOT NULL,
                    prompt_guess INTEGER NOT NULL CHECK (prompt_guess BETWEEN 0 AND 4),
                    recall_feedback INTEGER NOT NULL CHECK (recall_feedback BETWEEN 1 AND 4)
                );
                CREATE TABLE card_schedules (
                    datum_key TEXT PRIMARY KEY,
                    memory_state TEXT NOT NULL CHECK (memory_state IN ('new', 'learning', 'review', 'relearning')),
                    due_at_unix_seconds INTEGER NOT NULL,
                    stability_days REAL NOT NULL CHECK (stability_days >= 0),
                    difficulty REAL NOT NULL CHECK (difficulty >= 0),
                    elapsed_days INTEGER NOT NULL CHECK (elapsed_days >= 0),
                    scheduled_days INTEGER NOT NULL CHECK (scheduled_days >= 0),
                    repetitions INTEGER NOT NULL CHECK (repetitions >= 0),
                    lapses INTEGER NOT NULL CHECK (lapses >= 0),
                    last_reviewed_at_unix_seconds INTEGER,
                    last_recall_feedback INTEGER CHECK (
                        last_recall_feedback IS NULL OR last_recall_feedback BETWEEN 1 AND 4
                    )
                );
                ",
            )
            .unwrap();
        drop(connection);

        let store = SqliteReviewStore::open(&db_path).unwrap();

        assert_eq!(
            store.schema_versions().unwrap(),
            vec![CURRENT_SCHEMA_VERSION]
        );
        assert!(store.suspended_datum_keys().unwrap().is_empty());
        assert!(!store.table_exists("card_suspensions").unwrap());
        assert!(store.table_exists("card_states").unwrap());
        assert!(store
            .connection
            .query_row(
                "SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'review_passes'",
                [],
                |row| row.get::<_, String>(0),
            )
            .optional()
            .unwrap()
            .is_some());
    }

    #[test]
    fn opening_schema_version_two_database_adds_review_pass_tables() {
        let temp_dir = unique_temp_dir("migrate-to-review-passes");
        let db_path = temp_dir.join("reviews.sqlite3");
        let connection = Connection::open(&db_path).unwrap();
        connection
            .execute_batch(
                "
                CREATE TABLE cosimo_schema_version (
                    version INTEGER PRIMARY KEY
                );
                INSERT INTO cosimo_schema_version (version) VALUES (2);
                CREATE TABLE review_events (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    datum_key TEXT NOT NULL,
                    reviewed_at_unix_seconds INTEGER NOT NULL,
                    prompt_guess INTEGER NOT NULL CHECK (prompt_guess BETWEEN 0 AND 4),
                    recall_feedback INTEGER NOT NULL CHECK (recall_feedback BETWEEN 1 AND 4)
                );
                CREATE TABLE card_suspensions (
                    datum_key TEXT PRIMARY KEY
                );
                ",
            )
            .unwrap();
        drop(connection);

        let store = SqliteReviewStore::open(&db_path).unwrap();

        assert_eq!(
            store.schema_versions().unwrap(),
            vec![CURRENT_SCHEMA_VERSION]
        );
        for table_name in ["review_passes", "review_pass_items"] {
            assert!(
                store
                    .connection
                    .query_row(
                        "SELECT name FROM sqlite_master WHERE type = 'table' AND name = ?1",
                        [table_name],
                        |row| row.get::<_, String>(0),
                    )
                    .optional()
                    .unwrap()
                    .is_some()
            );
        }
        for index_name in ["review_passes_started_at", "review_pass_items_event"] {
            assert!(
                store
                    .connection
                    .query_row(
                        "SELECT name FROM sqlite_master WHERE type = 'index' AND name = ?1",
                        [index_name],
                        |row| row.get::<_, String>(0),
                    )
                    .optional()
                    .unwrap()
                    .is_some()
            );
        }
    }

    #[test]
    fn opening_schema_version_three_database_adds_schedule_interval_seconds() {
        let temp_dir = unique_temp_dir("migrate-to-schedule-interval-seconds");
        let db_path = temp_dir.join("reviews.sqlite3");
        let connection = Connection::open(&db_path).unwrap();
        connection
            .execute_batch(
                "
                CREATE TABLE cosimo_schema_version (
                    version INTEGER PRIMARY KEY
                );
                INSERT INTO cosimo_schema_version (version) VALUES (3);
                CREATE TABLE review_events (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    datum_key TEXT NOT NULL,
                    reviewed_at_unix_seconds INTEGER NOT NULL,
                    prompt_guess INTEGER NOT NULL CHECK (prompt_guess BETWEEN 0 AND 4),
                    recall_feedback INTEGER NOT NULL CHECK (recall_feedback BETWEEN 1 AND 4)
                );
                CREATE TABLE card_schedules (
                    datum_key TEXT PRIMARY KEY,
                    memory_state TEXT NOT NULL CHECK (memory_state IN ('new', 'learning', 'review', 'relearning')),
                    due_at_unix_seconds INTEGER NOT NULL,
                    stability_days REAL NOT NULL CHECK (stability_days >= 0),
                    difficulty REAL NOT NULL CHECK (difficulty >= 0),
                    elapsed_days INTEGER NOT NULL CHECK (elapsed_days >= 0),
                    scheduled_days INTEGER NOT NULL CHECK (scheduled_days >= 0),
                    repetitions INTEGER NOT NULL CHECK (repetitions >= 0),
                    lapses INTEGER NOT NULL CHECK (lapses >= 0),
                    last_reviewed_at_unix_seconds INTEGER,
                    last_recall_feedback INTEGER CHECK (
                        last_recall_feedback IS NULL OR last_recall_feedback BETWEEN 1 AND 4
                    )
                );
                INSERT INTO card_schedules (
                    datum_key,
                    memory_state,
                    due_at_unix_seconds,
                    stability_days,
                    difficulty,
                    elapsed_days,
                    scheduled_days,
                    repetitions,
                    lapses,
                    last_reviewed_at_unix_seconds,
                    last_recall_feedback
                ) VALUES (
                    'card:1',
                    'review',
                    1800345600,
                    4.0,
                    5.0,
                    1,
                    4,
                    2,
                    0,
                    1800000000,
                    3
                );
                ",
            )
            .unwrap();
        drop(connection);

        let store = SqliteReviewStore::open(&db_path).unwrap();

        assert_eq!(
            store.schema_versions().unwrap(),
            vec![CURRENT_SCHEMA_VERSION]
        );
        let schedule = store
            .schedule_for_datum_key(&DatumKey::from_storage_key("card:1".to_string()))
            .unwrap()
            .unwrap();
        assert_eq!(schedule.scheduled_days(), 4);
        assert_eq!(schedule.interval_seconds(), 4 * DAY_IN_SECONDS);
    }

    #[test]
    fn opening_schema_version_four_database_adds_card_reversals_table() {
        let temp_dir = unique_temp_dir("migrate-to-card-reversals");
        let db_path = temp_dir.join("reviews.sqlite3");
        let connection = Connection::open(&db_path).unwrap();
        connection
            .execute_batch(
                "
                CREATE TABLE cosimo_schema_version (
                    version INTEGER PRIMARY KEY
                );
                INSERT INTO cosimo_schema_version (version) VALUES (4);
                CREATE TABLE review_events (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    datum_key TEXT NOT NULL,
                    reviewed_at_unix_seconds INTEGER NOT NULL,
                    prompt_guess INTEGER NOT NULL CHECK (prompt_guess BETWEEN 0 AND 4),
                    recall_feedback INTEGER NOT NULL CHECK (recall_feedback BETWEEN 1 AND 4)
                );
                CREATE TABLE card_suspensions (
                    datum_key TEXT PRIMARY KEY
                );
                CREATE TABLE review_passes (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    started_at_unix_seconds INTEGER NOT NULL,
                    finished_at_unix_seconds INTEGER NOT NULL,
                    was_scheduled INTEGER NOT NULL CHECK (was_scheduled IN (0, 1)),
                    ended_early INTEGER NOT NULL CHECK (ended_early IN (0, 1)),
                    initial_questions INTEGER NOT NULL CHECK (initial_questions >= 0),
                    total_questions INTEGER NOT NULL CHECK (total_questions >= 0),
                    unique_cards_reviewed INTEGER NOT NULL CHECK (unique_cards_reviewed >= 0)
                );
                CREATE TABLE review_pass_items (
                    pass_id INTEGER NOT NULL,
                    review_event_id INTEGER NOT NULL,
                    position INTEGER NOT NULL CHECK (position >= 1),
                    was_repeat INTEGER NOT NULL CHECK (was_repeat IN (0, 1)),
                    PRIMARY KEY (pass_id, position),
                    FOREIGN KEY (pass_id) REFERENCES review_passes(id) ON DELETE CASCADE,
                    FOREIGN KEY (review_event_id) REFERENCES review_events(id) ON DELETE CASCADE
                );
                CREATE TABLE card_schedules (
                    datum_key TEXT PRIMARY KEY,
                    memory_state TEXT NOT NULL CHECK (memory_state IN ('new', 'learning', 'review', 'relearning')),
                    due_at_unix_seconds INTEGER NOT NULL,
                    stability_days REAL NOT NULL CHECK (stability_days >= 0),
                    difficulty REAL NOT NULL CHECK (difficulty >= 0),
                    elapsed_days INTEGER NOT NULL CHECK (elapsed_days >= 0),
                    scheduled_days INTEGER NOT NULL CHECK (scheduled_days >= 0),
                    interval_seconds INTEGER NOT NULL CHECK (interval_seconds >= 0),
                    repetitions INTEGER NOT NULL CHECK (repetitions >= 0),
                    lapses INTEGER NOT NULL CHECK (lapses >= 0),
                    last_reviewed_at_unix_seconds INTEGER,
                    last_recall_feedback INTEGER CHECK (
                        last_recall_feedback IS NULL OR last_recall_feedback BETWEEN 1 AND 4
                    )
                );
                ",
            )
            .unwrap();
        drop(connection);

        let store = SqliteReviewStore::open(&db_path).unwrap();

        assert_eq!(
            store.schema_versions().unwrap(),
            vec![CURRENT_SCHEMA_VERSION]
        );
        assert!(store.reversible_datum_keys().unwrap().is_empty());
        assert!(store.column_exists("card_reversals", "created_by").unwrap());
        assert!(store.table_exists("card_reversal_batches").unwrap());
    }

    #[test]
    fn opening_schema_version_five_database_marks_existing_reversals_manual() {
        let temp_dir = unique_temp_dir("migrate-reversal-metadata");
        let db_path = temp_dir.join("reviews.sqlite3");
        let connection = Connection::open(&db_path).unwrap();
        connection
            .execute_batch(
                "
                CREATE TABLE cosimo_schema_version (
                    version INTEGER PRIMARY KEY
                );
                INSERT INTO cosimo_schema_version (version) VALUES (5);
                CREATE TABLE card_reversals (
                    datum_key TEXT PRIMARY KEY
                );
                INSERT INTO card_reversals (datum_key) VALUES ('card:1');
                ",
            )
            .unwrap();
        drop(connection);

        let store = SqliteReviewStore::open(&db_path).unwrap();

        assert_eq!(
            store.schema_versions().unwrap(),
            vec![CURRENT_SCHEMA_VERSION]
        );
        assert!(store.column_exists("card_reversals", "created_by").unwrap());
        assert!(store.table_exists("card_reversal_batches").unwrap());
        let metadata: (String, Option<i64>, Option<i64>) = store
            .connection
            .query_row(
                "
                SELECT created_by, bulk_batch_id, created_at_unix_seconds
                FROM card_reversals
                WHERE datum_key = 'card:1'
                ",
                [],
                |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
            )
            .unwrap();
        assert_eq!(metadata, ("manual".to_string(), None, None));
    }

    #[test]
    fn opening_schema_version_six_database_retires_reversal_exclusions_table() {
        let temp_dir = unique_temp_dir("migrate-reversal-exclusions");
        let db_path = temp_dir.join("reviews.sqlite3");
        let connection = Connection::open(&db_path).unwrap();
        connection
            .execute_batch(
                "
                CREATE TABLE cosimo_schema_version (
                    version INTEGER PRIMARY KEY
                );
                INSERT INTO cosimo_schema_version (version) VALUES (6);
                CREATE TABLE card_reversals (
                    datum_key TEXT PRIMARY KEY,
                    created_by TEXT NOT NULL DEFAULT 'manual' CHECK (created_by IN ('manual', 'bulk')),
                    bulk_batch_id INTEGER,
                    created_at_unix_seconds INTEGER,
                    CHECK ((created_by = 'bulk') = (bulk_batch_id IS NOT NULL))
                );
                CREATE TABLE card_reversal_batches (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    created_at_unix_seconds INTEGER NOT NULL
                );
                ",
            )
            .unwrap();
        drop(connection);

        let store = SqliteReviewStore::open(&db_path).unwrap();

        assert_eq!(
            store.schema_versions().unwrap(),
            vec![CURRENT_SCHEMA_VERSION]
        );
        assert!(!store.table_exists("card_reversal_exclusions").unwrap());
        assert!(
            store
                .table_exists("legacy_card_reversal_exclusions")
                .unwrap()
        );
        assert!(
            store
                .legacy_reversal_excluded_datum_keys()
                .unwrap()
                .is_empty()
        );
    }

    #[test]
    fn opening_schema_version_seven_database_moves_suspensions_to_card_states_table() {
        let temp_dir = unique_temp_dir("migrate-card-states");
        let db_path = temp_dir.join("reviews.sqlite3");
        let connection = Connection::open(&db_path).unwrap();
        connection
            .execute_batch(
                "
                CREATE TABLE cosimo_schema_version (
                    version INTEGER PRIMARY KEY
                );
                INSERT INTO cosimo_schema_version (version) VALUES (7);
                CREATE TABLE card_reversal_exclusions (
                    datum_key TEXT PRIMARY KEY
                );
                CREATE TABLE card_suspensions (
                    datum_key TEXT PRIMARY KEY
                );
                INSERT INTO card_suspensions (datum_key) VALUES ('card:1');
                ",
            )
            .unwrap();
        drop(connection);

        let store = SqliteReviewStore::open(&db_path).unwrap();

        assert_eq!(
            store.schema_versions().unwrap(),
            vec![CURRENT_SCHEMA_VERSION]
        );
        assert!(store.table_exists("card_states").unwrap());
        assert!(!store.table_exists("card_suspensions").unwrap());
        assert_eq!(
            store.suspended_datum_keys().unwrap(),
            HashSet::from([DatumKey::for_card_id(1)])
        );
        assert!(store.flagged_datum_keys().unwrap().is_empty());
    }

    #[test]
    fn opening_schema_version_eight_database_allows_deferred_card_state() {
        let temp_dir = unique_temp_dir("migrate-deferred-card-state");
        let db_path = temp_dir.join("reviews.sqlite3");
        let connection = Connection::open(&db_path).unwrap();
        connection
            .execute_batch(
                "
                CREATE TABLE cosimo_schema_version (
                    version INTEGER PRIMARY KEY
                );
                INSERT INTO cosimo_schema_version (version) VALUES (8);
                CREATE TABLE card_states (
                    datum_key TEXT NOT NULL,
                    state_type TEXT NOT NULL CHECK (state_type IN ('flag', 'suspended')),
                    ended_at_unix_seconds INTEGER CHECK (
                        ended_at_unix_seconds IS NULL OR ended_at_unix_seconds >= 0
                    ),
                    PRIMARY KEY (datum_key, state_type)
                );
                INSERT INTO card_states (
                    datum_key,
                    state_type,
                    ended_at_unix_seconds
                ) VALUES ('card:1', 'flag', NULL);
                ",
            )
            .unwrap();
        drop(connection);

        let store = SqliteReviewStore::open(&db_path).unwrap();

        assert_eq!(
            store.schema_versions().unwrap(),
            vec![CURRENT_SCHEMA_VERSION]
        );
        assert_eq!(
            store.flagged_datum_keys().unwrap(),
            HashSet::from([DatumKey::for_card_id(1)])
        );
        store
            .defer_datum_until(&DatumKey::for_card_id(2), 2_000)
            .unwrap();
        assert_eq!(
            store.deferred_datum_keys(1_999).unwrap(),
            HashSet::from([DatumKey::for_card_id(2)])
        );
    }

    #[test]
    fn opening_schema_version_nine_database_retires_reversal_exclusions_for_deck_migration() {
        let temp_dir = unique_temp_dir("migrate-retire-reversal-exclusions");
        let db_path = temp_dir.join("reviews.sqlite3");
        let connection = Connection::open(&db_path).unwrap();
        connection
            .execute_batch(
                "
                CREATE TABLE cosimo_schema_version (
                    version INTEGER PRIMARY KEY
                );
                INSERT INTO cosimo_schema_version (version) VALUES (9);
                CREATE TABLE card_reversal_exclusions (
                    datum_key TEXT PRIMARY KEY
                );
                INSERT INTO card_reversal_exclusions (datum_key) VALUES ('card:1');
                ",
            )
            .unwrap();
        drop(connection);

        let store = SqliteReviewStore::open(&db_path).unwrap();

        assert_eq!(
            store.schema_versions().unwrap(),
            vec![CURRENT_SCHEMA_VERSION]
        );
        assert!(!store.table_exists("card_reversal_exclusions").unwrap());
        assert_eq!(
            store.legacy_reversal_excluded_datum_keys().unwrap(),
            HashSet::from([DatumKey::for_card_id(1)])
        );
    }

    #[test]
    fn opening_current_database_normalizes_schema_version_table_for_future_migrations() {
        let temp_dir = unique_temp_dir("constrained-version-table");
        let db_path = temp_dir.join("reviews.sqlite3");
        let connection = Connection::open(&db_path).unwrap();
        connection
            .execute_batch(
                "
                CREATE TABLE cosimo_schema_version (
                    version INTEGER PRIMARY KEY CHECK (version = 1)
                );
                INSERT INTO cosimo_schema_version (version) VALUES (1);
                ",
            )
            .unwrap();
        drop(connection);

        let store = SqliteReviewStore::open(&db_path).unwrap();

        let create_sql = store.schema_version_table_sql().unwrap().unwrap();
        assert!(!create_sql.contains("CHECK"));
        assert_eq!(
            store.schema_versions().unwrap(),
            vec![CURRENT_SCHEMA_VERSION]
        );
    }

    #[test]
    fn opening_unknown_old_schema_version_is_rejected() {
        let temp_dir = unique_temp_dir("unknown-old-review-db");
        let db_path = temp_dir.join("reviews.sqlite3");
        let connection = Connection::open(&db_path).unwrap();
        connection
            .execute_batch(
                "
                CREATE TABLE cosimo_schema_version (
                    version INTEGER PRIMARY KEY
                );
                INSERT INTO cosimo_schema_version (version) VALUES (0);
                ",
            )
            .unwrap();
        drop(connection);

        let error = match SqliteReviewStore::open(&db_path) {
            Ok(_) => panic!("unknown old schema version should be rejected"),
            Err(error) => error,
        };

        assert!(matches!(
            error,
            ReviewStoreError::UnsupportedSchemaVersion {
                stored_versions,
                supported_version: CURRENT_SCHEMA_VERSION,
            } if stored_versions == vec![0]
        ));
    }

    #[test]
    fn opening_future_schema_version_is_rejected() {
        let temp_dir = unique_temp_dir("future-review-db");
        let db_path = temp_dir.join("reviews.sqlite3");
        let connection = Connection::open(&db_path).unwrap();
        connection
            .execute_batch(
                "
                CREATE TABLE cosimo_schema_version (
                    version INTEGER PRIMARY KEY
                );
                INSERT INTO cosimo_schema_version (version) VALUES (11);
                ",
            )
            .unwrap();
        drop(connection);

        let error = match SqliteReviewStore::open(&db_path) {
            Ok(_) => panic!("future schema version should be rejected"),
            Err(error) => error,
        };

        assert!(matches!(
            error,
            ReviewStoreError::UnsupportedSchemaVersion {
                stored_versions,
                supported_version: CURRENT_SCHEMA_VERSION,
        } if stored_versions == vec![11]
        ));

        let connection = Connection::open(&db_path).unwrap();
        let review_events_table: Option<String> = connection
            .query_row(
                "SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'review_events'",
                [],
                |row| row.get(0),
            )
            .optional()
            .unwrap();
        assert_eq!(review_events_table, None);
    }

    #[test]
    fn opening_database_with_multiple_schema_versions_is_rejected() {
        let temp_dir = unique_temp_dir("multiple-review-db-versions");
        let db_path = temp_dir.join("reviews.sqlite3");
        let connection = Connection::open(&db_path).unwrap();
        connection
            .execute_batch(
                "
                CREATE TABLE cosimo_schema_version (
                    version INTEGER PRIMARY KEY
                );
                INSERT INTO cosimo_schema_version (version) VALUES (1);
                INSERT INTO cosimo_schema_version (version) VALUES (2);
                ",
            )
            .unwrap();
        drop(connection);

        let error = match SqliteReviewStore::open(&db_path) {
            Ok(_) => panic!("ambiguous schema versions should be rejected"),
            Err(error) => error,
        };

        assert!(matches!(
            error,
            ReviewStoreError::UnsupportedSchemaVersion {
                stored_versions,
                supported_version: CURRENT_SCHEMA_VERSION,
            } if stored_versions == vec![1, 2]
        ));
    }

    #[test]
    fn opening_valid_database_passes_integrity_check() {
        let temp_dir = unique_temp_dir("valid-review-db");
        let db_path = temp_dir.join("reviews.sqlite3");

        let store = SqliteReviewStore::open(&db_path).unwrap();

        store.verify_integrity().unwrap();
    }

    #[test]
    fn opening_corrupt_database_reports_integrity_failure() {
        let temp_dir = unique_temp_dir("corrupt-review-db");
        let db_path = temp_dir.join("reviews.sqlite3");
        fs::write(&db_path, b"not a sqlite database").unwrap();

        let error = match SqliteReviewStore::open(&db_path) {
            Ok(_) => panic!("corrupt database should fail integrity check"),
            Err(error) => error,
        };

        match error {
            ReviewStoreError::IntegrityCheckFailed(message) => {
                assert!(!message.is_empty());
            }
            other => panic!("unexpected error: {other}"),
        }
    }

    #[test]
    fn keeps_review_history_separate_for_different_data() {
        let store = SqliteReviewStore::open_in_memory().unwrap();
        let first = Datum::new("Capital of Spain", "Madrid").unwrap();
        let second = Datum::new("Capital of France", "Paris").unwrap();

        store
            .record_review_at(
                &first,
                Review::new(PromptGuess::DontKnow, RecallFeedback::Again),
                10,
            )
            .unwrap();
        store
            .record_review_at(
                &second,
                Review::new(PromptGuess::CertainlyKnow, RecallFeedback::Easy),
                11,
            )
            .unwrap();

        let first_reviews = store.reviews_for_datum(&first).unwrap();

        assert_eq!(first_reviews.len(), 1);
        assert_eq!(first_reviews[0].datum_key(), &DatumKey::for_datum(&first));
        assert_eq!(
            first_reviews[0].review(),
            &Review::new(PromptGuess::DontKnow, RecallFeedback::Again)
        );
    }

    #[test]
    fn migrating_datum_key_moves_card_schedule() {
        let store = SqliteReviewStore::open_in_memory().unwrap();
        let old_datum = Datum::new("Capital of Spain", "Madrid").unwrap();
        let new_datum = Datum::new("Capital of Spain", "Madrid")
            .unwrap()
            .with_card_id(7);

        store
            .record_review_at(
                &old_datum,
                Review::new(PromptGuess::ProbablyKnow, RecallFeedback::Easy),
                100,
            )
            .unwrap();
        store
            .set_datum_suspended(&DatumKey::for_legacy_content(&old_datum), true)
            .unwrap();
        store
            .migrate_datum_key(
                &DatumKey::for_legacy_content(&old_datum),
                &DatumKey::for_datum(&new_datum),
            )
            .unwrap();

        assert!(store.schedule_for_datum(&old_datum).unwrap().is_none());
        assert!(store.schedule_for_datum(&new_datum).unwrap().is_some());
        assert!(
            store
                .suspended_datum_keys()
                .unwrap()
                .contains(&DatumKey::for_datum(&new_datum))
        );
    }

    #[test]
    fn returns_reviews_in_exposure_order() {
        let store = SqliteReviewStore::open_in_memory().unwrap();
        let datum = Datum::new("Capital of Spain", "Madrid").unwrap();

        store
            .record_review_at(
                &datum,
                Review::new(PromptGuess::ProbablyKnow, RecallFeedback::Easy),
                20,
            )
            .unwrap();
        store
            .record_review_at(
                &datum,
                Review::new(PromptGuess::ProbablyDontKnow, RecallFeedback::Hard),
                10,
            )
            .unwrap();

        let reviews = store.reviews_for_datum(&datum).unwrap();

        assert_eq!(reviews[0].reviewed_at_unix_seconds(), 10);
        assert_eq!(reviews[1].reviewed_at_unix_seconds(), 20);
    }

    #[test]
    fn returns_empty_history_for_unknown_datum() {
        let store = SqliteReviewStore::open_in_memory().unwrap();
        let datum = Datum::new("Capital of Italy", "Rome").unwrap();

        assert!(store.reviews_for_datum(&datum).unwrap().is_empty());
    }

    #[test]
    fn preserves_insertion_order_for_reviews_with_same_timestamp() {
        let store = SqliteReviewStore::open_in_memory().unwrap();
        let datum = Datum::new("Capital of Spain", "Madrid").unwrap();

        let first = store
            .record_review_at(
                &datum,
                Review::new(PromptGuess::DontKnow, RecallFeedback::Again),
                42,
            )
            .unwrap();
        let second = store
            .record_review_at(
                &datum,
                Review::new(PromptGuess::ProbablyKnow, RecallFeedback::Easy),
                42,
            )
            .unwrap();

        let reviews = store.reviews_for_datum(&datum).unwrap();

        assert_eq!(reviews[0].id(), first.id());
        assert_eq!(reviews[1].id(), second.id());
    }

    #[test]
    fn reports_invalid_stored_prompt_guess_values() {
        let store = SqliteReviewStore::open_in_memory().unwrap();
        let datum = Datum::new("Capital of Spain", "Madrid").unwrap();
        let datum_key = DatumKey::for_datum(&datum);

        store
            .connection
            .execute_batch("PRAGMA ignore_check_constraints = ON;")
            .unwrap();
        store
            .connection
            .execute(
                "
                INSERT INTO review_events (
                    datum_key,
                    reviewed_at_unix_seconds,
                    prompt_guess,
                    recall_feedback
                ) VALUES (?1, ?2, ?3, ?4)
                ",
                params![datum_key.as_str(), 1_800_000_000_i64, 9_u8, 4_u8],
            )
            .unwrap();

        let error = store.reviews_for_datum_key(&datum_key).unwrap_err();

        match error {
            ReviewStoreError::InvalidStoredPromptGuess(source) => {
                assert_eq!(source.value(), 9);
            }
            other => panic!("unexpected error: {other}"),
        }
    }
}
