use std::collections::{BTreeMap, BTreeSet};
use std::fmt;
use std::fs;
use std::fs::OpenOptions;
use std::io::{self, Cursor, Read, Write};
use std::path::{Component, Path, PathBuf};
use std::str::FromStr;
use std::time::{SystemTime, UNIX_EPOCH};

use crate::deck_format::parse_deck_file;
use rusqlite::Transaction;
use rusqlite::{Connection, Row, params};
use sha2::{Digest, Sha256};
use zip::write::SimpleFileOptions;
use zip::{CompressionMethod, ZipArchive, ZipWriter};

const SHA256_HASH_PREFIX: &str = "sha256:";
const SYNC_BUNDLE_HASH_PREFIX: &str = SHA256_HASH_PREFIX;
const SYNC_BEARER_TOKEN_VERIFIER_PREFIX: &str = SHA256_HASH_PREFIX;
const HEX: &[u8; 16] = b"0123456789abcdef";
const SYNC_BUNDLE_DEFLATE_LEVEL: i64 = 1;
const DEFAULT_SYNC_MAX_BUNDLE_BYTES: u64 = 512 * 1024 * 1024;
const DEFAULT_SYNC_MAX_BUNDLE_EXPANDED_BYTES: u64 = 1024 * 1024 * 1024;
const DEFAULT_SYNC_MAX_BUNDLE_ENTRIES: usize = 20_000;
const DEFAULT_SYNC_MAX_DECK_FILE_BYTES: u64 = 50 * 1024 * 1024;
const DEFAULT_SYNC_MAX_OAUTH_STARTS_PER_CLIENT_PER_HOUR: usize = 30;
const DEFAULT_SYNC_MAX_OAUTH_STARTS_PER_INSTANCE_PER_HOUR: usize = 200;
const DEFAULT_SYNC_MAX_PENDING_OAUTH_LOGINS_PER_INSTANCE: usize = 100;
const SYNC_OAUTH_RATE_LIMIT_WINDOW_SECONDS: i64 = 60 * 60;
const SYNC_OAUTH_PENDING_LOGIN_TTL_SECONDS: i64 = 15 * 60;
const SYNC_OAUTH_BROWSER_PENDING_INSTANCE: &str = "https://pending.invalid";
const SYNC_OAUTH_BROWSER_PENDING_SECRET: &str = "pending";
const SYNC_OAUTH_BROWSER_PENDING_REDIRECT_URI: &str = "pending";
const SYNC_AUDIO_MANIFEST_FILE_NAME: &str = "audio-manifest.tsv";
const SYNC_REVIEW_DB_SUFFIX: &str = ".reviews.sqlite3";
const SQLITE_SYNC_DELETE_TOMBSTONE_PREFIX: &str = ".delete-";
const SQLITE_SYNC_SCHEMA_V4_SQL: &str = "
CREATE TABLE IF NOT EXISTS decks (
    deck_uuid TEXT PRIMARY KEY NOT NULL,
    owner_identity TEXT NOT NULL,
    visibility TEXT NOT NULL,
    revision INTEGER NOT NULL,
    content_bundle_hash TEXT NOT NULL,
    study_history_bundle_hash TEXT,
    last_upload_device_name TEXT,
    deck_title TEXT,
    deck_name TEXT,
    deck_card_count INTEGER,
    deck_audio_file_count INTEGER,
    deck_managed_audio_file_count INTEGER,
    deck_audio_formats TEXT NOT NULL DEFAULT ''
);
CREATE INDEX IF NOT EXISTS idx_sync_decks_owner ON decks(owner_identity, deck_uuid);
CREATE INDEX IF NOT EXISTS idx_sync_decks_visibility ON decks(visibility, deck_uuid);
CREATE TABLE IF NOT EXISTS deck_identity_owners (
    deck_uuid TEXT PRIMARY KEY NOT NULL,
    owner_identity TEXT
);
CREATE INDEX IF NOT EXISTS idx_deck_identity_owners_owner
    ON deck_identity_owners(owner_identity, deck_uuid);
CREATE TABLE IF NOT EXISTS auth_sessions (
    bearer_token TEXT PRIMARY KEY NOT NULL,
    owner_identity TEXT NOT NULL,
    created_at_unix_seconds INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_auth_sessions_owner ON auth_sessions(owner_identity);
CREATE TABLE IF NOT EXISTS oauth_pending_logins (
    login_token TEXT PRIMARY KEY NOT NULL,
    instance_url TEXT NOT NULL,
    client_id TEXT NOT NULL,
    client_secret TEXT NOT NULL,
    redirect_uri TEXT NOT NULL,
    expected_owner_identity TEXT,
    created_at_unix_seconds INTEGER NOT NULL,
    accepted_notice_version TEXT,
    accepted_notice_hash TEXT,
    accepted_lawful_basis TEXT,
    accepted_at_unix_seconds INTEGER
);
CREATE INDEX IF NOT EXISTS idx_oauth_pending_logins_instance
    ON oauth_pending_logins(instance_url);
CREATE TABLE IF NOT EXISTS account_privacy_acceptances (
    owner_identity TEXT NOT NULL,
    notice_version TEXT NOT NULL,
    notice_hash TEXT NOT NULL,
    lawful_basis TEXT NOT NULL,
    accepted_at_unix_seconds INTEGER NOT NULL,
    login_token TEXT NOT NULL,
    PRIMARY KEY(owner_identity, notice_version, notice_hash)
);
CREATE TABLE IF NOT EXISTS oauth_completed_logins (
    login_token TEXT PRIMARY KEY NOT NULL,
    owner_identity TEXT NOT NULL,
    completed_at_unix_seconds INTEGER NOT NULL
);
CREATE TABLE IF NOT EXISTS oauth_start_attempts (
    instance_url TEXT NOT NULL,
    client_key TEXT NOT NULL,
    attempted_at_unix_seconds INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_oauth_start_attempts_client
    ON oauth_start_attempts(client_key, attempted_at_unix_seconds);
CREATE INDEX IF NOT EXISTS idx_oauth_start_attempts_instance
    ON oauth_start_attempts(instance_url, attempted_at_unix_seconds);
";

#[derive(Debug, Clone, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub struct SyncUuid(String);

impl SyncUuid {
    pub fn new_v4() -> Result<Self, SyncUuidGenerationError> {
        Ok(Self::from_random_bytes(random_uuid_bytes()?))
    }

    pub fn as_str(&self) -> &str {
        &self.0
    }

    fn from_random_bytes(mut bytes: [u8; 16]) -> Self {
        bytes[6] = (bytes[6] & 0x0f) | 0x40;
        bytes[8] = (bytes[8] & 0x3f) | 0x80;

        Self(format!(
            "{:02x}{:02x}{:02x}{:02x}-{:02x}{:02x}-{:02x}{:02x}-{:02x}{:02x}-{:02x}{:02x}{:02x}{:02x}{:02x}{:02x}",
            bytes[0],
            bytes[1],
            bytes[2],
            bytes[3],
            bytes[4],
            bytes[5],
            bytes[6],
            bytes[7],
            bytes[8],
            bytes[9],
            bytes[10],
            bytes[11],
            bytes[12],
            bytes[13],
            bytes[14],
            bytes[15]
        ))
    }
}

impl FromStr for SyncUuid {
    type Err = SyncUuidParseError;

    fn from_str(value: &str) -> Result<Self, Self::Err> {
        let value = value.trim();
        if value.len() != 36 {
            return Err(SyncUuidParseError::WrongLength);
        }

        let mut has_non_zero_digit = false;
        for (index, byte) in value.bytes().enumerate() {
            if matches!(index, 8 | 13 | 18 | 23) {
                if byte != b'-' {
                    return Err(SyncUuidParseError::MissingHyphen);
                }
                continue;
            }

            if !byte.is_ascii_hexdigit() {
                return Err(SyncUuidParseError::InvalidHexDigit);
            }
            has_non_zero_digit |= byte != b'0';
        }

        if !has_non_zero_digit {
            return Err(SyncUuidParseError::Nil);
        }

        Ok(Self(value.to_ascii_lowercase()))
    }
}

impl fmt::Display for SyncUuid {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(&self.0)
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct SyncServerLimits {
    pub max_bundle_bytes: Option<u64>,
    pub max_bundle_expanded_bytes: Option<u64>,
    pub max_bundle_entries: Option<usize>,
    pub max_deck_file_bytes: Option<u64>,
    pub max_decks_per_owner: Option<usize>,
    pub max_storage_bytes_per_owner: Option<u64>,
    pub max_storage_bytes_total: Option<u64>,
    pub max_oauth_starts_per_client_per_hour: Option<usize>,
    pub max_oauth_starts_per_instance_per_hour: Option<usize>,
    pub max_pending_oauth_logins_per_instance: Option<usize>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SyncQuotaStatus {
    max_decks_per_owner: Option<usize>,
    max_storage_bytes_per_owner: Option<u64>,
    max_storage_bytes_total: Option<u64>,
    owner_deck_count: Option<usize>,
    owner_storage_bytes: Option<u64>,
}

impl SyncQuotaStatus {
    pub fn from_parts(
        max_decks_per_owner: Option<usize>,
        max_storage_bytes_per_owner: Option<u64>,
        max_storage_bytes_total: Option<u64>,
        owner_deck_count: Option<usize>,
        owner_storage_bytes: Option<u64>,
    ) -> Self {
        Self {
            max_decks_per_owner,
            max_storage_bytes_per_owner,
            max_storage_bytes_total,
            owner_deck_count,
            owner_storage_bytes,
        }
    }

    pub fn max_decks_per_owner(&self) -> Option<usize> {
        self.max_decks_per_owner
    }

    pub fn max_storage_bytes_per_owner(&self) -> Option<u64> {
        self.max_storage_bytes_per_owner
    }

    pub fn max_storage_bytes_total(&self) -> Option<u64> {
        self.max_storage_bytes_total
    }

    pub fn owner_deck_count(&self) -> Option<usize> {
        self.owner_deck_count
    }

    pub fn owner_storage_bytes(&self) -> Option<u64> {
        self.owner_storage_bytes
    }
}

impl Default for SyncServerLimits {
    fn default() -> Self {
        Self {
            max_bundle_bytes: Some(DEFAULT_SYNC_MAX_BUNDLE_BYTES),
            max_bundle_expanded_bytes: Some(DEFAULT_SYNC_MAX_BUNDLE_EXPANDED_BYTES),
            max_bundle_entries: Some(DEFAULT_SYNC_MAX_BUNDLE_ENTRIES),
            max_deck_file_bytes: Some(DEFAULT_SYNC_MAX_DECK_FILE_BYTES),
            max_decks_per_owner: None,
            max_storage_bytes_per_owner: None,
            max_storage_bytes_total: None,
            max_oauth_starts_per_client_per_hour: Some(
                DEFAULT_SYNC_MAX_OAUTH_STARTS_PER_CLIENT_PER_HOUR,
            ),
            max_oauth_starts_per_instance_per_hour: Some(
                DEFAULT_SYNC_MAX_OAUTH_STARTS_PER_INSTANCE_PER_HOUR,
            ),
            max_pending_oauth_logins_per_instance: Some(
                DEFAULT_SYNC_MAX_PENDING_OAUTH_LOGINS_PER_INSTANCE,
            ),
        }
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SyncUuidParseError {
    WrongLength,
    MissingHyphen,
    InvalidHexDigit,
    Nil,
}

impl fmt::Display for SyncUuidParseError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::WrongLength => f.write_str("UUID must contain 36 characters"),
            Self::MissingHyphen => f.write_str("UUID must use canonical hyphen positions"),
            Self::InvalidHexDigit => f.write_str("UUID contains a non-hexadecimal digit"),
            Self::Nil => f.write_str("nil UUID is not a valid sync identity"),
        }
    }
}

impl std::error::Error for SyncUuidParseError {}

#[derive(Debug)]
pub enum SyncUuidGenerationError {
    Io(std::io::Error),
    UnsupportedPlatform,
    WindowsStatus(i32),
}

impl fmt::Display for SyncUuidGenerationError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Io(error) => write!(f, "could not read operating-system randomness: {error}"),
            Self::UnsupportedPlatform => {
                f.write_str("UUID generation is not implemented on this platform")
            }
            Self::WindowsStatus(status) => write!(
                f,
                "could not read operating-system randomness: Windows status {status:#x}"
            ),
        }
    }
}

impl std::error::Error for SyncUuidGenerationError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::Io(error) => Some(error),
            Self::UnsupportedPlatform | Self::WindowsStatus(_) => None,
        }
    }
}

impl From<std::io::Error> for SyncUuidGenerationError {
    fn from(error: std::io::Error) -> Self {
        Self::Io(error)
    }
}

#[cfg(unix)]
fn random_uuid_bytes() -> Result<[u8; 16], SyncUuidGenerationError> {
    use std::io::Read;

    let mut bytes = [0; 16];
    std::fs::File::open("/dev/urandom")?.read_exact(&mut bytes)?;
    Ok(bytes)
}

#[cfg(windows)]
fn random_uuid_bytes() -> Result<[u8; 16], SyncUuidGenerationError> {
    use windows_sys::Win32::Security::Cryptography::{
        BCRYPT_USE_SYSTEM_PREFERRED_RNG, BCryptGenRandom,
    };

    let mut bytes = [0; 16];
    // SAFETY: bytes is a valid writable buffer and the Windows API is asked to
    // write no more than its exact length. A null algorithm handle is valid
    // when BCRYPT_USE_SYSTEM_PREFERRED_RNG is supplied.
    let status = unsafe {
        BCryptGenRandom(
            std::ptr::null_mut(),
            bytes.as_mut_ptr(),
            bytes.len() as u32,
            BCRYPT_USE_SYSTEM_PREFERRED_RNG,
        )
    };
    if status < 0 {
        return Err(SyncUuidGenerationError::WindowsStatus(status));
    }

    Ok(bytes)
}

#[cfg(not(any(unix, windows)))]
fn random_uuid_bytes() -> Result<[u8; 16], SyncUuidGenerationError> {
    Err(SyncUuidGenerationError::UnsupportedPlatform)
}

#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub enum DeckVisibility {
    Public,
    Unlisted,
    #[default]
    Private,
}

impl DeckVisibility {
    pub const ALL: &'static [Self] = &[Self::Private, Self::Unlisted, Self::Public];

    pub fn from_code(code: &str) -> Option<Self> {
        match code {
            "public" => Some(Self::Public),
            "unlisted" => Some(Self::Unlisted),
            "private" => Some(Self::Private),
            _ => None,
        }
    }

    pub fn code(self) -> &'static str {
        match self {
            Self::Public => "public",
            Self::Unlisted => "unlisted",
            Self::Private => "private",
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct LocalDeviceIdentity {
    device_name: Option<String>,
}

impl LocalDeviceIdentity {
    pub fn new(device_name: Option<String>) -> Self {
        Self {
            device_name: normalize_device_name(device_name.as_deref()),
        }
    }

    pub fn device_name(&self) -> Option<&str> {
        self.device_name.as_deref()
    }
}

pub fn normalize_device_name(value: Option<&str>) -> Option<String> {
    let value = value?.trim();
    if value.is_empty() || value.contains('\n') || value.contains('\r') {
        None
    } else {
        Some(value.to_string())
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct LocalDeckSyncState {
    deck_uuid: SyncUuid,
    sync_server_url: String,
    owner_identity: Option<String>,
    last_seen_remote_revision: Option<String>,
    last_uploaded_local_bundle_hash: Option<String>,
    last_downloaded_remote_bundle_hash: Option<String>,
}

impl LocalDeckSyncState {
    pub fn from_parts(
        deck_uuid: SyncUuid,
        sync_server_url: &str,
        owner_identity: Option<&str>,
        last_seen_remote_revision: Option<&str>,
        last_uploaded_local_bundle_hash: Option<&str>,
        last_downloaded_remote_bundle_hash: Option<&str>,
    ) -> Result<Self, LocalDeckSyncStateError> {
        Ok(Self {
            deck_uuid,
            sync_server_url: required_config_value("sync_server_url", sync_server_url)?,
            owner_identity: optional_config_value("owner_identity", owner_identity)?,
            last_seen_remote_revision: optional_config_value(
                "last_seen_remote_revision",
                last_seen_remote_revision,
            )?,
            last_uploaded_local_bundle_hash: optional_config_value(
                "last_uploaded_local_bundle_hash",
                last_uploaded_local_bundle_hash,
            )?,
            last_downloaded_remote_bundle_hash: optional_config_value(
                "last_downloaded_remote_bundle_hash",
                last_downloaded_remote_bundle_hash,
            )?,
        })
    }

    pub fn deck_uuid(&self) -> &SyncUuid {
        &self.deck_uuid
    }

    pub fn sync_server_url(&self) -> &str {
        &self.sync_server_url
    }

    pub fn owner_identity(&self) -> Option<&str> {
        self.owner_identity.as_deref()
    }

    pub fn last_seen_remote_revision(&self) -> Option<&str> {
        self.last_seen_remote_revision.as_deref()
    }

    pub fn last_uploaded_local_bundle_hash(&self) -> Option<&str> {
        self.last_uploaded_local_bundle_hash.as_deref()
    }

    pub fn last_downloaded_remote_bundle_hash(&self) -> Option<&str> {
        self.last_downloaded_remote_bundle_hash.as_deref()
    }

    pub fn same_remote_slot(&self, other: &Self) -> bool {
        self.deck_uuid == other.deck_uuid
            && self.sync_server_url == other.sync_server_url
            && self.owner_identity == other.owner_identity
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum LocalDeckSyncStateError {
    EmptyValue { field: &'static str },
    MultilineValue { field: &'static str },
}

impl fmt::Display for LocalDeckSyncStateError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::EmptyValue { field } => write!(f, "{field} must not be empty"),
            Self::MultilineValue { field } => write!(f, "{field} must fit on one line"),
        }
    }
}

impl std::error::Error for LocalDeckSyncStateError {}

fn required_config_value(
    field: &'static str,
    value: &str,
) -> Result<String, LocalDeckSyncStateError> {
    optional_config_value(field, Some(value))?.ok_or(LocalDeckSyncStateError::EmptyValue { field })
}

fn optional_config_value(
    field: &'static str,
    value: Option<&str>,
) -> Result<Option<String>, LocalDeckSyncStateError> {
    let Some(value) = value else {
        return Ok(None);
    };
    let value = value.trim();
    if value.is_empty() {
        return Ok(None);
    }
    if value.contains('\n') || value.contains('\r') {
        return Err(LocalDeckSyncStateError::MultilineValue { field });
    }
    Ok(Some(value.to_string()))
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SyncBundlePayload {
    bytes: Vec<u8>,
    bundle_hash: String,
    includes_review_database: bool,
}

impl SyncBundlePayload {
    pub fn from_parts(
        bytes: Vec<u8>,
        bundle_hash: &str,
        includes_review_database: bool,
    ) -> Result<Self, SyncValidationError> {
        if bytes.is_empty() {
            return Err(SyncValidationError::EmptyValue { field: "bytes" });
        }
        let bundle_hash = required_sync_text("bundle_hash", bundle_hash)?;
        let actual_hash = sync_bundle_hash(&bytes);
        if bundle_hash != actual_hash {
            return Err(SyncValidationError::BundleHashMismatch {
                declared: bundle_hash,
                actual: actual_hash,
            });
        }
        Ok(Self {
            bytes,
            bundle_hash,
            includes_review_database,
        })
    }

    pub fn bytes(&self) -> &[u8] {
        &self.bytes
    }

    pub fn bundle_hash(&self) -> &str {
        &self.bundle_hash
    }

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

pub fn sync_bundle_hash(bytes: &[u8]) -> String {
    sha256_hex_hash(SYNC_BUNDLE_HASH_PREFIX, bytes)
}

fn sync_bearer_token_verifier(bearer_token: &str) -> String {
    sha256_hex_hash(SYNC_BEARER_TOKEN_VERIFIER_PREFIX, bearer_token.as_bytes())
}

fn sha256_hex_hash(prefix: &str, bytes: &[u8]) -> String {
    let digest = Sha256::digest(bytes);
    let mut hash = String::with_capacity(prefix.len() + digest.len() * 2);
    hash.push_str(prefix);
    for byte in digest {
        hash.push(HEX[(byte >> 4) as usize] as char);
        hash.push(HEX[(byte & 0x0f) as usize] as char);
    }
    hash
}

fn derive_content_only_bundle_from_study_history(
    bundle: &SyncBundlePayload,
) -> Result<SyncBundlePayload, SyncBundleDerivationError> {
    if !bundle.includes_review_database() {
        return Ok(bundle.clone());
    }

    let derived_bytes = derive_content_only_bundle_bytes(bundle.bytes())?;
    let derived_hash = sync_bundle_hash(&derived_bytes);
    SyncBundlePayload::from_parts(derived_bytes, &derived_hash, false)
        .map_err(|error| SyncBundleDerivationError::InvalidDerivedBundle(error.to_string()))
}

fn derive_content_only_bundle_bytes(bytes: &[u8]) -> Result<Vec<u8>, SyncBundleDerivationError> {
    let cursor = Cursor::new(bytes);
    let mut source = ZipArchive::new(cursor).map_err(SyncBundleDerivationError::Zip)?;
    let mut target = ZipWriter::new(Cursor::new(Vec::new()));

    for index in 0..source.len() {
        let mut entry = source
            .by_index(index)
            .map_err(SyncBundleDerivationError::Zip)?;
        let entry_name = entry.name().to_string();
        if is_root_review_database_zip_name(&entry_name) {
            continue;
        }
        if entry.is_dir() {
            target
                .add_directory(entry_name, sync_bundle_directory_options())
                .map_err(SyncBundleDerivationError::Zip)?;
            continue;
        }

        target
            .start_file(entry_name, sync_bundle_file_options())
            .map_err(SyncBundleDerivationError::Zip)?;
        io::copy(&mut entry, &mut target).map_err(SyncBundleDerivationError::Io)?;
    }

    let cursor = target.finish().map_err(SyncBundleDerivationError::Zip)?;
    Ok(cursor.into_inner())
}

fn is_root_review_database_zip_name(name: &str) -> bool {
    !name.contains('/') && name.ends_with(".reviews.sqlite3")
}

fn sync_bundle_file_options() -> SimpleFileOptions {
    SimpleFileOptions::default()
        .compression_method(CompressionMethod::Deflated)
        .compression_level(Some(SYNC_BUNDLE_DEFLATE_LEVEL))
        .unix_permissions(0o644)
}

fn sync_bundle_directory_options() -> SimpleFileOptions {
    SimpleFileOptions::default()
        .compression_method(CompressionMethod::Stored)
        .unix_permissions(0o755)
}

#[derive(Debug)]
enum SyncBundleDerivationError {
    Io(io::Error),
    Zip(zip::result::ZipError),
    InvalidDerivedBundle(String),
}

impl fmt::Display for SyncBundleDerivationError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Io(error) => write!(f, "could not derive content-only bundle: {error}"),
            Self::Zip(error) => write!(f, "could not derive content-only bundle: {error}"),
            Self::InvalidDerivedBundle(error) => {
                write!(f, "could not derive content-only bundle: {error}")
            }
        }
    }
}

impl std::error::Error for SyncBundleDerivationError {}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SyncUploadRequest {
    deck_uuid: SyncUuid,
    owner_identity: String,
    visibility: DeckVisibility,
    expected_revision: Option<String>,
    device_identity: LocalDeviceIdentity,
    content_bundle: Option<SyncBundlePayload>,
    study_history_bundle: Option<SyncBundlePayload>,
}

impl SyncUploadRequest {
    pub fn from_parts(
        deck_uuid: SyncUuid,
        owner_identity: &str,
        visibility: DeckVisibility,
        expected_revision: Option<&str>,
        device_identity: LocalDeviceIdentity,
        content_bundle: Option<SyncBundlePayload>,
        study_history_bundle: Option<SyncBundlePayload>,
    ) -> Result<Self, SyncValidationError> {
        if content_bundle.is_none() && study_history_bundle.is_none() {
            return Err(SyncValidationError::MissingUploadBundle);
        }
        if content_bundle
            .as_ref()
            .is_some_and(SyncBundlePayload::includes_review_database)
        {
            return Err(SyncValidationError::ContentBundleContainsReviewDatabase);
        }
        if study_history_bundle
            .as_ref()
            .is_some_and(|bundle| !bundle.includes_review_database())
        {
            return Err(SyncValidationError::StudyHistoryBundleMissingReviewDatabase);
        }
        Ok(Self {
            deck_uuid,
            owner_identity: required_sync_text("owner_identity", owner_identity)?,
            visibility,
            expected_revision: optional_sync_text("expected_revision", expected_revision)?,
            device_identity,
            content_bundle,
            study_history_bundle,
        })
    }

    pub fn deck_uuid(&self) -> &SyncUuid {
        &self.deck_uuid
    }

    pub fn owner_identity(&self) -> &str {
        &self.owner_identity
    }

    pub fn visibility(&self) -> DeckVisibility {
        self.visibility
    }

    pub fn expected_revision(&self) -> Option<&str> {
        self.expected_revision.as_deref()
    }

    pub fn device_identity(&self) -> &LocalDeviceIdentity {
        &self.device_identity
    }

    pub fn content_bundle(&self) -> Option<&SyncBundlePayload> {
        self.content_bundle.as_ref()
    }

    pub fn study_history_bundle(&self) -> Option<&SyncBundlePayload> {
        self.study_history_bundle.as_ref()
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SyncUploadOutcome {
    revision: String,
    bundle_hash: String,
}

impl SyncUploadOutcome {
    pub fn from_parts(revision: &str, bundle_hash: &str) -> Result<Self, SyncValidationError> {
        Ok(Self {
            revision: required_sync_text("revision", revision)?,
            bundle_hash: required_sync_text("bundle_hash", bundle_hash)?,
        })
    }

    pub fn revision(&self) -> &str {
        &self.revision
    }

    pub fn bundle_hash(&self) -> &str {
        &self.bundle_hash
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SyncDownloadKind {
    SharedContent,
    OwnerStudyHistory,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SyncDownload {
    revision: String,
    visibility: DeckVisibility,
    bundle: SyncBundlePayload,
    last_upload_device_name: Option<String>,
}

impl SyncDownload {
    pub fn from_parts(
        revision: &str,
        visibility: DeckVisibility,
        bundle: SyncBundlePayload,
        last_upload_device_name: Option<&str>,
    ) -> Result<Self, SyncValidationError> {
        Ok(Self {
            revision: required_sync_text("revision", revision)?,
            visibility,
            bundle,
            last_upload_device_name: optional_sync_text(
                "last_upload_device_name",
                last_upload_device_name,
            )?,
        })
    }

    pub fn revision(&self) -> &str {
        &self.revision
    }

    pub fn visibility(&self) -> DeckVisibility {
        self.visibility
    }

    pub fn bundle(&self) -> &SyncBundlePayload {
        &self.bundle
    }

    pub fn last_upload_device_name(&self) -> Option<&str> {
        self.last_upload_device_name.as_deref()
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SyncDeckSummary {
    title: Option<String>,
    deck_name: Option<String>,
    card_count: Option<usize>,
    audio_file_count: Option<usize>,
    managed_audio_file_count: Option<usize>,
    audio_formats: Vec<String>,
}

impl SyncDeckSummary {
    pub fn from_parts(
        title: Option<&str>,
        deck_name: Option<&str>,
        card_count: Option<usize>,
        audio_file_count: Option<usize>,
        managed_audio_file_count: Option<usize>,
        audio_formats: Vec<String>,
    ) -> Result<Self, SyncValidationError> {
        let mut summary = Self {
            title: optional_sync_text("title", title)?,
            deck_name: optional_sync_text("deck_name", deck_name)?,
            card_count,
            audio_file_count,
            managed_audio_file_count,
            audio_formats: normalise_sync_audio_formats(audio_formats)?,
        };
        if let (Some(managed), Some(total)) =
            (summary.managed_audio_file_count, summary.audio_file_count)
        {
            summary.managed_audio_file_count = Some(managed.min(total));
        }
        Ok(summary)
    }

    pub fn unknown() -> Self {
        Self {
            title: None,
            deck_name: None,
            card_count: None,
            audio_file_count: None,
            managed_audio_file_count: None,
            audio_formats: Vec::new(),
        }
    }

    pub fn title(&self) -> Option<&str> {
        self.title.as_deref()
    }

    pub fn deck_name(&self) -> Option<&str> {
        self.deck_name.as_deref()
    }

    pub fn display_name(&self) -> Option<&str> {
        self.title().or_else(|| self.deck_name())
    }

    pub fn card_count(&self) -> Option<usize> {
        self.card_count
    }

    pub fn audio_file_count(&self) -> Option<usize> {
        self.audio_file_count
    }

    pub fn managed_audio_file_count(&self) -> Option<usize> {
        self.managed_audio_file_count
    }

    pub fn manual_audio_file_count(&self) -> Option<usize> {
        match (self.audio_file_count, self.managed_audio_file_count) {
            (Some(total), Some(managed)) => Some(total.saturating_sub(managed)),
            _ => None,
        }
    }

    pub fn audio_formats(&self) -> &[String] {
        &self.audio_formats
    }
}

impl Default for SyncDeckSummary {
    fn default() -> Self {
        Self::unknown()
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SyncRemoteDeckStatus {
    deck_uuid: SyncUuid,
    revision: String,
    visibility: DeckVisibility,
    owner_identity: String,
    content_bundle_hash: String,
    study_history_bundle_hash: Option<String>,
    last_upload_device_name: Option<String>,
    summary: SyncDeckSummary,
}

impl SyncRemoteDeckStatus {
    pub fn from_shared_status(
        shared: SyncSharedDeckStatus,
        study_history_bundle_hash: Option<&str>,
        last_upload_device_name: Option<&str>,
    ) -> Result<Self, SyncValidationError> {
        Ok(Self {
            deck_uuid: shared.deck_uuid,
            revision: shared.revision,
            visibility: shared.visibility,
            owner_identity: shared.owner_identity,
            content_bundle_hash: shared.content_bundle_hash,
            study_history_bundle_hash: optional_sync_text(
                "study_history_bundle_hash",
                study_history_bundle_hash,
            )?,
            last_upload_device_name: optional_sync_text(
                "last_upload_device_name",
                last_upload_device_name,
            )?,
            summary: shared.summary,
        })
    }

    pub fn deck_uuid(&self) -> &SyncUuid {
        &self.deck_uuid
    }

    pub fn revision(&self) -> &str {
        &self.revision
    }

    pub fn visibility(&self) -> DeckVisibility {
        self.visibility
    }

    pub fn owner_identity(&self) -> &str {
        &self.owner_identity
    }

    pub fn content_bundle_hash(&self) -> &str {
        &self.content_bundle_hash
    }

    pub fn study_history_bundle_hash(&self) -> Option<&str> {
        self.study_history_bundle_hash.as_deref()
    }

    pub fn has_study_history(&self) -> bool {
        self.study_history_bundle_hash.is_some()
    }

    pub fn last_upload_device_name(&self) -> Option<&str> {
        self.last_upload_device_name.as_deref()
    }

    pub fn summary(&self) -> &SyncDeckSummary {
        &self.summary
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SyncSharedDeckStatus {
    deck_uuid: SyncUuid,
    revision: String,
    visibility: DeckVisibility,
    owner_identity: String,
    content_bundle_hash: String,
    summary: SyncDeckSummary,
}

impl SyncSharedDeckStatus {
    pub fn from_parts(
        deck_uuid: SyncUuid,
        revision: &str,
        visibility: DeckVisibility,
        owner_identity: &str,
        content_bundle_hash: &str,
    ) -> Result<Self, SyncValidationError> {
        Self::from_parts_with_summary(
            deck_uuid,
            revision,
            visibility,
            owner_identity,
            content_bundle_hash,
            SyncDeckSummary::unknown(),
        )
    }

    pub fn from_parts_with_summary(
        deck_uuid: SyncUuid,
        revision: &str,
        visibility: DeckVisibility,
        owner_identity: &str,
        content_bundle_hash: &str,
        summary: SyncDeckSummary,
    ) -> Result<Self, SyncValidationError> {
        Ok(Self {
            deck_uuid,
            revision: required_sync_text("revision", revision)?,
            visibility,
            owner_identity: required_sync_text("owner_identity", owner_identity)?,
            content_bundle_hash: required_sync_text("content_bundle_hash", content_bundle_hash)?,
            summary,
        })
    }

    pub fn deck_uuid(&self) -> &SyncUuid {
        &self.deck_uuid
    }

    pub fn revision(&self) -> &str {
        &self.revision
    }

    pub fn visibility(&self) -> DeckVisibility {
        self.visibility
    }

    pub fn owner_identity(&self) -> &str {
        &self.owner_identity
    }

    pub fn content_bundle_hash(&self) -> &str {
        &self.content_bundle_hash
    }

    pub fn summary(&self) -> &SyncDeckSummary {
        &self.summary
    }
}

#[derive(Debug, Default)]
pub struct InMemorySyncServer {
    next_revision: u64,
    decks: BTreeMap<SyncUuid, StoredSyncDeck>,
    deck_identity_owners: BTreeMap<SyncUuid, Option<String>>,
    auth_sessions: BTreeMap<String, String>,
    pending_oauth_logins: BTreeMap<String, SyncOAuthPendingLogin>,
    completed_oauth_logins: BTreeMap<String, String>,
    oauth_start_attempts: Vec<SyncOAuthStartAttempt>,
    account_privacy_acceptances: BTreeSet<SyncAccountPrivacyAcceptanceKey>,
}

impl InMemorySyncServer {
    pub fn new() -> Self {
        Self::default()
    }

    pub fn create_auth_session_for_verified_owner(
        &mut self,
        owner_identity: &str,
    ) -> Result<SyncAuthSession, SyncAuthError> {
        let session = SyncAuthSession::new(owner_identity)?;
        let verifier = sync_bearer_token_verifier(session.bearer_token());
        self.auth_sessions
            .insert(verifier, session.owner_identity().to_string());
        Ok(session)
    }

    pub fn authenticate_bearer_token(&self, bearer_token: &str) -> Option<String> {
        let verifier = sync_bearer_token_verifier(bearer_token);
        self.auth_sessions.get(&verifier).cloned()
    }

    pub fn owner_account_exists(&self, owner_identity: &str) -> Result<bool, SyncAuthError> {
        let owner_identity = required_sync_text("owner_identity", owner_identity)
            .map_err(SyncAuthError::Validation)?;
        Ok(self
            .decks
            .values()
            .any(|deck| deck.owner_identity.as_str() == owner_identity.as_str())
            || self
                .deck_identity_owners
                .values()
                .any(|owner| owner.as_deref() == Some(owner_identity.as_str()))
            || self
                .auth_sessions
                .values()
                .any(|owner| owner.as_str() == owner_identity.as_str())
            || self
                .account_privacy_acceptances
                .iter()
                .any(|acceptance| acceptance.owner_identity.as_str() == owner_identity.as_str()))
    }

    pub fn revoke_bearer_token(&mut self, bearer_token: &str) -> bool {
        let verifier = sync_bearer_token_verifier(bearer_token);
        self.auth_sessions.remove(&verifier).is_some()
    }

    pub fn revoke_owner_bearer_tokens(&mut self, owner_identity: &str) -> usize {
        let before = self.auth_sessions.len();
        self.auth_sessions
            .retain(|_, owner| owner.as_str() != owner_identity);
        before - self.auth_sessions.len()
    }

    pub fn delete_owner_account(
        &mut self,
        owner_identity: &str,
    ) -> Result<SyncAccountDeletionOutcome, SyncAuthError> {
        let owner_identity = required_sync_text("owner_identity", owner_identity)
            .map_err(SyncAuthError::Validation)?;
        let before_decks = self.decks.len();
        self.decks
            .retain(|_, deck| deck.owner_identity != owner_identity);
        let decks_deleted = before_decks - self.decks.len();
        for owner in self.deck_identity_owners.values_mut() {
            if owner.as_deref() == Some(owner_identity.as_str()) {
                *owner = None;
            }
        }
        let sessions_revoked = self.revoke_owner_bearer_tokens(&owner_identity);
        self.pending_oauth_logins.retain(|_, pending| {
            pending.expected_owner_identity() != Some(owner_identity.as_str())
        });
        self.completed_oauth_logins
            .retain(|_, owner| owner.as_str() != owner_identity);
        self.account_privacy_acceptances
            .retain(|acceptance| acceptance.owner_identity.as_str() != owner_identity);
        Ok(SyncAccountDeletionOutcome::from_parts(
            decks_deleted,
            sessions_revoked,
        ))
    }

    pub fn create_oauth_browser_pending_login(
        &mut self,
    ) -> Result<SyncOAuthPendingLogin, SyncAuthError> {
        self.prune_oauth_login_state(current_unix_seconds());
        let pending = SyncOAuthPendingLogin::new_with_redirect_uri(
            SYNC_OAUTH_BROWSER_PENDING_INSTANCE,
            SYNC_OAUTH_BROWSER_PENDING_SECRET,
            SYNC_OAUTH_BROWSER_PENDING_SECRET,
            SYNC_OAUTH_BROWSER_PENDING_REDIRECT_URI,
        )?;
        self.pending_oauth_logins
            .insert(pending.login_token().to_string(), pending.clone());
        Ok(pending)
    }

    pub fn create_oauth_account_action_pending_login_with_limits(
        &mut self,
        instance_url: &str,
        client_id: &str,
        client_secret: &str,
        redirect_uri: &str,
        expected_owner_identity: &str,
        limits: SyncServerLimits,
    ) -> Result<SyncOAuthPendingLogin, SyncAuthError> {
        self.prune_oauth_login_state(current_unix_seconds());
        enforce_pending_oauth_login_limit(
            self.pending_oauth_logins
                .values()
                .filter(|pending| pending.instance_url() == instance_url)
                .count(),
            limits.max_pending_oauth_logins_per_instance,
        )?;
        let pending = SyncOAuthPendingLogin::new_with_redirect_uri_and_expected_owner_identity(
            instance_url,
            client_id,
            client_secret,
            redirect_uri,
            expected_owner_identity,
        )?;
        self.pending_oauth_logins
            .insert(pending.login_token().to_string(), pending.clone());
        Ok(pending)
    }

    #[allow(clippy::too_many_arguments)]
    pub fn configure_oauth_pending_login_for_browser(
        &mut self,
        login_token: &str,
        instance_url: &str,
        client_id: &str,
        client_secret: &str,
        redirect_uri: &str,
        expected_owner_identity: Option<&str>,
        notice_version: &str,
        notice_hash: &str,
        lawful_basis: &str,
    ) -> Result<Option<SyncOAuthPendingLogin>, SyncAuthError> {
        let Some(pending) = self.pending_oauth_logins.get_mut(login_token) else {
            return Ok(None);
        };
        pending.configure_for_browser(
            instance_url,
            client_id,
            client_secret,
            redirect_uri,
            expected_owner_identity,
        )?;
        pending.accept_privacy_notice(notice_version, notice_hash, lawful_basis)?;
        Ok(Some(pending.clone()))
    }

    pub fn configure_oauth_pending_login_for_browser_authorization(
        &mut self,
        login_token: &str,
        instance_url: &str,
        client_id: &str,
        client_secret: &str,
        redirect_uri: &str,
        expected_owner_identity: Option<&str>,
    ) -> Result<Option<SyncOAuthPendingLogin>, SyncAuthError> {
        let Some(pending) = self.pending_oauth_logins.get_mut(login_token) else {
            return Ok(None);
        };
        pending.configure_for_browser(
            instance_url,
            client_id,
            client_secret,
            redirect_uri,
            expected_owner_identity,
        )?;
        Ok(Some(pending.clone()))
    }

    pub fn take_oauth_pending_login(&mut self, login_token: &str) -> Option<SyncOAuthPendingLogin> {
        self.pending_oauth_logins.remove(login_token)
    }

    pub fn record_privacy_acceptance_for_verified_owner(
        &mut self,
        owner_identity: &str,
        pending: &SyncOAuthPendingLogin,
    ) -> Result<(), SyncAuthError> {
        let acceptance = pending
            .privacy_acceptance_for_owner(owner_identity)
            .ok_or(SyncAuthError::MissingPrivacyAcceptance)?;
        self.account_privacy_acceptances.insert(acceptance.key());
        Ok(())
    }

    pub fn store_completed_oauth_login(
        &mut self,
        login_token: &str,
        owner_identity: &str,
    ) -> Result<(), SyncAuthError> {
        let login_token =
            required_sync_text("login_token", login_token).map_err(SyncAuthError::Validation)?;
        let owner_identity = required_sync_text("owner_identity", owner_identity)
            .map_err(SyncAuthError::Validation)?;
        self.completed_oauth_logins
            .insert(login_token, owner_identity);
        Ok(())
    }

    pub fn take_completed_oauth_login(&mut self, login_token: &str) -> Option<String> {
        self.completed_oauth_logins.remove(login_token)
    }

    pub fn record_oauth_start_attempt_with_limits(
        &mut self,
        instance_url: &str,
        client_key: &str,
        limits: SyncServerLimits,
    ) -> Result<(), SyncAuthError> {
        let instance_url =
            required_sync_text("instance_url", instance_url).map_err(SyncAuthError::Validation)?;
        let client_key =
            required_sync_text("client_key", client_key).map_err(SyncAuthError::Validation)?;
        self.prune_oauth_login_state(current_unix_seconds());
        enforce_oauth_start_attempt_limit(
            self.oauth_start_attempts
                .iter()
                .filter(|attempt| attempt.client_key == client_key)
                .count(),
            limits.max_oauth_starts_per_client_per_hour,
            "client",
        )?;
        enforce_oauth_start_attempt_limit(
            self.oauth_start_attempts
                .iter()
                .filter(|attempt| attempt.instance_url == instance_url)
                .count(),
            limits.max_oauth_starts_per_instance_per_hour,
            "instance",
        )?;
        self.oauth_start_attempts.push(SyncOAuthStartAttempt {
            instance_url,
            client_key,
            attempted_at_unix_seconds: current_unix_seconds(),
        });
        Ok(())
    }

    fn prune_oauth_login_state(&mut self, now_unix_seconds: i64) {
        let pending_cutoff = now_unix_seconds - SYNC_OAUTH_PENDING_LOGIN_TTL_SECONDS;
        self.pending_oauth_logins
            .retain(|_, pending| pending.created_at_unix_seconds() >= pending_cutoff);
        let attempt_cutoff = now_unix_seconds - SYNC_OAUTH_RATE_LIMIT_WINDOW_SECONDS;
        self.oauth_start_attempts
            .retain(|attempt| attempt.attempted_at_unix_seconds >= attempt_cutoff);
    }

    pub fn upload(
        &mut self,
        request: SyncUploadRequest,
    ) -> Result<SyncUploadOutcome, SyncUploadError> {
        let current = self.decks.get(request.deck_uuid());
        if let Some(current) = current
            && current.owner_identity != request.owner_identity()
        {
            return Err(SyncUploadError::Forbidden);
        }
        match self.deck_identity_owners.get(request.deck_uuid()) {
            Some(Some(owner)) if owner != request.owner_identity() => {
                return Err(SyncUploadError::ReservedDeckIdentity);
            }
            Some(None) => return Err(SyncUploadError::RetiredDeckIdentity),
            _ => {}
        }

        let actual_revision = current.map(|deck| deck.revision.as_str());
        if request.expected_revision() != actual_revision {
            return Err(SyncUploadError::Conflict {
                expected: request.expected_revision().map(str::to_string),
                actual: actual_revision.map(str::to_string),
            });
        }

        self.next_revision += 1;
        let revision = self.next_revision.to_string();
        let content_bundle = if let Some(content_bundle) = request.content_bundle() {
            content_bundle.clone()
        } else {
            derive_content_only_bundle_from_study_history(
                request
                    .study_history_bundle()
                    .expect("upload request validated at least one bundle"),
            )
            .map_err(|error| SyncUploadError::InvalidBundle(error.to_string()))?
        };
        let content_bundle_hash = content_bundle.bundle_hash().to_string();
        let bundle_hash = request
            .study_history_bundle()
            .or(request.content_bundle())
            .expect("upload request validated at least one bundle")
            .bundle_hash()
            .to_string();
        let summary = sync_deck_summary_from_upload_request(&request).unwrap_or_default();

        self.deck_identity_owners
            .entry(request.deck_uuid.clone())
            .or_insert_with(|| Some(request.owner_identity.clone()));
        self.decks.insert(
            request.deck_uuid.clone(),
            StoredSyncDeck {
                owner_identity: request.owner_identity,
                visibility: request.visibility,
                revision: revision.clone(),
                content_bundle: request.content_bundle,
                content_bundle_hash,
                study_history_bundle: request.study_history_bundle,
                last_upload_device_name: request.device_identity.device_name().map(str::to_string),
                summary,
            },
        );

        Ok(SyncUploadOutcome {
            revision,
            bundle_hash,
        })
    }

    pub fn download(
        &self,
        deck_uuid: &SyncUuid,
        kind: SyncDownloadKind,
        requester_owner_identity: Option<&str>,
    ) -> Result<SyncDownload, SyncDownloadError> {
        let deck = self
            .decks
            .get(deck_uuid)
            .ok_or(SyncDownloadError::NotFound)?;

        let bundle = match kind {
            SyncDownloadKind::SharedContent => {
                if deck.visibility == DeckVisibility::Private {
                    return Err(SyncDownloadError::Forbidden);
                }
                match (&deck.content_bundle, &deck.study_history_bundle) {
                    (Some(content_bundle), _) => content_bundle.clone(),
                    (None, Some(study_history_bundle)) => {
                        derive_content_only_bundle_from_study_history(study_history_bundle)
                            .map_err(|_| SyncDownloadError::NotFound)?
                    }
                    (None, None) => return Err(SyncDownloadError::NotFound),
                }
            }
            SyncDownloadKind::OwnerStudyHistory => {
                if requester_owner_identity != Some(deck.owner_identity.as_str()) {
                    return Err(SyncDownloadError::Forbidden);
                }
                deck.study_history_bundle
                    .clone()
                    .or_else(|| deck.content_bundle.clone())
                    .ok_or(SyncDownloadError::NotFound)?
            }
        };

        Ok(SyncDownload {
            revision: deck.revision.clone(),
            visibility: deck.visibility,
            bundle,
            last_upload_device_name: deck.last_upload_device_name.clone(),
        })
    }

    pub fn inspect_owner_deck(
        &self,
        deck_uuid: &SyncUuid,
        requester_owner_identity: &str,
    ) -> Result<SyncRemoteDeckStatus, SyncDownloadError> {
        let deck = self
            .decks
            .get(deck_uuid)
            .ok_or(SyncDownloadError::NotFound)?;
        if deck.owner_identity != requester_owner_identity {
            return Err(SyncDownloadError::Forbidden);
        }

        Ok(SyncRemoteDeckStatus::from_stored(deck_uuid.clone(), deck))
    }

    pub fn inspect_shared_deck(
        &self,
        deck_uuid: &SyncUuid,
    ) -> Result<SyncSharedDeckStatus, SyncDownloadError> {
        let deck = self
            .decks
            .get(deck_uuid)
            .ok_or(SyncDownloadError::NotFound)?;
        if deck.visibility == DeckVisibility::Private {
            return Err(SyncDownloadError::Forbidden);
        }

        Ok(SyncSharedDeckStatus::from_stored(deck_uuid.clone(), deck))
    }

    pub fn list_owner_decks(&self, requester_owner_identity: &str) -> Vec<SyncRemoteDeckStatus> {
        self.decks
            .iter()
            .filter(|(_, deck)| deck.owner_identity == requester_owner_identity)
            .map(|(deck_uuid, deck)| SyncRemoteDeckStatus::from_stored(deck_uuid.clone(), deck))
            .collect()
    }

    pub fn list_public_decks(&self) -> Vec<SyncSharedDeckStatus> {
        self.decks
            .iter()
            .filter(|(_, deck)| deck.visibility == DeckVisibility::Public)
            .map(|(deck_uuid, deck)| SyncSharedDeckStatus::from_stored(deck_uuid.clone(), deck))
            .collect()
    }

    pub fn quota_status(
        &self,
        limits: SyncServerLimits,
        owner_identity: Option<&str>,
    ) -> SyncQuotaStatus {
        let (owner_deck_count, owner_storage_bytes) = owner_identity
            .map(|owner_identity| {
                let mut deck_count = 0usize;
                let mut storage_bytes = 0u64;
                for deck in self
                    .decks
                    .values()
                    .filter(|deck| deck.owner_identity == owner_identity)
                {
                    deck_count += 1;
                    storage_bytes =
                        storage_bytes.saturating_add(stored_sync_deck_storage_bytes(deck));
                }
                (Some(deck_count), Some(storage_bytes))
            })
            .unwrap_or((None, None));

        SyncQuotaStatus::from_parts(
            limits.max_decks_per_owner,
            limits.max_storage_bytes_per_owner,
            limits.max_storage_bytes_total,
            owner_deck_count,
            owner_storage_bytes,
        )
    }

    pub fn delete(
        &mut self,
        deck_uuid: &SyncUuid,
        requester_owner_identity: &str,
    ) -> Result<(), SyncDeleteError> {
        let deck = self.decks.get(deck_uuid).ok_or(SyncDeleteError::NotFound)?;
        if deck.owner_identity != requester_owner_identity {
            return Err(SyncDeleteError::Forbidden);
        }
        self.decks.remove(deck_uuid);
        Ok(())
    }
}

fn stored_sync_deck_storage_bytes(deck: &StoredSyncDeck) -> u64 {
    deck.study_history_bundle
        .as_ref()
        .or(deck.content_bundle.as_ref())
        .map(|bundle| bundle.bytes().len() as u64)
        .unwrap_or(0)
}

#[derive(Debug, Clone)]
pub struct SqliteSyncServer {
    root: PathBuf,
}

#[derive(Debug, Clone, Copy)]
struct ExpectedCommittedUpload<'a> {
    deck_uuid: &'a SyncUuid,
    owner_identity: &'a str,
    visibility_code: &'a str,
    revision: &'a str,
    content_bundle_hash: &'a str,
    study_history_bundle_hash: Option<&'a str>,
    required_bundle_file_name: &'a str,
}

impl SqliteSyncServer {
    pub fn new(root: impl Into<PathBuf>) -> Self {
        Self { root: root.into() }
    }

    pub fn root(&self) -> &Path {
        &self.root
    }

    pub fn create_auth_session_for_verified_owner(
        &self,
        owner_identity: &str,
    ) -> Result<SyncAuthSession, SqliteSyncError> {
        let session = SyncAuthSession::new(owner_identity)?;
        let connection = self.open_connection()?;
        let verifier = sync_bearer_token_verifier(session.bearer_token());
        connection.execute(
            "
            INSERT INTO auth_sessions (
                bearer_token,
                owner_identity,
                created_at_unix_seconds
            )
            VALUES (?1, ?2, ?3)
            ",
            params![verifier, session.owner_identity(), current_unix_seconds()],
        )?;
        Ok(session)
    }

    pub fn authenticate_bearer_token(
        &self,
        bearer_token: &str,
    ) -> Result<Option<String>, SqliteSyncError> {
        let connection = self.open_connection()?;
        let verifier = sync_bearer_token_verifier(bearer_token);
        let mut statement = connection.prepare(
            "
            SELECT owner_identity
            FROM auth_sessions
            WHERE bearer_token = ?1
            ",
        )?;
        let mut rows = statement.query(params![verifier])?;
        let Some(row) = rows.next()? else {
            return Ok(None);
        };
        let owner_identity: String = row.get(0)?;
        Ok(Some(owner_identity))
    }

    pub fn owner_account_exists(&self, owner_identity: &str) -> Result<bool, SqliteSyncError> {
        let owner_identity = required_sync_text("owner_identity", owner_identity)
            .map_err(SyncAuthError::Validation)?;
        let connection = self.open_connection()?;
        let exists: i64 = connection.query_row(
            "
            SELECT
                EXISTS(
                    SELECT 1
                    FROM account_privacy_acceptances
                    WHERE owner_identity = ?1
                )
                OR EXISTS(
                    SELECT 1
                    FROM decks
                    WHERE owner_identity = ?1
                )
                OR EXISTS(
                    SELECT 1
                    FROM auth_sessions
                    WHERE owner_identity = ?1
                )
                OR EXISTS(
                    SELECT 1
                    FROM deck_identity_owners
                    WHERE owner_identity = ?1
                )
            ",
            params![owner_identity.as_str()],
            |row| row.get(0),
        )?;
        Ok(exists != 0)
    }

    pub fn revoke_bearer_token(&self, bearer_token: &str) -> Result<bool, SqliteSyncError> {
        let connection = self.open_connection()?;
        let verifier = sync_bearer_token_verifier(bearer_token);
        let deleted = connection.execute(
            "DELETE FROM auth_sessions WHERE bearer_token = ?1",
            params![verifier],
        )?;
        Ok(deleted > 0)
    }

    pub fn revoke_owner_bearer_tokens(
        &self,
        owner_identity: &str,
    ) -> Result<usize, SqliteSyncError> {
        let connection = self.open_connection()?;
        let deleted = connection.execute(
            "DELETE FROM auth_sessions WHERE owner_identity = ?1",
            params![owner_identity],
        )?;
        Ok(deleted)
    }

    pub fn delete_owner_account(
        &self,
        owner_identity: &str,
    ) -> Result<SyncAccountDeletionOutcome, SqliteSyncError> {
        let owner_identity = required_sync_text("owner_identity", owner_identity)
            .map_err(SyncAuthError::Validation)?;
        let mut connection = self.open_connection()?;
        let transaction =
            connection.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
        let deck_uuids = {
            let mut statement = transaction.prepare(
                "
                SELECT deck_uuid
                FROM decks
                WHERE owner_identity = ?1
                ORDER BY deck_uuid
                ",
            )?;
            let rows = statement.query_map(params![owner_identity.as_str()], |row| {
                row.get::<_, String>(0)
            })?;
            let mut deck_uuids = Vec::new();
            for row in rows {
                let deck_uuid = row?;
                deck_uuids.push(deck_uuid.parse::<SyncUuid>().map_err(|error| {
                    SqliteSyncError::InvalidMetadata(format!(
                        "invalid stored deck UUID during account deletion: {error}"
                    ))
                })?);
            }
            deck_uuids
        };

        let mut tombstones = Vec::new();
        for deck_uuid in &deck_uuids {
            match rename_sqlite_deck_bundle_dir_for_delete(&self.root, deck_uuid) {
                Ok(tombstone) => tombstones.push((deck_uuid.clone(), tombstone)),
                Err(error) => {
                    for (restored_deck_uuid, restored_tombstone) in tombstones {
                        restore_sqlite_deck_bundle_dir_from_tombstone(
                            &self.root,
                            &restored_deck_uuid,
                            restored_tombstone,
                        )?;
                    }
                    return Err(error);
                }
            }
        }

        let result = (|| -> Result<SyncAccountDeletionOutcome, SqliteSyncError> {
            let decks_deleted = transaction.execute(
                "
                DELETE FROM decks
                WHERE owner_identity = ?1
                ",
                params![owner_identity.as_str()],
            )?;
            let sessions_revoked = transaction.execute(
                "
                DELETE FROM auth_sessions
                WHERE owner_identity = ?1
                ",
                params![owner_identity.as_str()],
            )?;
            transaction.execute(
                "
                UPDATE deck_identity_owners
                SET owner_identity = NULL
                WHERE owner_identity = ?1
                ",
                params![owner_identity.as_str()],
            )?;
            transaction.execute(
                "
                DELETE FROM account_privacy_acceptances
                WHERE owner_identity = ?1
                ",
                params![owner_identity.as_str()],
            )?;
            transaction.execute(
                "
                DELETE FROM oauth_completed_logins
                WHERE owner_identity = ?1
                ",
                params![owner_identity.as_str()],
            )?;
            transaction.execute(
                "
                DELETE FROM oauth_pending_logins
                WHERE expected_owner_identity = ?1
                ",
                params![owner_identity.as_str()],
            )?;
            transaction.commit()?;
            Ok(SyncAccountDeletionOutcome::from_parts(
                decks_deleted,
                sessions_revoked,
            ))
        })();

        match result {
            Ok(outcome) => {
                for (_, tombstone) in tombstones {
                    remove_sqlite_delete_tombstone(tombstone)?;
                }
                Ok(outcome)
            }
            Err(error) => {
                for (deck_uuid, tombstone) in tombstones {
                    restore_sqlite_deck_bundle_dir_from_tombstone(
                        &self.root, &deck_uuid, tombstone,
                    )?;
                }
                Err(error)
            }
        }
    }

    pub fn create_oauth_browser_pending_login(
        &self,
    ) -> Result<SyncOAuthPendingLogin, SqliteSyncError> {
        let pending = SyncOAuthPendingLogin::new_with_redirect_uri(
            SYNC_OAUTH_BROWSER_PENDING_INSTANCE,
            SYNC_OAUTH_BROWSER_PENDING_SECRET,
            SYNC_OAUTH_BROWSER_PENDING_SECRET,
            SYNC_OAUTH_BROWSER_PENDING_REDIRECT_URI,
        )?;
        let mut connection = self.open_connection()?;
        let transaction =
            connection.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
        prune_sqlite_oauth_pending_logins(&transaction, current_unix_seconds())?;
        transaction.execute(
            "
            INSERT INTO oauth_pending_logins (
                login_token,
                instance_url,
                client_id,
                client_secret,
                redirect_uri,
                created_at_unix_seconds
            )
            VALUES (?1, ?2, ?3, ?4, ?5, ?6)
            ",
            params![
                pending.login_token(),
                pending.instance_url(),
                pending.client_id(),
                pending.client_secret(),
                pending.redirect_uri(),
                pending.created_at_unix_seconds()
            ],
        )?;
        transaction.commit()?;
        Ok(pending)
    }

    pub fn create_oauth_account_action_pending_login_with_limits(
        &self,
        instance_url: &str,
        client_id: &str,
        client_secret: &str,
        redirect_uri: &str,
        expected_owner_identity: &str,
        limits: SyncServerLimits,
    ) -> Result<SyncOAuthPendingLogin, SqliteSyncError> {
        let pending = SyncOAuthPendingLogin::new_with_redirect_uri_and_expected_owner_identity(
            instance_url,
            client_id,
            client_secret,
            redirect_uri,
            expected_owner_identity,
        )?;
        let mut connection = self.open_connection()?;
        let transaction =
            connection.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
        prune_sqlite_oauth_pending_logins(&transaction, current_unix_seconds())?;
        let pending_count: i64 = transaction.query_row(
            "
            SELECT COUNT(*)
            FROM oauth_pending_logins
            WHERE instance_url = ?1
            ",
            params![pending.instance_url()],
            |row| row.get(0),
        )?;
        enforce_pending_oauth_login_limit(
            pending_count as usize,
            limits.max_pending_oauth_logins_per_instance,
        )?;
        transaction.execute(
            "
            INSERT INTO oauth_pending_logins (
                login_token,
                instance_url,
                client_id,
                client_secret,
                redirect_uri,
                expected_owner_identity,
                created_at_unix_seconds
            )
            VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)
            ",
            params![
                pending.login_token(),
                pending.instance_url(),
                pending.client_id(),
                pending.client_secret(),
                pending.redirect_uri(),
                pending.expected_owner_identity(),
                pending.created_at_unix_seconds()
            ],
        )?;
        transaction.commit()?;
        Ok(pending)
    }

    pub fn take_oauth_pending_login(
        &self,
        login_token: &str,
    ) -> Result<Option<SyncOAuthPendingLogin>, SqliteSyncError> {
        let mut connection = self.open_connection()?;
        let transaction =
            connection.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
        let pending = {
            let mut statement = transaction.prepare(
                "
                SELECT login_token,
                       instance_url,
                       client_id,
                       client_secret,
                       redirect_uri,
                       expected_owner_identity,
                       created_at_unix_seconds,
                       accepted_notice_version,
                       accepted_notice_hash,
                       accepted_lawful_basis,
                       accepted_at_unix_seconds
                FROM oauth_pending_logins
                WHERE login_token = ?1
                ",
            )?;
            let mut rows = statement.query(params![login_token])?;
            let Some(row) = rows.next()? else {
                return Ok(None);
            };
            Some(SyncOAuthPendingLogin::from_parts_with_redirect_uri(
                SyncOAuthPendingLoginParts {
                    login_token: row.get::<_, String>(0)?.as_str(),
                    instance_url: row.get::<_, String>(1)?.as_str(),
                    client_id: row.get::<_, String>(2)?.as_str(),
                    client_secret: row.get::<_, String>(3)?.as_str(),
                    redirect_uri: row.get::<_, String>(4)?.as_str(),
                    expected_owner_identity: row.get::<_, Option<String>>(5)?.as_deref(),
                    created_at_unix_seconds: row.get::<_, i64>(6)?,
                    accepted_notice_version: row.get::<_, Option<String>>(7)?.as_deref(),
                    accepted_notice_hash: row.get::<_, Option<String>>(8)?.as_deref(),
                    accepted_lawful_basis: row.get::<_, Option<String>>(9)?.as_deref(),
                    accepted_at_unix_seconds: row.get::<_, Option<i64>>(10)?,
                },
            )?)
        };
        transaction.execute(
            "DELETE FROM oauth_pending_logins WHERE login_token = ?1",
            params![login_token],
        )?;
        transaction.commit()?;
        Ok(pending)
    }

    #[allow(clippy::too_many_arguments)]
    pub fn configure_oauth_pending_login_for_browser(
        &self,
        login_token: &str,
        instance_url: &str,
        client_id: &str,
        client_secret: &str,
        redirect_uri: &str,
        expected_owner_identity: Option<&str>,
        notice_version: &str,
        notice_hash: &str,
        lawful_basis: &str,
    ) -> Result<Option<SyncOAuthPendingLogin>, SqliteSyncError> {
        let mut pending = match self.read_oauth_pending_login(login_token)? {
            Some(pending) => pending,
            None => return Ok(None),
        };
        pending.configure_for_browser(
            instance_url,
            client_id,
            client_secret,
            redirect_uri,
            expected_owner_identity,
        )?;
        pending.accept_privacy_notice(notice_version, notice_hash, lawful_basis)?;
        let connection = self.open_connection()?;
        connection.execute(
            "
            UPDATE oauth_pending_logins
            SET instance_url = ?1,
                client_id = ?2,
                client_secret = ?3,
                redirect_uri = ?4,
                expected_owner_identity = ?5,
                accepted_notice_version = ?6,
                accepted_notice_hash = ?7,
                accepted_lawful_basis = ?8,
                accepted_at_unix_seconds = ?9
            WHERE login_token = ?10
            ",
            params![
                pending.instance_url(),
                pending.client_id(),
                pending.client_secret(),
                pending.redirect_uri(),
                pending.expected_owner_identity(),
                pending.accepted_notice_version(),
                pending.accepted_notice_hash(),
                pending.accepted_lawful_basis(),
                pending.accepted_at_unix_seconds(),
                pending.login_token(),
            ],
        )?;
        Ok(Some(pending))
    }

    pub fn record_privacy_acceptance_for_verified_owner(
        &self,
        owner_identity: &str,
        pending: &SyncOAuthPendingLogin,
    ) -> Result<(), SqliteSyncError> {
        let acceptance = pending
            .privacy_acceptance_for_owner(owner_identity)
            .ok_or(SyncAuthError::MissingPrivacyAcceptance)?;
        let connection = self.open_connection()?;
        connection.execute(
            "
            INSERT OR IGNORE INTO account_privacy_acceptances (
                owner_identity,
                notice_version,
                notice_hash,
                lawful_basis,
                accepted_at_unix_seconds,
                login_token
            )
            VALUES (?1, ?2, ?3, ?4, ?5, ?6)
            ",
            params![
                acceptance.owner_identity,
                acceptance.notice_version,
                acceptance.notice_hash,
                acceptance.lawful_basis,
                acceptance.accepted_at_unix_seconds,
                acceptance.login_token,
            ],
        )?;
        Ok(())
    }

    pub fn configure_oauth_pending_login_for_browser_authorization(
        &self,
        login_token: &str,
        instance_url: &str,
        client_id: &str,
        client_secret: &str,
        redirect_uri: &str,
        expected_owner_identity: Option<&str>,
    ) -> Result<Option<SyncOAuthPendingLogin>, SqliteSyncError> {
        let mut pending = match self.read_oauth_pending_login(login_token)? {
            Some(pending) => pending,
            None => return Ok(None),
        };
        pending.configure_for_browser(
            instance_url,
            client_id,
            client_secret,
            redirect_uri,
            expected_owner_identity,
        )?;
        let connection = self.open_connection()?;
        connection.execute(
            "
            UPDATE oauth_pending_logins
            SET
                instance_url = ?1,
                client_id = ?2,
                client_secret = ?3,
                redirect_uri = ?4,
                expected_owner_identity = ?5
            WHERE login_token = ?6
            ",
            params![
                pending.instance_url(),
                pending.client_id(),
                pending.client_secret(),
                pending.redirect_uri(),
                pending.expected_owner_identity(),
                pending.login_token(),
            ],
        )?;
        Ok(Some(pending))
    }

    pub fn store_completed_oauth_login(
        &self,
        login_token: &str,
        owner_identity: &str,
    ) -> Result<(), SqliteSyncError> {
        let login_token =
            required_sync_text("login_token", login_token).map_err(SyncAuthError::Validation)?;
        let owner_identity = required_sync_text("owner_identity", owner_identity)
            .map_err(SyncAuthError::Validation)?;
        let connection = self.open_connection()?;
        connection.execute(
            "
            INSERT OR REPLACE INTO oauth_completed_logins (
                login_token,
                owner_identity,
                completed_at_unix_seconds
            )
            VALUES (?1, ?2, ?3)
            ",
            params![login_token, owner_identity, current_unix_seconds()],
        )?;
        Ok(())
    }

    pub fn take_completed_oauth_login(
        &self,
        login_token: &str,
    ) -> Result<Option<String>, SqliteSyncError> {
        let mut connection = self.open_connection()?;
        let transaction =
            connection.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
        let owner_identity = {
            let mut statement = transaction.prepare(
                "
                SELECT owner_identity
                FROM oauth_completed_logins
                WHERE login_token = ?1
                ",
            )?;
            let mut rows = statement.query(params![login_token])?;
            let Some(row) = rows.next()? else {
                return Ok(None);
            };
            row.get::<_, String>(0)?
        };
        transaction.execute(
            "DELETE FROM oauth_completed_logins WHERE login_token = ?1",
            params![login_token],
        )?;
        transaction.commit()?;
        Ok(Some(owner_identity))
    }

    fn read_oauth_pending_login(
        &self,
        login_token: &str,
    ) -> Result<Option<SyncOAuthPendingLogin>, SqliteSyncError> {
        let connection = self.open_connection()?;
        let mut statement = connection.prepare(
            "
            SELECT login_token,
                   instance_url,
                   client_id,
                   client_secret,
                   redirect_uri,
                   expected_owner_identity,
                   created_at_unix_seconds,
                   accepted_notice_version,
                   accepted_notice_hash,
                   accepted_lawful_basis,
                   accepted_at_unix_seconds
            FROM oauth_pending_logins
            WHERE login_token = ?1
            ",
        )?;
        let mut rows = statement.query(params![login_token])?;
        let Some(row) = rows.next()? else {
            return Ok(None);
        };
        Ok(Some(SyncOAuthPendingLogin::from_parts_with_redirect_uri(
            SyncOAuthPendingLoginParts {
                login_token: row.get::<_, String>(0)?.as_str(),
                instance_url: row.get::<_, String>(1)?.as_str(),
                client_id: row.get::<_, String>(2)?.as_str(),
                client_secret: row.get::<_, String>(3)?.as_str(),
                redirect_uri: row.get::<_, String>(4)?.as_str(),
                expected_owner_identity: row.get::<_, Option<String>>(5)?.as_deref(),
                created_at_unix_seconds: row.get::<_, i64>(6)?,
                accepted_notice_version: row.get::<_, Option<String>>(7)?.as_deref(),
                accepted_notice_hash: row.get::<_, Option<String>>(8)?.as_deref(),
                accepted_lawful_basis: row.get::<_, Option<String>>(9)?.as_deref(),
                accepted_at_unix_seconds: row.get::<_, Option<i64>>(10)?,
            },
        )?))
    }

    pub fn record_oauth_start_attempt_with_limits(
        &self,
        instance_url: &str,
        client_key: &str,
        limits: SyncServerLimits,
    ) -> Result<(), SqliteSyncError> {
        let instance_url =
            required_sync_text("instance_url", instance_url).map_err(SyncAuthError::Validation)?;
        let client_key =
            required_sync_text("client_key", client_key).map_err(SyncAuthError::Validation)?;
        let now = current_unix_seconds();
        let mut connection = self.open_connection()?;
        let transaction =
            connection.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
        prune_sqlite_oauth_start_attempts(&transaction, now)?;
        let client_count: i64 = transaction.query_row(
            "
            SELECT COUNT(*)
            FROM oauth_start_attempts
            WHERE client_key = ?1
            ",
            params![client_key],
            |row| row.get(0),
        )?;
        enforce_oauth_start_attempt_limit(
            client_count as usize,
            limits.max_oauth_starts_per_client_per_hour,
            "client",
        )?;
        let instance_count: i64 = transaction.query_row(
            "
            SELECT COUNT(*)
            FROM oauth_start_attempts
            WHERE instance_url = ?1
            ",
            params![instance_url],
            |row| row.get(0),
        )?;
        enforce_oauth_start_attempt_limit(
            instance_count as usize,
            limits.max_oauth_starts_per_instance_per_hour,
            "instance",
        )?;
        transaction.execute(
            "
            INSERT INTO oauth_start_attempts (
                instance_url,
                client_key,
                attempted_at_unix_seconds
            )
            VALUES (?1, ?2, ?3)
            ",
            params![instance_url, client_key, now],
        )?;
        transaction.commit()?;
        Ok(())
    }

    pub fn upload(&self, request: SyncUploadRequest) -> Result<SyncUploadOutcome, SqliteSyncError> {
        self.upload_with_limits(request, SyncServerLimits::default())
    }

    pub fn upload_with_limits(
        &self,
        request: SyncUploadRequest,
        limits: SyncServerLimits,
    ) -> Result<SyncUploadOutcome, SqliteSyncError> {
        let _ = self.open_connection()?;
        let summary = validate_sync_upload_bundle_payloads(&request, limits)?;
        let derived_content_bundle =
            if let Some(study_history_bundle) = request.study_history_bundle() {
                derive_content_only_bundle_from_study_history(study_history_bundle)
                    .map_err(|error| SyncUploadError::InvalidBundle(error.to_string()))?
            } else {
                request
                    .content_bundle()
                    .expect("upload request validated at least one bundle")
                    .clone()
            };
        let staged_upload = stage_sqlite_sync_upload_bundles(&self.root, &request)?;

        let upload_result = (|| {
            let mut connection = self.open_connection_without_bundle_reconciliation()?;
            // Acquire the write lock before reading the current revision. A CGI-style
            // deployment may run concurrent processes; serialising here keeps the
            // expected-revision check authoritative. Bundle bytes have already been
            // staged outside this transaction, so the lock is held only for the
            // revision/quota checks, metadata update, and atomic file promotion.
            let transaction =
                connection.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
            let current = read_sqlite_metadata_in_transaction(&transaction, request.deck_uuid())?;
            if let Some(current) = current.as_ref()
                && current.owner_identity != request.owner_identity()
            {
                return Err(SyncUploadError::Forbidden.into());
            }
            let reserved_owner =
                read_sqlite_deck_identity_owner_in_transaction(&transaction, request.deck_uuid())?;
            match reserved_owner.as_ref() {
                Some(Some(owner)) if owner != request.owner_identity() => {
                    return Err(SyncUploadError::ReservedDeckIdentity.into());
                }
                Some(None) => return Err(SyncUploadError::RetiredDeckIdentity.into()),
                _ => {}
            }

            let actual_revision = current.as_ref().map(|deck| deck.revision.as_str());
            if request.expected_revision() != actual_revision {
                return Err(SyncUploadError::Conflict {
                    expected: request.expected_revision().map(str::to_string),
                    actual: actual_revision.map(str::to_string),
                }
                .into());
            }
            if let Some(current) = current.as_ref() {
                let _ = prune_old_sqlite_sync_bundle_files(
                    &self.root,
                    request.deck_uuid(),
                    &current.revision,
                );
            }
            enforce_sqlite_sync_upload_limits(
                &transaction,
                &self.root,
                &request,
                current.as_ref(),
                limits,
            )?;

            let revision_number = actual_revision
                .and_then(|revision| revision.parse::<u64>().ok())
                .unwrap_or(0)
                + 1;
            let revision = revision_number.to_string();
            if reserved_owner.is_none() {
                transaction.execute(
                    "
                    INSERT INTO deck_identity_owners (
                        deck_uuid,
                        owner_identity
                    )
                    VALUES (?1, ?2)
                    ",
                    params![request.deck_uuid().as_str(), request.owner_identity()],
                )?;
            }
            staged_upload.promote(&self.root, request.deck_uuid(), &revision)?;

            if let Err(error) = transaction.execute(
                "
            INSERT INTO decks (
                deck_uuid,
                owner_identity,
                visibility,
                revision,
                content_bundle_hash,
                study_history_bundle_hash,
                last_upload_device_name,
                deck_title,
                deck_name,
                deck_card_count,
                deck_audio_file_count,
                deck_managed_audio_file_count,
                deck_audio_formats
            )
            VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)
            ON CONFLICT(deck_uuid) DO UPDATE SET
                owner_identity = excluded.owner_identity,
                visibility = excluded.visibility,
                revision = excluded.revision,
                content_bundle_hash = excluded.content_bundle_hash,
                study_history_bundle_hash = excluded.study_history_bundle_hash,
                last_upload_device_name = excluded.last_upload_device_name,
                deck_title = excluded.deck_title,
                deck_name = excluded.deck_name,
                deck_card_count = excluded.deck_card_count,
                deck_audio_file_count = excluded.deck_audio_file_count,
                deck_managed_audio_file_count = excluded.deck_managed_audio_file_count,
                deck_audio_formats = excluded.deck_audio_formats
            ",
                params![
                    request.deck_uuid().as_str(),
                    request.owner_identity(),
                    request.visibility().code(),
                    revision_number as i64,
                    derived_content_bundle.bundle_hash(),
                    request
                        .study_history_bundle()
                        .map(SyncBundlePayload::bundle_hash),
                    request.device_identity().device_name(),
                    summary.title(),
                    summary.deck_name(),
                    summary.card_count().map(|value| value as i64),
                    summary.audio_file_count().map(|value| value as i64),
                    summary.managed_audio_file_count().map(|value| value as i64),
                    sync_audio_formats_storage(summary.audio_formats()),
                ],
            ) {
                let _ = remove_promoted_sqlite_sync_upload_bundles(
                    &self.root,
                    request.deck_uuid(),
                    &revision,
                );
                return Err(error.into());
            }
            if let Err(error) = transaction.commit() {
                let _ = remove_promoted_sqlite_sync_upload_bundles(
                    &self.root,
                    request.deck_uuid(),
                    &revision,
                );
                return Err(error.into());
            }
            let required_bundle_file_name = if request.study_history_bundle().is_some() {
                sqlite_study_history_bundle_file_name(&revision)
            } else {
                sqlite_content_bundle_file_name(&revision)
            };
            self.verify_committed_upload(ExpectedCommittedUpload {
                deck_uuid: request.deck_uuid(),
                owner_identity: request.owner_identity(),
                visibility_code: request.visibility().code(),
                revision: &revision,
                content_bundle_hash: derived_content_bundle.bundle_hash(),
                study_history_bundle_hash: request
                    .study_history_bundle()
                    .map(SyncBundlePayload::bundle_hash),
                required_bundle_file_name: &required_bundle_file_name,
            })?;
            let _ = prune_old_sqlite_sync_bundle_files(&self.root, request.deck_uuid(), &revision);

            let bundle_hash = request
                .study_history_bundle()
                .or(request.content_bundle())
                .expect("upload request validated at least one bundle")
                .bundle_hash()
                .to_string();
            Ok(SyncUploadOutcome {
                revision,
                bundle_hash,
            })
        })();
        if upload_result.is_err() {
            let _ = staged_upload.remove_temporary_files(&self.root, request.deck_uuid());
        }
        upload_result
    }

    fn verify_committed_upload(
        &self,
        expected: ExpectedCommittedUpload<'_>,
    ) -> Result<(), SqliteSyncError> {
        let connection = self.open_connection_without_bundle_reconciliation()?;
        let metadata = read_sqlite_metadata(&connection, expected.deck_uuid)?.ok_or_else(|| {
            SqliteSyncError::InvalidMetadata(format!(
                "committed sync upload is missing deck metadata for {}",
                expected.deck_uuid.as_str()
            ))
        })?;
        if metadata.owner_identity != expected.owner_identity {
            return Err(SqliteSyncError::InvalidMetadata(format!(
                "committed sync upload owner mismatch for {}",
                expected.deck_uuid.as_str()
            )));
        }
        if metadata.visibility.code() != expected.visibility_code {
            return Err(SqliteSyncError::InvalidMetadata(format!(
                "committed sync upload visibility mismatch for {}",
                expected.deck_uuid.as_str()
            )));
        }
        if metadata.revision != expected.revision {
            return Err(SqliteSyncError::InvalidMetadata(format!(
                "committed sync upload revision mismatch for {}",
                expected.deck_uuid.as_str()
            )));
        }
        if metadata.content_bundle_hash != expected.content_bundle_hash {
            return Err(SqliteSyncError::InvalidMetadata(format!(
                "committed sync upload content hash mismatch for {}",
                expected.deck_uuid.as_str()
            )));
        }
        if metadata.study_history_bundle_hash.as_deref() != expected.study_history_bundle_hash {
            return Err(SqliteSyncError::InvalidMetadata(format!(
                "committed sync upload study-history hash mismatch for {}",
                expected.deck_uuid.as_str()
            )));
        }

        let deck_dir = ensure_sqlite_deck_bundle_dir(&self.root, expected.deck_uuid, false)?;
        let bundle_path = deck_dir.join(expected.required_bundle_file_name);
        if !ensure_plain_regular_file_if_exists(
            &bundle_path,
            "could not inspect committed sync server bundle file",
        )? {
            return Err(SqliteSyncError::InvalidMetadata(format!(
                "committed sync upload is missing bundle file {} for {}",
                expected.required_bundle_file_name,
                expected.deck_uuid.as_str()
            )));
        }
        Ok(())
    }

    pub fn download(
        &self,
        deck_uuid: &SyncUuid,
        kind: SyncDownloadKind,
        requester_owner_identity: Option<&str>,
    ) -> Result<SyncDownload, SqliteSyncError> {
        let deck = self
            .read_stored_deck(deck_uuid)?
            .ok_or(SyncDownloadError::NotFound)?;
        let bundle = match kind {
            SyncDownloadKind::SharedContent => {
                if deck.visibility == DeckVisibility::Private {
                    return Err(SyncDownloadError::Forbidden.into());
                }
                match (&deck.content_bundle, &deck.study_history_bundle) {
                    (Some(content_bundle), _) => content_bundle.clone(),
                    (None, Some(study_history_bundle)) => {
                        derive_content_only_bundle_from_study_history(study_history_bundle)
                            .map_err(|error| SqliteSyncError::InvalidMetadata(error.to_string()))?
                    }
                    (None, None) => return Err(SyncDownloadError::NotFound.into()),
                }
            }
            SyncDownloadKind::OwnerStudyHistory => {
                if requester_owner_identity != Some(deck.owner_identity.as_str()) {
                    return Err(SyncDownloadError::Forbidden.into());
                }
                deck.study_history_bundle
                    .clone()
                    .or_else(|| deck.content_bundle.clone())
                    .ok_or(SyncDownloadError::NotFound)?
            }
        };

        Ok(SyncDownload {
            revision: deck.revision,
            visibility: deck.visibility,
            bundle,
            last_upload_device_name: deck.last_upload_device_name,
        })
    }

    pub fn inspect_owner_deck(
        &self,
        deck_uuid: &SyncUuid,
        requester_owner_identity: &str,
    ) -> Result<SyncRemoteDeckStatus, SqliteSyncError> {
        let metadata = self
            .read_stored_deck_metadata(deck_uuid)?
            .ok_or(SyncDownloadError::NotFound)?;
        if metadata.owner_identity != requester_owner_identity {
            return Err(SyncDownloadError::Forbidden.into());
        }

        Ok(metadata.into())
    }

    pub fn inspect_shared_deck(
        &self,
        deck_uuid: &SyncUuid,
    ) -> Result<SyncSharedDeckStatus, SqliteSyncError> {
        let metadata = self
            .read_stored_deck_metadata(deck_uuid)?
            .ok_or(SyncDownloadError::NotFound)?;
        if metadata.visibility == DeckVisibility::Private {
            return Err(SyncDownloadError::Forbidden.into());
        }

        Ok(SyncSharedDeckStatus::from_metadata(metadata))
    }

    pub fn list_owner_decks(
        &self,
        requester_owner_identity: &str,
    ) -> Result<Vec<SyncRemoteDeckStatus>, SqliteSyncError> {
        let connection = self.open_connection()?;
        let mut statement = connection.prepare(
            "
            SELECT deck_uuid,
                   owner_identity,
                   visibility,
                   revision,
                   content_bundle_hash,
                   study_history_bundle_hash,
                   last_upload_device_name,
                   deck_title,
                   deck_name,
                   deck_card_count,
                   deck_audio_file_count,
                   deck_managed_audio_file_count,
                   deck_audio_formats
            FROM decks
            WHERE owner_identity = ?1
            ORDER BY deck_uuid
            ",
        )?;
        let mut rows = statement.query(params![requester_owner_identity])?;
        let mut decks = Vec::new();
        while let Some(row) = rows.next()? {
            decks.push(SyncRemoteDeckStatus::from(sqlite_metadata_from_row(row)?));
        }
        Ok(decks)
    }

    pub fn list_public_decks(&self) -> Result<Vec<SyncSharedDeckStatus>, SqliteSyncError> {
        let connection = self.open_connection()?;
        let mut statement = connection.prepare(
            "
            SELECT deck_uuid,
                   owner_identity,
                   visibility,
                   revision,
                   content_bundle_hash,
                   study_history_bundle_hash,
                   last_upload_device_name,
                   deck_title,
                   deck_name,
                   deck_card_count,
                   deck_audio_file_count,
                   deck_managed_audio_file_count,
                   deck_audio_formats
            FROM decks
            WHERE visibility = 'public'
            ORDER BY deck_uuid
            ",
        )?;
        let mut rows = statement.query([])?;
        let mut decks = Vec::new();
        while let Some(row) = rows.next()? {
            decks.push(SyncSharedDeckStatus::from_metadata(
                sqlite_metadata_from_row(row)?,
            ));
        }
        Ok(decks)
    }

    pub fn quota_status(
        &self,
        limits: SyncServerLimits,
        owner_identity: Option<&str>,
    ) -> Result<SyncQuotaStatus, SqliteSyncError> {
        let mut connection = self.open_connection()?;
        let transaction = connection.transaction()?;
        let (owner_deck_count, owner_storage_bytes) = match owner_identity {
            Some(owner_identity) => (
                Some(sqlite_owner_deck_count_in_transaction(
                    &transaction,
                    owner_identity,
                )?),
                Some(sqlite_owner_storage_bytes_in_transaction(
                    &transaction,
                    &self.root,
                    owner_identity,
                )?),
            ),
            None => (None, None),
        };

        Ok(SyncQuotaStatus::from_parts(
            limits.max_decks_per_owner,
            limits.max_storage_bytes_per_owner,
            limits.max_storage_bytes_total,
            owner_deck_count,
            owner_storage_bytes,
        ))
    }

    pub fn delete(
        &self,
        deck_uuid: &SyncUuid,
        requester_owner_identity: &str,
    ) -> Result<(), SqliteSyncError> {
        let mut connection = self.open_connection()?;
        let transaction =
            connection.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
        let metadata = read_sqlite_metadata_in_transaction(&transaction, deck_uuid)?
            .ok_or(SyncDeleteError::NotFound)?;
        if metadata.owner_identity != requester_owner_identity {
            return Err(SyncDeleteError::Forbidden.into());
        }

        let tombstone = rename_sqlite_deck_bundle_dir_for_delete(&self.root, deck_uuid)?;
        if let Err(error) = transaction.execute(
            "
            DELETE FROM decks
            WHERE deck_uuid = ?1
            ",
            params![deck_uuid.as_str()],
        ) {
            restore_sqlite_deck_bundle_dir_from_tombstone(&self.root, deck_uuid, tombstone)?;
            return Err(error.into());
        }
        if let Err(error) = transaction.commit() {
            restore_sqlite_deck_bundle_dir_from_tombstone(&self.root, deck_uuid, tombstone)?;
            return Err(error.into());
        }
        remove_sqlite_delete_tombstone(tombstone)?;
        Ok(())
    }

    fn open_connection(&self) -> Result<Connection, SqliteSyncError> {
        let connection = self.open_connection_without_bundle_reconciliation()?;
        reconcile_sqlite_delete_tombstones(&self.root, &connection)?;
        reconcile_sqlite_bundle_files(&self.root, &connection)?;
        Ok(connection)
    }

    fn open_connection_without_bundle_reconciliation(&self) -> Result<Connection, SqliteSyncError> {
        let db_path = sqlite_sync_database_path(&self.root)?;
        let mut connection = Connection::open(&db_path)?;
        connection.execute_batch(
            "
            PRAGMA busy_timeout = 15000;
            PRAGMA secure_delete = ON;
            CREATE TABLE IF NOT EXISTS sync_schema_version (
                version INTEGER PRIMARY KEY
            );
            ",
        )?;
        let versions = sqlite_sync_schema_versions(&connection)?;
        match versions.as_slice() {
            [] => {
                create_sqlite_sync_schema_v4(&connection)?;
                set_sqlite_sync_schema_version_v4(&connection)?;
            }
            [1] => {
                migrate_sqlite_sync_schema_v1_to_v4(&mut connection)?;
            }
            [2] => {
                migrate_sqlite_sync_schema_v2_to_v4(&mut connection)?;
            }
            [3] => {
                migrate_sqlite_sync_schema_v3_to_v4(&mut connection)?;
            }
            [4] => {
                create_sqlite_sync_schema_v4(&connection)?;
            }
            _ => {
                return Err(SqliteSyncError::InvalidMetadata(
                    "unsupported sync server schema version".to_string(),
                ));
            }
        }
        ensure_sqlite_deck_identity_owners(&connection)?;
        upgrade_sqlite_auth_token_verifiers(&connection)?;
        Ok(connection)
    }

    fn read_stored_deck(
        &self,
        deck_uuid: &SyncUuid,
    ) -> Result<Option<StoredSyncDeck>, SqliteSyncError> {
        let Some(metadata) = self.read_stored_deck_metadata(deck_uuid)? else {
            return Ok(None);
        };
        let (content_bundle, study_history_bundle) =
            if let Some(history_hash) = &metadata.study_history_bundle_hash {
                let history_bytes = read_sqlite_sync_bundle_file(
                    &self.root,
                    deck_uuid,
                    &sqlite_study_history_bundle_file_name(&metadata.revision),
                )?;
                (
                    None,
                    Some(SyncBundlePayload::from_parts(
                        history_bytes,
                        history_hash,
                        true,
                    )?),
                )
            } else {
                let content_bytes = read_sqlite_sync_bundle_file(
                    &self.root,
                    deck_uuid,
                    &sqlite_content_bundle_file_name(&metadata.revision),
                )?;
                (
                    Some(SyncBundlePayload::from_parts(
                        content_bytes,
                        &metadata.content_bundle_hash,
                        false,
                    )?),
                    None,
                )
            };

        Ok(Some(StoredSyncDeck {
            owner_identity: metadata.owner_identity,
            visibility: metadata.visibility,
            revision: metadata.revision,
            content_bundle_hash: metadata.content_bundle_hash,
            content_bundle,
            study_history_bundle,
            last_upload_device_name: metadata.last_upload_device_name,
            summary: metadata.summary,
        }))
    }

    fn read_stored_deck_metadata(
        &self,
        deck_uuid: &SyncUuid,
    ) -> Result<Option<StoredSyncDeckMetadata>, SqliteSyncError> {
        let connection = self.open_connection()?;
        read_sqlite_metadata(&connection, deck_uuid)
    }
}

fn sqlite_sync_schema_versions(connection: &Connection) -> Result<Vec<i64>, SqliteSyncError> {
    let mut statement =
        connection.prepare("SELECT version FROM sync_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 create_sqlite_sync_schema_v4(connection: &Connection) -> Result<(), SqliteSyncError> {
    connection.execute_batch(SQLITE_SYNC_SCHEMA_V4_SQL)?;
    Ok(())
}

fn set_sqlite_sync_schema_version_v4(connection: &Connection) -> Result<(), SqliteSyncError> {
    connection.execute_batch(
        "
        DROP TABLE IF EXISTS sync_schema_version;
        CREATE TABLE sync_schema_version (
            version INTEGER PRIMARY KEY CHECK (version = 4)
        );
        INSERT INTO sync_schema_version(version) VALUES (4);
        ",
    )?;
    Ok(())
}

fn ensure_sqlite_deck_identity_owners(connection: &Connection) -> Result<(), SqliteSyncError> {
    connection.execute_batch(
        "
        CREATE TABLE IF NOT EXISTS deck_identity_owners (
            deck_uuid TEXT PRIMARY KEY NOT NULL,
            owner_identity TEXT
        );
        CREATE INDEX IF NOT EXISTS idx_deck_identity_owners_owner
            ON deck_identity_owners(owner_identity, deck_uuid);
        INSERT OR IGNORE INTO deck_identity_owners (
            deck_uuid,
            owner_identity
        )
        SELECT deck_uuid,
               owner_identity
        FROM decks;
        ",
    )?;
    Ok(())
}

fn migrate_sqlite_sync_schema_v1_to_v4(connection: &mut Connection) -> Result<(), SqliteSyncError> {
    let transaction = connection.transaction()?;
    let deck_table_count: i64 = transaction.query_row(
        "
        SELECT COUNT(*)
        FROM sqlite_master
        WHERE type = 'table' AND name = 'decks'
        ",
        [],
        |row| row.get(0),
    )?;
    if deck_table_count > 0 {
        transaction.execute_batch(
            "
            ALTER TABLE decks RENAME TO decks_v1;
            CREATE TABLE decks (
                deck_uuid TEXT PRIMARY KEY NOT NULL,
                owner_identity TEXT NOT NULL,
                visibility TEXT NOT NULL,
                revision INTEGER NOT NULL,
                content_bundle_hash TEXT NOT NULL,
                study_history_bundle_hash TEXT,
                last_upload_device_name TEXT,
                deck_title TEXT,
                deck_name TEXT,
                deck_card_count INTEGER,
                deck_audio_file_count INTEGER,
                deck_managed_audio_file_count INTEGER,
                deck_audio_formats TEXT NOT NULL DEFAULT ''
            );
            INSERT INTO decks (
                deck_uuid,
                owner_identity,
                visibility,
                revision,
                content_bundle_hash,
                study_history_bundle_hash,
                last_upload_device_name,
                deck_title,
                deck_name,
                deck_card_count,
                deck_audio_file_count,
                deck_managed_audio_file_count,
                deck_audio_formats
            )
            SELECT deck_uuid,
                   owner_identity,
                   visibility,
                   revision,
                   content_bundle_hash,
                   study_history_bundle_hash,
                   last_upload_device_name,
                   NULL,
                   NULL,
                   NULL,
                   NULL,
                   NULL,
                   ''
            FROM decks_v1;
            DROP TABLE decks_v1;
            ",
        )?;
    }
    transaction.execute_batch(SQLITE_SYNC_SCHEMA_V4_SQL)?;
    transaction.execute_batch(
        "
        DROP TABLE IF EXISTS sync_schema_version;
        CREATE TABLE sync_schema_version (
            version INTEGER PRIMARY KEY CHECK (version = 4)
        );
        INSERT INTO sync_schema_version(version) VALUES (4);
        ",
    )?;
    transaction.commit()?;
    Ok(())
}

fn migrate_sqlite_sync_schema_v2_to_v4(connection: &mut Connection) -> Result<(), SqliteSyncError> {
    let transaction = connection.transaction()?;
    transaction.execute_batch(
        "
        ALTER TABLE decks ADD COLUMN deck_title TEXT;
        ALTER TABLE decks ADD COLUMN deck_name TEXT;
        ALTER TABLE decks ADD COLUMN deck_card_count INTEGER;
        ALTER TABLE decks ADD COLUMN deck_audio_file_count INTEGER;
        ALTER TABLE decks ADD COLUMN deck_managed_audio_file_count INTEGER;
        ALTER TABLE decks ADD COLUMN deck_audio_formats TEXT NOT NULL DEFAULT '';
        ",
    )?;
    transaction.execute_batch(SQLITE_SYNC_SCHEMA_V4_SQL)?;
    transaction.execute_batch(
        "
        DROP TABLE IF EXISTS sync_schema_version;
        CREATE TABLE sync_schema_version (
            version INTEGER PRIMARY KEY CHECK (version = 4)
        );
        INSERT INTO sync_schema_version(version) VALUES (4);
        ",
    )?;
    transaction.commit()?;
    Ok(())
}

fn migrate_sqlite_sync_schema_v3_to_v4(connection: &mut Connection) -> Result<(), SqliteSyncError> {
    ensure_sqlite_sync_v3_privacy_schema(connection)?;
    let transaction = connection.transaction()?;
    transaction.execute_batch(
        "
        ALTER TABLE oauth_pending_logins RENAME TO oauth_pending_logins_v3;
        CREATE TABLE oauth_pending_logins (
            login_token TEXT PRIMARY KEY NOT NULL,
            instance_url TEXT NOT NULL,
            client_id TEXT NOT NULL,
            client_secret TEXT NOT NULL,
            redirect_uri TEXT NOT NULL,
            expected_owner_identity TEXT,
            created_at_unix_seconds INTEGER NOT NULL,
            accepted_notice_version TEXT,
            accepted_notice_hash TEXT,
            accepted_lawful_basis TEXT,
            accepted_at_unix_seconds INTEGER
        );
        INSERT INTO oauth_pending_logins (
            login_token,
            instance_url,
            client_id,
            client_secret,
            redirect_uri,
            expected_owner_identity,
            created_at_unix_seconds,
            accepted_notice_version,
            accepted_notice_hash,
            accepted_lawful_basis,
            accepted_at_unix_seconds
        )
        SELECT login_token,
               instance_url,
               client_id,
               client_secret,
               redirect_uri,
               expected_owner_identity,
               created_at_unix_seconds,
               accepted_notice_version,
               accepted_notice_hash,
               accepted_lawful_basis,
               accepted_at_unix_seconds
        FROM oauth_pending_logins_v3;
        DROP TABLE oauth_pending_logins_v3;
        CREATE INDEX IF NOT EXISTS idx_oauth_pending_logins_instance
            ON oauth_pending_logins(instance_url);
        DROP TABLE IF EXISTS oauth_applications;
        DROP TABLE IF EXISTS sync_schema_version;
        CREATE TABLE sync_schema_version (
            version INTEGER PRIMARY KEY CHECK (version = 4)
        );
        INSERT INTO sync_schema_version(version) VALUES (4);
        ",
    )?;
    transaction.commit()?;
    Ok(())
}

fn ensure_sqlite_sync_v3_privacy_schema(connection: &Connection) -> Result<(), SqliteSyncError> {
    for (column, column_type) in [
        (
            "redirect_uri",
            "TEXT NOT NULL DEFAULT 'urn:ietf:wg:oauth:2.0:oob'",
        ),
        ("expected_owner_identity", "TEXT"),
        ("accepted_notice_version", "TEXT"),
        ("accepted_notice_hash", "TEXT"),
        ("accepted_lawful_basis", "TEXT"),
        ("accepted_at_unix_seconds", "INTEGER"),
    ] {
        if !sqlite_table_has_column(connection, "oauth_pending_logins", column)? {
            connection.execute_batch(&format!(
                "ALTER TABLE oauth_pending_logins ADD COLUMN {column} {column_type};"
            ))?;
        }
    }
    connection.execute_batch(
        "
        CREATE TABLE IF NOT EXISTS account_privacy_acceptances (
            owner_identity TEXT NOT NULL,
            notice_version TEXT NOT NULL,
            notice_hash TEXT NOT NULL,
            lawful_basis TEXT NOT NULL,
            accepted_at_unix_seconds INTEGER NOT NULL,
            login_token TEXT NOT NULL,
            PRIMARY KEY(owner_identity, notice_version, notice_hash)
        );
        CREATE TABLE IF NOT EXISTS oauth_completed_logins (
            login_token TEXT PRIMARY KEY NOT NULL,
            owner_identity TEXT NOT NULL,
            completed_at_unix_seconds INTEGER NOT NULL
        );
        ",
    )?;
    Ok(())
}

fn sqlite_table_has_column(
    connection: &Connection,
    table: &str,
    column: &str,
) -> Result<bool, SqliteSyncError> {
    let mut statement = connection.prepare(&format!("PRAGMA table_info({table})"))?;
    let rows = statement.query_map([], |row| row.get::<_, String>(1))?;
    for row in rows {
        if row? == column {
            return Ok(true);
        }
    }
    Ok(false)
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SyncAuthSession {
    owner_identity: String,
    bearer_token: String,
}

impl SyncAuthSession {
    pub fn from_parts(owner_identity: &str, bearer_token: &str) -> Result<Self, SyncAuthError> {
        let bearer_token =
            required_sync_text("bearer_token", bearer_token).map_err(SyncAuthError::Validation)?;
        if bearer_token.bytes().any(|byte| byte.is_ascii_whitespace()) {
            return Err(SyncAuthError::WhitespaceBearerToken);
        }
        Ok(Self {
            owner_identity: required_sync_text("owner_identity", owner_identity)
                .map_err(SyncAuthError::Validation)?,
            bearer_token,
        })
    }

    fn new(owner_identity: &str) -> Result<Self, SyncAuthError> {
        let first = SyncUuid::new_v4().map_err(SyncAuthError::TokenGeneration)?;
        let second = SyncUuid::new_v4().map_err(SyncAuthError::TokenGeneration)?;
        Self::from_parts(
            owner_identity,
            &format!(
                "cosimo_{}_{}",
                first.as_str().replace('-', ""),
                second.as_str().replace('-', "")
            ),
        )
    }

    pub fn owner_identity(&self) -> &str {
        &self.owner_identity
    }

    pub fn bearer_token(&self) -> &str {
        &self.bearer_token
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct SyncAccountDeletionOutcome {
    decks_deleted: usize,
    sessions_revoked: usize,
}

impl SyncAccountDeletionOutcome {
    pub fn from_parts(decks_deleted: usize, sessions_revoked: usize) -> Self {
        Self {
            decks_deleted,
            sessions_revoked,
        }
    }

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

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

#[derive(Debug)]
pub enum SyncAuthError {
    Validation(SyncValidationError),
    TokenGeneration(SyncUuidGenerationError),
    WhitespaceBearerToken,
    RateLimited(String),
    MissingPrivacyAcceptance,
}

impl fmt::Display for SyncAuthError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Validation(error) => write!(f, "{error}"),
            Self::TokenGeneration(error) => write!(f, "could not create sync auth token: {error}"),
            Self::WhitespaceBearerToken => {
                f.write_str("sync auth bearer token must not contain whitespace")
            }
            Self::RateLimited(message) => {
                write!(f, "sync OAuth login is temporarily limited: {message}")
            }
            Self::MissingPrivacyAcceptance => {
                f.write_str("sync privacy notice must be accepted before login can complete")
            }
        }
    }
}

impl std::error::Error for SyncAuthError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::Validation(error) => Some(error),
            Self::TokenGeneration(error) => Some(error),
            Self::WhitespaceBearerToken => None,
            Self::RateLimited(_) => None,
            Self::MissingPrivacyAcceptance => None,
        }
    }
}

impl From<SyncAuthError> for SqliteSyncError {
    fn from(error: SyncAuthError) -> Self {
        Self::Auth(error)
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
struct SyncOAuthStartAttempt {
    instance_url: String,
    client_key: String,
    attempted_at_unix_seconds: i64,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SyncOAuthPendingLogin {
    login_token: String,
    instance_url: String,
    client_id: String,
    client_secret: String,
    redirect_uri: String,
    expected_owner_identity: Option<String>,
    created_at_unix_seconds: i64,
    accepted_notice_version: Option<String>,
    accepted_notice_hash: Option<String>,
    accepted_lawful_basis: Option<String>,
    accepted_at_unix_seconds: Option<i64>,
}

#[derive(Debug, Clone, Copy)]
pub struct SyncOAuthPendingLoginParts<'a> {
    pub login_token: &'a str,
    pub instance_url: &'a str,
    pub client_id: &'a str,
    pub client_secret: &'a str,
    pub redirect_uri: &'a str,
    pub expected_owner_identity: Option<&'a str>,
    pub created_at_unix_seconds: i64,
    pub accepted_notice_version: Option<&'a str>,
    pub accepted_notice_hash: Option<&'a str>,
    pub accepted_lawful_basis: Option<&'a str>,
    pub accepted_at_unix_seconds: Option<i64>,
}

impl SyncOAuthPendingLogin {
    pub fn from_parts_with_redirect_uri(
        parts: SyncOAuthPendingLoginParts<'_>,
    ) -> Result<Self, SyncValidationError> {
        Ok(Self {
            login_token: required_sync_text("login_token", parts.login_token)?,
            instance_url: required_sync_text("instance_url", parts.instance_url)?,
            client_id: required_sync_text("client_id", parts.client_id)?,
            client_secret: required_sync_text("client_secret", parts.client_secret)?,
            redirect_uri: required_sync_text("redirect_uri", parts.redirect_uri)?,
            expected_owner_identity: optional_sync_text(
                "expected_owner_identity",
                parts.expected_owner_identity,
            )?,
            created_at_unix_seconds: parts.created_at_unix_seconds,
            accepted_notice_version: optional_sync_text(
                "accepted_notice_version",
                parts.accepted_notice_version,
            )?,
            accepted_notice_hash: optional_sync_text(
                "accepted_notice_hash",
                parts.accepted_notice_hash,
            )?,
            accepted_lawful_basis: optional_sync_text(
                "accepted_lawful_basis",
                parts.accepted_lawful_basis,
            )?,
            accepted_at_unix_seconds: parts.accepted_at_unix_seconds,
        })
    }

    fn with_expected_owner_identity(
        mut self,
        expected_owner_identity: Option<&str>,
    ) -> Result<Self, SyncValidationError> {
        self.expected_owner_identity =
            optional_sync_text("expected_owner_identity", expected_owner_identity)?;
        Ok(self)
    }

    fn new_with_redirect_uri(
        instance_url: &str,
        client_id: &str,
        client_secret: &str,
        redirect_uri: &str,
    ) -> Result<Self, SyncAuthError> {
        let login_token = SyncUuid::new_v4()
            .map_err(SyncAuthError::TokenGeneration)?
            .as_str()
            .replace('-', "");
        Self::from_parts_with_redirect_uri(SyncOAuthPendingLoginParts {
            login_token: &login_token,
            instance_url,
            client_id,
            client_secret,
            redirect_uri,
            expected_owner_identity: None,
            created_at_unix_seconds: current_unix_seconds(),
            accepted_notice_version: None,
            accepted_notice_hash: None,
            accepted_lawful_basis: None,
            accepted_at_unix_seconds: None,
        })
        .map_err(SyncAuthError::Validation)
    }

    fn new_with_redirect_uri_and_expected_owner_identity(
        instance_url: &str,
        client_id: &str,
        client_secret: &str,
        redirect_uri: &str,
        expected_owner_identity: &str,
    ) -> Result<Self, SyncAuthError> {
        Self::new_with_redirect_uri(instance_url, client_id, client_secret, redirect_uri).and_then(
            |pending| {
                pending
                    .with_expected_owner_identity(Some(expected_owner_identity))
                    .map_err(SyncAuthError::Validation)
            },
        )
    }

    fn configure_for_browser(
        &mut self,
        instance_url: &str,
        client_id: &str,
        client_secret: &str,
        redirect_uri: &str,
        expected_owner_identity: Option<&str>,
    ) -> Result<(), SyncAuthError> {
        self.instance_url =
            required_sync_text("instance_url", instance_url).map_err(SyncAuthError::Validation)?;
        self.client_id =
            required_sync_text("client_id", client_id).map_err(SyncAuthError::Validation)?;
        self.client_secret = required_sync_text("client_secret", client_secret)
            .map_err(SyncAuthError::Validation)?;
        self.redirect_uri =
            required_sync_text("redirect_uri", redirect_uri).map_err(SyncAuthError::Validation)?;
        self.expected_owner_identity =
            optional_sync_text("expected_owner_identity", expected_owner_identity)
                .map_err(SyncAuthError::Validation)?;
        Ok(())
    }

    pub fn login_token(&self) -> &str {
        &self.login_token
    }

    pub fn instance_url(&self) -> &str {
        &self.instance_url
    }

    pub fn client_id(&self) -> &str {
        &self.client_id
    }

    pub fn client_secret(&self) -> &str {
        &self.client_secret
    }

    pub fn redirect_uri(&self) -> &str {
        &self.redirect_uri
    }

    pub fn expected_owner_identity(&self) -> Option<&str> {
        self.expected_owner_identity.as_deref()
    }

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

    pub fn accepted_notice_version(&self) -> Option<&str> {
        self.accepted_notice_version.as_deref()
    }

    pub fn accepted_notice_hash(&self) -> Option<&str> {
        self.accepted_notice_hash.as_deref()
    }

    pub fn accepted_lawful_basis(&self) -> Option<&str> {
        self.accepted_lawful_basis.as_deref()
    }

    pub fn accepted_at_unix_seconds(&self) -> Option<i64> {
        self.accepted_at_unix_seconds
    }

    pub fn has_privacy_acceptance(&self) -> bool {
        self.accepted_notice_version.is_some()
            && self.accepted_notice_hash.is_some()
            && self.accepted_lawful_basis.is_some()
            && self.accepted_at_unix_seconds.is_some()
    }

    fn accept_privacy_notice(
        &mut self,
        notice_version: &str,
        notice_hash: &str,
        lawful_basis: &str,
    ) -> Result<(), SyncAuthError> {
        self.accepted_notice_version = Some(
            required_sync_text("accepted_notice_version", notice_version)
                .map_err(SyncAuthError::Validation)?,
        );
        self.accepted_notice_hash = Some(
            required_sync_text("accepted_notice_hash", notice_hash)
                .map_err(SyncAuthError::Validation)?,
        );
        self.accepted_lawful_basis = Some(
            required_sync_text("accepted_lawful_basis", lawful_basis)
                .map_err(SyncAuthError::Validation)?,
        );
        self.accepted_at_unix_seconds = Some(current_unix_seconds());
        Ok(())
    }

    fn privacy_acceptance_for_owner(
        &self,
        owner_identity: &str,
    ) -> Option<SyncAccountPrivacyAcceptance> {
        Some(SyncAccountPrivacyAcceptance {
            owner_identity: required_sync_text("owner_identity", owner_identity).ok()?,
            notice_version: self.accepted_notice_version.clone()?,
            notice_hash: self.accepted_notice_hash.clone()?,
            lawful_basis: self.accepted_lawful_basis.clone()?,
            accepted_at_unix_seconds: self.accepted_at_unix_seconds?,
            login_token: self.login_token.clone(),
        })
    }
}

#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
struct SyncAccountPrivacyAcceptanceKey {
    owner_identity: String,
    notice_version: String,
    notice_hash: String,
}

#[derive(Debug, Clone, PartialEq, Eq)]
struct SyncAccountPrivacyAcceptance {
    owner_identity: String,
    notice_version: String,
    notice_hash: String,
    lawful_basis: String,
    accepted_at_unix_seconds: i64,
    login_token: String,
}

impl SyncAccountPrivacyAcceptance {
    fn key(&self) -> SyncAccountPrivacyAcceptanceKey {
        SyncAccountPrivacyAcceptanceKey {
            owner_identity: self.owner_identity.clone(),
            notice_version: self.notice_version.clone(),
            notice_hash: self.notice_hash.clone(),
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SyncOAuthStart {
    login_token: String,
    authorization_url: String,
    registration_url: Option<String>,
}

impl SyncOAuthStart {
    pub fn from_parts(
        login_token: &str,
        authorization_url: &str,
    ) -> Result<Self, SyncValidationError> {
        Self::from_parts_with_registration_url(login_token, authorization_url, None)
    }

    pub fn from_parts_with_registration_url(
        login_token: &str,
        authorization_url: &str,
        registration_url: Option<&str>,
    ) -> Result<Self, SyncValidationError> {
        Ok(Self {
            login_token: required_sync_text("login_token", login_token)?,
            authorization_url: required_sync_text("authorization_url", authorization_url)?,
            registration_url: optional_sync_text("registration_url", registration_url)?,
        })
    }

    pub fn login_token(&self) -> &str {
        &self.login_token
    }

    pub fn authorization_url(&self) -> &str {
        &self.authorization_url
    }

    pub fn registration_url(&self) -> Option<&str> {
        self.registration_url.as_deref()
    }
}

const SQLITE_SYNC_DATABASE_FILE: &str = "cosimo-sync.sqlite3";
const SQLITE_SYNC_BUNDLES_DIR: &str = "bundles";

#[derive(Debug, Clone, PartialEq, Eq)]
struct StoredSyncDeck {
    owner_identity: String,
    visibility: DeckVisibility,
    revision: String,
    content_bundle: Option<SyncBundlePayload>,
    content_bundle_hash: String,
    study_history_bundle: Option<SyncBundlePayload>,
    last_upload_device_name: Option<String>,
    summary: SyncDeckSummary,
}

#[derive(Debug, Clone, PartialEq, Eq)]
struct StoredSyncDeckMetadata {
    deck_uuid: SyncUuid,
    owner_identity: String,
    visibility: DeckVisibility,
    revision: String,
    content_bundle_hash: String,
    study_history_bundle_hash: Option<String>,
    last_upload_device_name: Option<String>,
    summary: SyncDeckSummary,
}

impl From<StoredSyncDeckMetadata> for SyncRemoteDeckStatus {
    fn from(metadata: StoredSyncDeckMetadata) -> Self {
        Self {
            deck_uuid: metadata.deck_uuid,
            revision: metadata.revision,
            visibility: metadata.visibility,
            owner_identity: metadata.owner_identity,
            content_bundle_hash: metadata.content_bundle_hash,
            study_history_bundle_hash: metadata.study_history_bundle_hash,
            last_upload_device_name: metadata.last_upload_device_name,
            summary: metadata.summary,
        }
    }
}

impl SyncRemoteDeckStatus {
    fn from_stored(deck_uuid: SyncUuid, deck: &StoredSyncDeck) -> Self {
        Self {
            deck_uuid,
            revision: deck.revision.clone(),
            visibility: deck.visibility,
            owner_identity: deck.owner_identity.clone(),
            content_bundle_hash: deck.content_bundle_hash.clone(),
            study_history_bundle_hash: deck
                .study_history_bundle
                .as_ref()
                .map(|bundle| bundle.bundle_hash().to_string()),
            last_upload_device_name: deck.last_upload_device_name.clone(),
            summary: deck.summary.clone(),
        }
    }
}

impl SyncSharedDeckStatus {
    fn from_stored(deck_uuid: SyncUuid, deck: &StoredSyncDeck) -> Self {
        Self {
            deck_uuid,
            revision: deck.revision.clone(),
            visibility: deck.visibility,
            owner_identity: deck.owner_identity.clone(),
            content_bundle_hash: deck.content_bundle_hash.clone(),
            summary: deck.summary.clone(),
        }
    }

    fn from_metadata(metadata: StoredSyncDeckMetadata) -> Self {
        Self {
            deck_uuid: metadata.deck_uuid,
            revision: metadata.revision,
            visibility: metadata.visibility,
            owner_identity: metadata.owner_identity,
            content_bundle_hash: metadata.content_bundle_hash,
            summary: metadata.summary,
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SyncUploadError {
    Conflict {
        expected: Option<String>,
        actual: Option<String>,
    },
    Forbidden,
    ReservedDeckIdentity,
    RetiredDeckIdentity,
    QuotaExceeded(String),
    InvalidBundle(String),
}

impl fmt::Display for SyncUploadError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Conflict { expected, actual } => write!(
                f,
                "remote revision conflict: expected {}, found {}",
                expected.as_deref().unwrap_or("none"),
                actual.as_deref().unwrap_or("none")
            ),
            Self::Forbidden => f.write_str("sync upload is not allowed for this owner"),
            Self::ReservedDeckIdentity => f.write_str(
                "sync upload is not allowed because this deck UUID is reserved for another owner",
            ),
            Self::RetiredDeckIdentity => {
                f.write_str("sync upload is not allowed because this deck UUID has been retired")
            }
            Self::QuotaExceeded(message) => write!(f, "sync upload quota exceeded: {message}"),
            Self::InvalidBundle(message) => write!(f, "invalid sync bundle: {message}"),
        }
    }
}

impl std::error::Error for SyncUploadError {}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SyncDownloadError {
    NotFound,
    Forbidden,
}

impl fmt::Display for SyncDownloadError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::NotFound => f.write_str("remote deck was not found"),
            Self::Forbidden => f.write_str("sync download is not allowed"),
        }
    }
}

impl std::error::Error for SyncDownloadError {}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SyncDeleteError {
    NotFound,
    Forbidden,
}

impl fmt::Display for SyncDeleteError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::NotFound => f.write_str("remote deck was not found"),
            Self::Forbidden => f.write_str("sync deletion is not allowed"),
        }
    }
}

impl std::error::Error for SyncDeleteError {}

#[derive(Debug)]
pub enum SqliteSyncError {
    IoPath {
        action: &'static str,
        path: PathBuf,
        source: io::Error,
    },
    Database(rusqlite::Error),
    InvalidMetadata(String),
    Upload(SyncUploadError),
    Download(SyncDownloadError),
    Delete(SyncDeleteError),
    Validation(SyncValidationError),
    Auth(SyncAuthError),
}

impl SqliteSyncError {
    pub fn is_database_busy(&self) -> bool {
        matches!(
            self,
            Self::Database(rusqlite::Error::SqliteFailure(error, _))
                if matches!(
                    error.code,
                    rusqlite::ErrorCode::DatabaseBusy | rusqlite::ErrorCode::DatabaseLocked
                )
        )
    }
}

impl fmt::Display for SqliteSyncError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::IoPath {
                action,
                path,
                source,
            } => {
                write!(f, "{action}: {}: {source}", path.display())
            }
            Self::Database(error) => write!(f, "sync server database error: {error}"),
            Self::InvalidMetadata(message) => {
                write!(f, "invalid sync server metadata: {message}")
            }
            Self::Upload(error) => error.fmt(f),
            Self::Download(error) => error.fmt(f),
            Self::Delete(error) => error.fmt(f),
            Self::Validation(error) => error.fmt(f),
            Self::Auth(error) => error.fmt(f),
        }
    }
}

impl std::error::Error for SqliteSyncError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::IoPath { source, .. } => Some(source),
            Self::Database(error) => Some(error),
            Self::Upload(error) => Some(error),
            Self::Download(error) => Some(error),
            Self::Delete(error) => Some(error),
            Self::Validation(error) => Some(error),
            Self::Auth(error) => Some(error),
            Self::InvalidMetadata(_) => None,
        }
    }
}

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

impl From<SyncUploadError> for SqliteSyncError {
    fn from(error: SyncUploadError) -> Self {
        Self::Upload(error)
    }
}

impl From<SyncDownloadError> for SqliteSyncError {
    fn from(error: SyncDownloadError) -> Self {
        Self::Download(error)
    }
}

impl From<SyncDeleteError> for SqliteSyncError {
    fn from(error: SyncDeleteError) -> Self {
        Self::Delete(error)
    }
}

impl From<SyncValidationError> for SqliteSyncError {
    fn from(error: SyncValidationError) -> Self {
        Self::Validation(error)
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SyncValidationError {
    EmptyValue { field: &'static str },
    MultilineValue { field: &'static str },
    BundleHashMismatch { declared: String, actual: String },
    MissingUploadBundle,
    ContentBundleContainsReviewDatabase,
    StudyHistoryBundleMissingReviewDatabase,
}

impl fmt::Display for SyncValidationError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::EmptyValue { field } => write!(f, "{field} must not be empty"),
            Self::MultilineValue { field } => write!(f, "{field} must fit on one line"),
            Self::BundleHashMismatch { declared, actual } => write!(
                f,
                "sync bundle hash mismatch: declared {declared}, actual {actual}"
            ),
            Self::MissingUploadBundle => {
                f.write_str("sync upload must include a content or study-history bundle")
            }
            Self::ContentBundleContainsReviewDatabase => {
                f.write_str("content-only sync bundles must not include a review database")
            }
            Self::StudyHistoryBundleMissingReviewDatabase => {
                f.write_str("study-history sync bundles must include a review database")
            }
        }
    }
}

impl std::error::Error for SyncValidationError {}

fn required_sync_text(field: &'static str, value: &str) -> Result<String, SyncValidationError> {
    optional_sync_text(field, Some(value))?.ok_or(SyncValidationError::EmptyValue { field })
}

fn optional_sync_text(
    field: &'static str,
    value: Option<&str>,
) -> Result<Option<String>, SyncValidationError> {
    let Some(value) = value else {
        return Ok(None);
    };
    let value = value.trim();
    if value.is_empty() {
        return Ok(None);
    }
    if value.contains('\n') || value.contains('\r') {
        return Err(SyncValidationError::MultilineValue { field });
    }
    Ok(Some(value.to_string()))
}

fn normalise_sync_audio_formats(
    audio_formats: Vec<String>,
) -> Result<Vec<String>, SyncValidationError> {
    let mut formats = BTreeSet::new();
    for audio_format in audio_formats {
        let Some(audio_format) = optional_sync_text("audio_format", Some(&audio_format))? else {
            continue;
        };
        let audio_format = audio_format.trim_start_matches('.').to_ascii_lowercase();
        if is_sync_audio_extension(&audio_format) {
            formats.insert(audio_format);
        }
    }
    Ok(formats.into_iter().collect())
}

fn sync_audio_formats_storage(audio_formats: &[String]) -> String {
    audio_formats.join(",")
}

fn sync_audio_formats_from_storage(value: &str) -> Result<Vec<String>, SyncValidationError> {
    if value.trim().is_empty() {
        return Ok(Vec::new());
    }
    normalise_sync_audio_formats(value.split(',').map(str::to_string).collect())
}

fn optional_sqlite_usize(
    value: Option<i64>,
    field: &'static str,
) -> Result<Option<usize>, SqliteSyncError> {
    value
        .map(|value| {
            usize::try_from(value).map_err(|_| {
                SqliteSyncError::InvalidMetadata(format!("{field} must be a non-negative integer"))
            })
        })
        .transpose()
}

fn read_sqlite_metadata(
    connection: &Connection,
    deck_uuid: &SyncUuid,
) -> Result<Option<StoredSyncDeckMetadata>, SqliteSyncError> {
    let mut statement = connection.prepare(
        "
        SELECT deck_uuid,
               owner_identity,
               visibility,
               revision,
               content_bundle_hash,
               study_history_bundle_hash,
               last_upload_device_name,
               deck_title,
               deck_name,
               deck_card_count,
               deck_audio_file_count,
               deck_managed_audio_file_count,
               deck_audio_formats
        FROM decks
        WHERE deck_uuid = ?1
        ",
    )?;
    let mut rows = statement.query(params![deck_uuid.as_str()])?;
    let Some(row) = rows.next()? else {
        return Ok(None);
    };
    Ok(Some(sqlite_metadata_from_row(row)?))
}

fn read_sqlite_metadata_in_transaction(
    transaction: &rusqlite::Transaction<'_>,
    deck_uuid: &SyncUuid,
) -> Result<Option<StoredSyncDeckMetadata>, SqliteSyncError> {
    let mut statement = transaction.prepare(
        "
        SELECT deck_uuid,
               owner_identity,
               visibility,
               revision,
               content_bundle_hash,
               study_history_bundle_hash,
               last_upload_device_name,
               deck_title,
               deck_name,
               deck_card_count,
               deck_audio_file_count,
               deck_managed_audio_file_count,
               deck_audio_formats
        FROM decks
        WHERE deck_uuid = ?1
        ",
    )?;
    let mut rows = statement.query(params![deck_uuid.as_str()])?;
    let Some(row) = rows.next()? else {
        return Ok(None);
    };
    Ok(Some(sqlite_metadata_from_row(row)?))
}

fn read_sqlite_deck_identity_owner_in_transaction(
    transaction: &Transaction<'_>,
    deck_uuid: &SyncUuid,
) -> Result<Option<Option<String>>, SqliteSyncError> {
    let mut statement = transaction.prepare(
        "
        SELECT owner_identity
        FROM deck_identity_owners
        WHERE deck_uuid = ?1
        ",
    )?;
    let mut rows = statement.query(params![deck_uuid.as_str()])?;
    let Some(row) = rows.next()? else {
        return Ok(None);
    };
    Ok(Some(row.get(0)?))
}

fn sqlite_metadata_from_row(row: &Row<'_>) -> Result<StoredSyncDeckMetadata, SqliteSyncError> {
    let deck_uuid_text: String = row.get(0)?;
    let deck_uuid = deck_uuid_text.parse::<SyncUuid>().map_err(|error| {
        SqliteSyncError::InvalidMetadata(format!("invalid deck UUID in database: {error}"))
    })?;
    let owner_identity: String = row.get(1)?;
    let visibility_code: String = row.get(2)?;
    let visibility = DeckVisibility::from_code(&visibility_code)
        .ok_or_else(|| SqliteSyncError::InvalidMetadata("unknown deck visibility".to_string()))?;
    let revision_number: i64 = row.get(3)?;
    if revision_number < 1 {
        return Err(SqliteSyncError::InvalidMetadata(
            "stored revision must be positive".to_string(),
        ));
    }
    let content_bundle_hash: String = row.get(4)?;
    let study_history_bundle_hash: Option<String> = row.get(5)?;
    let last_upload_device_name: Option<String> = row.get(6)?;
    let deck_title: Option<String> = row.get(7)?;
    let deck_name: Option<String> = row.get(8)?;
    let deck_card_count = optional_sqlite_usize(row.get(9)?, "deck_card_count")?;
    let deck_audio_file_count = optional_sqlite_usize(row.get(10)?, "deck_audio_file_count")?;
    let deck_managed_audio_file_count =
        optional_sqlite_usize(row.get(11)?, "deck_managed_audio_file_count")?;
    let deck_audio_formats: String = row.get(12)?;
    let summary = SyncDeckSummary::from_parts(
        deck_title.as_deref(),
        deck_name.as_deref(),
        deck_card_count,
        deck_audio_file_count,
        deck_managed_audio_file_count,
        sync_audio_formats_from_storage(&deck_audio_formats)?,
    )
    .map_err(|error| SqliteSyncError::InvalidMetadata(error.to_string()))?;

    Ok(StoredSyncDeckMetadata {
        deck_uuid,
        owner_identity,
        visibility,
        revision: revision_number.to_string(),
        content_bundle_hash,
        study_history_bundle_hash,
        last_upload_device_name,
        summary,
    })
}

fn sync_deck_summary_from_upload_request(
    request: &SyncUploadRequest,
) -> Result<SyncDeckSummary, SyncUploadError> {
    validate_sync_upload_bundle_payloads(request, SyncServerLimits::default())
}

fn validate_sync_upload_bundle_payloads(
    request: &SyncUploadRequest,
    limits: SyncServerLimits,
) -> Result<SyncDeckSummary, SyncUploadError> {
    if let Some(study_history_bundle) = request.study_history_bundle() {
        validate_sync_bundle_payload(study_history_bundle, limits)
    } else if let Some(content_bundle) = request.content_bundle() {
        validate_sync_bundle_payload(content_bundle, limits)
    } else {
        Ok(SyncDeckSummary::unknown())
    }
}

fn validate_sync_bundle_payload(
    bundle: &SyncBundlePayload,
    limits: SyncServerLimits,
) -> Result<SyncDeckSummary, SyncUploadError> {
    if let Some(max_bytes) = limits.max_bundle_bytes
        && bundle.bytes().len() as u64 > max_bytes
    {
        return Err(SyncUploadError::QuotaExceeded(format!(
            "bundle is {} bytes, limit is {max_bytes} bytes",
            bundle.bytes().len()
        )));
    }

    let cursor = Cursor::new(bundle.bytes());
    let mut archive = ZipArchive::new(cursor).map_err(|error| {
        SyncUploadError::InvalidBundle(format!("bundle is not a zip file: {error}"))
    })?;
    if let Some(max_entries) = limits.max_bundle_entries
        && archive.len() > max_entries
    {
        return Err(SyncUploadError::QuotaExceeded(format!(
            "bundle contains {} entries, limit is {max_entries}",
            archive.len()
        )));
    }

    let mut seen_paths = std::collections::HashSet::new();
    let mut entries = Vec::new();
    let mut deck_entry_index = None;
    let mut deck_relative_path = None;
    let mut expanded_bytes = 0u64;
    for index in 0..archive.len() {
        let entry = archive
            .by_index(index)
            .map_err(|error| SyncUploadError::InvalidBundle(error.to_string()))?;
        let relative_path = safe_sync_bundle_path(entry.name(), entry.enclosed_name())?;
        if !seen_paths.insert(relative_path.clone()) {
            return Err(SyncUploadError::InvalidBundle(format!(
                "duplicate bundle entry: {}",
                relative_path.display()
            )));
        }
        entries.push((index, relative_path.clone(), entry.is_dir()));
        expanded_bytes = expanded_bytes.saturating_add(entry.size());
        if let Some(max_expanded_bytes) = limits.max_bundle_expanded_bytes
            && expanded_bytes > max_expanded_bytes
        {
            return Err(SyncUploadError::QuotaExceeded(format!(
                "bundle expands to more than {max_expanded_bytes} bytes"
            )));
        }
        if is_sync_root_deck_path(&relative_path) {
            if entry.is_dir() {
                return Err(SyncUploadError::InvalidBundle(
                    "deck bundle entry is a directory".to_string(),
                ));
            }
            if deck_entry_index.is_some() {
                return Err(SyncUploadError::InvalidBundle(
                    "bundle contains more than one root deck file".to_string(),
                ));
            }
            deck_entry_index = Some(index);
            deck_relative_path = Some(relative_path);
        }
    }

    let Some(deck_entry_index) = deck_entry_index else {
        return Err(SyncUploadError::InvalidBundle(
            "bundle does not contain a root .cosimo-deck file".to_string(),
        ));
    };
    let deck_relative_path = deck_relative_path.expect("deck index and path are set together");
    let deck_file_name = deck_relative_path
        .file_name()
        .and_then(|name| name.to_str())
        .ok_or_else(|| SyncUploadError::InvalidBundle("deck file name is invalid".to_string()))?;
    let deck_stem = Path::new(deck_file_name)
        .file_stem()
        .and_then(|stem| stem.to_str())
        .filter(|stem| !stem.is_empty())
        .ok_or_else(|| SyncUploadError::InvalidBundle("deck file stem is invalid".to_string()))?;
    let expected_review_db_name = format!("{deck_stem}{SYNC_REVIEW_DB_SUFFIX}");
    let mut has_review_database = false;
    let mut audio_file_names = BTreeSet::new();
    let mut audio_formats = BTreeSet::new();
    let mut manifest_entry_indices = Vec::new();
    for (index, relative_path, is_dir) in &entries {
        validate_sync_bundle_entry(
            relative_path,
            *is_dir,
            &deck_relative_path,
            deck_stem,
            &expected_review_db_name,
        )?;
        if relative_path == Path::new(&expected_review_db_name) {
            has_review_database = true;
        }
        if *is_dir {
            continue;
        }
        if is_sync_audio_manifest_path(relative_path) {
            manifest_entry_indices.push(*index);
        } else if let Some(extension) = relative_path
            .extension()
            .and_then(|extension| extension.to_str())
            && is_sync_audio_extension(extension)
        {
            audio_formats.insert(extension.to_ascii_lowercase());
            if let Some(file_name) = sync_sidecar_relative_file_name(relative_path, deck_stem) {
                audio_file_names.insert(file_name);
            }
        }
    }

    if has_review_database != bundle.includes_review_database() {
        return Err(SyncUploadError::InvalidBundle(
            if bundle.includes_review_database() {
                "study-history bundle does not contain the expected root review database"
                    .to_string()
            } else {
                "content-only bundle contains a review database".to_string()
            },
        ));
    }

    let deck_text = read_sync_deck_zip_entry_to_string(
        &mut archive,
        deck_entry_index,
        limits.max_deck_file_bytes,
    )?;
    let deck_file = parse_deck_file(&deck_text).map_err(|error| {
        SyncUploadError::InvalidBundle(format!("deck file is invalid: {error}"))
    })?;
    let manifest_audio_file_names =
        read_sync_audio_manifest_file_names(&mut archive, &manifest_entry_indices);
    let managed_audio_file_count = audio_file_names
        .iter()
        .filter(|file_name| manifest_audio_file_names.contains(*file_name))
        .count();
    SyncDeckSummary::from_parts(
        deck_file.metadata().get("title"),
        Some(deck_stem),
        Some(deck_file.data().len()),
        Some(audio_file_names.len()),
        Some(managed_audio_file_count),
        audio_formats.into_iter().collect(),
    )
    .map_err(|error| SyncUploadError::InvalidBundle(error.to_string()))
}

fn safe_sync_bundle_path(
    raw_name: &str,
    enclosed_name: Option<PathBuf>,
) -> Result<PathBuf, SyncUploadError> {
    if raw_name.is_empty() {
        return Err(SyncUploadError::InvalidBundle(
            "bundle entry names must not be empty".to_string(),
        ));
    }
    if raw_name.contains('\\') {
        return Err(SyncUploadError::InvalidBundle(format!(
            "bundle entry uses a backslash path separator: {raw_name}"
        )));
    }
    let Some(path) = enclosed_name else {
        return Err(SyncUploadError::InvalidBundle(format!(
            "bundle entry has an unsafe path: {raw_name}"
        )));
    };
    if path.as_os_str().is_empty() {
        return Err(SyncUploadError::InvalidBundle(
            "bundle entry names must not be empty".to_string(),
        ));
    }
    for component in path.components() {
        match component {
            Component::Normal(component) => {
                let Some(component) = component.to_str() else {
                    return Err(SyncUploadError::InvalidBundle(format!(
                        "bundle entry is not valid Unicode: {raw_name}"
                    )));
                };
                if component.is_empty()
                    || component.chars().any(|character| {
                        character.is_control()
                            || matches!(
                                character,
                                '<' | '>' | ':' | '"' | '/' | '\\' | '|' | '?' | '*'
                            )
                    })
                {
                    return Err(SyncUploadError::InvalidBundle(format!(
                        "bundle entry contains an unsupported path component: {raw_name}"
                    )));
                }
            }
            _ => {
                return Err(SyncUploadError::InvalidBundle(format!(
                    "bundle entry has an unsafe path: {raw_name}"
                )));
            }
        }
    }
    Ok(path)
}

fn validate_sync_bundle_entry(
    relative_path: &Path,
    is_dir: bool,
    deck_relative_path: &Path,
    deck_stem: &str,
    expected_review_db_name: &str,
) -> Result<(), SyncUploadError> {
    if relative_path == deck_relative_path {
        if is_dir {
            return Err(SyncUploadError::InvalidBundle(
                "deck bundle entry is a directory".to_string(),
            ));
        }
        return Ok(());
    }
    if relative_path == Path::new(expected_review_db_name) {
        if is_dir {
            return Err(SyncUploadError::InvalidBundle(
                "review database bundle entry is a directory".to_string(),
            ));
        }
        return Ok(());
    }
    let mut components = relative_path.components();
    let Some(Component::Normal(root)) = components.next() else {
        return Err(SyncUploadError::InvalidBundle(format!(
            "unexpected bundle entry: {}",
            relative_path.display()
        )));
    };
    if root.to_str() != Some(deck_stem) {
        return Err(SyncUploadError::InvalidBundle(format!(
            "unexpected bundle entry: {}",
            relative_path.display()
        )));
    }
    if is_dir {
        return Ok(());
    }
    if components.next().is_none() {
        return Err(SyncUploadError::InvalidBundle(format!(
            "sidecar root must be a directory: {}",
            relative_path.display()
        )));
    }
    if relative_path
        .file_name()
        .and_then(|name| name.to_str())
        .is_some_and(|name| name.eq_ignore_ascii_case(SYNC_AUDIO_MANIFEST_FILE_NAME))
    {
        return Ok(());
    }
    if relative_path
        .extension()
        .and_then(|extension| extension.to_str())
        .is_some_and(is_sync_audio_extension)
    {
        return Ok(());
    }
    Err(SyncUploadError::InvalidBundle(format!(
        "unexpected bundle entry: {}",
        relative_path.display()
    )))
}

fn is_sync_audio_manifest_path(relative_path: &Path) -> bool {
    relative_path
        .file_name()
        .and_then(|name| name.to_str())
        .is_some_and(|name| name.eq_ignore_ascii_case(SYNC_AUDIO_MANIFEST_FILE_NAME))
}

fn is_sync_audio_extension(extension: &str) -> bool {
    extension.eq_ignore_ascii_case("mp3")
        || extension.eq_ignore_ascii_case("opus")
        || extension.eq_ignore_ascii_case("wav")
}

fn sync_sidecar_relative_file_name(relative_path: &Path, deck_stem: &str) -> Option<String> {
    let mut components = relative_path.components();
    let Some(Component::Normal(root)) = components.next() else {
        return None;
    };
    if root.to_str() != Some(deck_stem) {
        return None;
    }
    let parts: Vec<&str> = components
        .filter_map(|component| match component {
            Component::Normal(part) => part.to_str(),
            _ => None,
        })
        .collect();
    (!parts.is_empty()).then(|| parts.join("/"))
}

fn is_sync_root_deck_path(path: &Path) -> bool {
    path.parent()
        .is_none_or(|parent| parent.as_os_str().is_empty())
        && path
            .extension()
            .and_then(|extension| extension.to_str())
            .is_some_and(|extension| extension.eq_ignore_ascii_case("cosimo-deck"))
}

fn read_sync_deck_zip_entry_to_string(
    archive: &mut ZipArchive<Cursor<&[u8]>>,
    index: usize,
    max_bytes: Option<u64>,
) -> Result<String, SyncUploadError> {
    let mut entry = archive
        .by_index(index)
        .map_err(|error| SyncUploadError::InvalidBundle(error.to_string()))?;
    if let Some(max_bytes) = max_bytes
        && entry.size() > max_bytes
    {
        return Err(SyncUploadError::QuotaExceeded(format!(
            "deck file is {} bytes, limit is {max_bytes} bytes",
            entry.size()
        )));
    }
    let mut text = String::new();
    let read_limit = max_bytes
        .and_then(|limit| limit.checked_add(1))
        .unwrap_or(u64::MAX);
    let bytes_read = entry
        .by_ref()
        .take(read_limit)
        .read_to_string(&mut text)
        .map_err(|error| SyncUploadError::InvalidBundle(error.to_string()))?;
    if let Some(max_bytes) = max_bytes
        && bytes_read as u64 > max_bytes
    {
        return Err(SyncUploadError::QuotaExceeded(format!(
            "deck file expands beyond {max_bytes} bytes"
        )));
    }
    Ok(text)
}

fn read_sync_audio_manifest_file_names(
    archive: &mut ZipArchive<Cursor<&[u8]>>,
    indices: &[usize],
) -> BTreeSet<String> {
    let mut file_names = BTreeSet::new();
    for index in indices {
        let Ok(mut entry) = archive.by_index(*index) else {
            continue;
        };
        let mut bytes = Vec::new();
        if entry.read_to_end(&mut bytes).is_err() {
            continue;
        }
        for line in String::from_utf8_lossy(&bytes).lines() {
            let Some((file_name, _)) = line.split_once('\t') else {
                continue;
            };
            let file_name = file_name.trim();
            if file_name.is_empty() || file_name == "version" {
                continue;
            }
            file_names.insert(file_name.to_string());
        }
    }
    file_names
}

fn enforce_oauth_start_attempt_limit(
    current_count: usize,
    max_count: Option<usize>,
    scope: &'static str,
) -> Result<(), SyncAuthError> {
    if let Some(max_count) = max_count
        && current_count >= max_count
    {
        return Err(SyncAuthError::RateLimited(format!(
            "too many login starts for this {scope}; try again later"
        )));
    }
    Ok(())
}

fn enforce_pending_oauth_login_limit(
    current_count: usize,
    max_count: Option<usize>,
) -> Result<(), SyncAuthError> {
    if let Some(max_count) = max_count
        && current_count >= max_count
    {
        return Err(SyncAuthError::RateLimited(
            "too many pending logins for this instance; complete an existing login or try again later"
                .to_string(),
        ));
    }
    Ok(())
}

fn prune_sqlite_oauth_pending_logins(
    transaction: &Transaction<'_>,
    now_unix_seconds: i64,
) -> Result<(), SqliteSyncError> {
    transaction.execute(
        "
        DELETE FROM oauth_pending_logins
        WHERE created_at_unix_seconds < ?1
        ",
        params![now_unix_seconds - SYNC_OAUTH_PENDING_LOGIN_TTL_SECONDS],
    )?;
    Ok(())
}

fn prune_sqlite_oauth_start_attempts(
    transaction: &Transaction<'_>,
    now_unix_seconds: i64,
) -> Result<(), SqliteSyncError> {
    transaction.execute(
        "
        DELETE FROM oauth_start_attempts
        WHERE attempted_at_unix_seconds < ?1
        ",
        params![now_unix_seconds - SYNC_OAUTH_RATE_LIMIT_WINDOW_SECONDS],
    )?;
    Ok(())
}

fn upgrade_sqlite_auth_token_verifiers(connection: &Connection) -> Result<(), SqliteSyncError> {
    let mut statement = connection.prepare(
        "
        SELECT bearer_token
        FROM auth_sessions
        WHERE bearer_token NOT LIKE 'sha256:%'
        ",
    )?;
    let mut rows = statement.query([])?;
    let mut plaintext_tokens = Vec::new();
    while let Some(row) = rows.next()? {
        plaintext_tokens.push(row.get::<_, String>(0)?);
    }
    drop(rows);
    drop(statement);

    for plaintext_token in plaintext_tokens {
        connection.execute(
            "
            UPDATE auth_sessions
            SET bearer_token = ?1
            WHERE bearer_token = ?2
            ",
            params![
                sync_bearer_token_verifier(&plaintext_token),
                plaintext_token
            ],
        )?;
    }
    Ok(())
}

fn enforce_sqlite_sync_upload_limits(
    transaction: &Transaction<'_>,
    root: &Path,
    request: &SyncUploadRequest,
    current: Option<&StoredSyncDeckMetadata>,
    limits: SyncServerLimits,
) -> Result<(), SqliteSyncError> {
    if let Some(max_decks) = limits.max_decks_per_owner
        && current.is_none()
    {
        let deck_count =
            sqlite_owner_deck_count_in_transaction(transaction, request.owner_identity())?;
        if deck_count >= max_decks {
            return Err(SyncUploadError::QuotaExceeded(format!(
                "owner already has {deck_count} deck(s), limit is {max_decks}"
            ))
            .into());
        }
    }

    let incoming_bytes = stored_sync_upload_bytes(request);
    if let Some(max_owner_bytes) = limits.max_storage_bytes_per_owner {
        let current_bytes =
            sqlite_owner_storage_bytes_in_transaction(transaction, root, request.owner_identity())?;
        if current_bytes.saturating_add(incoming_bytes) > max_owner_bytes {
            return Err(SyncUploadError::QuotaExceeded(format!(
                "owner storage would be {} bytes, limit is {max_owner_bytes} bytes",
                current_bytes.saturating_add(incoming_bytes)
            ))
            .into());
        }
    }
    if let Some(max_total_bytes) = limits.max_storage_bytes_total {
        let current_bytes = sqlite_total_storage_bytes(root)?;
        if current_bytes.saturating_add(incoming_bytes) > max_total_bytes {
            return Err(SyncUploadError::QuotaExceeded(format!(
                "server storage would be {} bytes, limit is {max_total_bytes} bytes",
                current_bytes.saturating_add(incoming_bytes)
            ))
            .into());
        }
    }
    Ok(())
}

fn stored_sync_upload_bytes(request: &SyncUploadRequest) -> u64 {
    request
        .study_history_bundle()
        .or(request.content_bundle())
        .map(|bundle| bundle.bytes().len() as u64)
        .unwrap_or(0)
}

fn sqlite_owner_deck_count_in_transaction(
    transaction: &Transaction<'_>,
    owner_identity: &str,
) -> Result<usize, SqliteSyncError> {
    let count: i64 = transaction.query_row(
        "SELECT COUNT(*) FROM decks WHERE owner_identity = ?1",
        params![owner_identity],
        |row| row.get(0),
    )?;
    usize::try_from(count).map_err(|_| {
        SqliteSyncError::InvalidMetadata("owner deck count is out of range".to_string())
    })
}

fn sqlite_owner_storage_bytes_in_transaction(
    transaction: &Transaction<'_>,
    root: &Path,
    owner_identity: &str,
) -> Result<u64, SqliteSyncError> {
    let mut statement = transaction.prepare(
        "
        SELECT deck_uuid
        FROM decks
        WHERE owner_identity = ?1
        ",
    )?;
    let mut rows = statement.query(params![owner_identity])?;
    let mut total = 0u64;
    while let Some(row) = rows.next()? {
        let deck_uuid_text: String = row.get(0)?;
        let deck_uuid = deck_uuid_text.parse::<SyncUuid>().map_err(|error| {
            SqliteSyncError::InvalidMetadata(format!("invalid deck UUID in database: {error}"))
        })?;
        total = total.saturating_add(sqlite_deck_bundle_storage_bytes(root, &deck_uuid)?);
    }
    Ok(total)
}

fn sqlite_deck_bundle_storage_bytes(
    root: &Path,
    deck_uuid: &SyncUuid,
) -> Result<u64, SqliteSyncError> {
    let root_canonical = ensure_sqlite_sync_storage_root(root)?;
    let bundles_dir = root.join(SQLITE_SYNC_BUNDLES_DIR);
    if !bundles_dir.exists() {
        return Ok(0);
    }
    ensure_plain_existing_directory(
        &bundles_dir,
        "could not inspect sync server bundle directory",
    )?;
    canonical_contained_path(&bundles_dir, &root_canonical)?;
    let deck_dir = bundles_dir.join(deck_uuid.as_str());
    if !deck_dir.exists() {
        return Ok(0);
    }
    ensure_plain_existing_directory(&deck_dir, "could not inspect sync server bundle directory")?;
    canonical_contained_path(&deck_dir, &root_canonical)?;
    sum_plain_regular_files_one_level(&deck_dir)
}

fn sqlite_total_storage_bytes(root: &Path) -> Result<u64, SqliteSyncError> {
    let root_canonical = ensure_sqlite_sync_storage_root(root)?;
    sum_plain_regular_files_recursive(root, &root_canonical)
}

fn sum_plain_regular_files_one_level(directory: &Path) -> Result<u64, SqliteSyncError> {
    let mut total = 0u64;
    let entries = fs::read_dir(directory).map_err(|error| SqliteSyncError::IoPath {
        action: "could not read sync server storage directory",
        path: directory.to_path_buf(),
        source: error,
    })?;
    for entry in entries {
        let entry = entry.map_err(|error| SqliteSyncError::IoPath {
            action: "could not read sync server storage directory entry",
            path: directory.to_path_buf(),
            source: error,
        })?;
        let path = entry.path();
        let metadata = fs::symlink_metadata(&path).map_err(|error| SqliteSyncError::IoPath {
            action: "could not inspect sync server storage file",
            path: path.clone(),
            source: error,
        })?;
        if metadata.file_type().is_symlink() {
            return Err(sync_storage_invalid(format!(
                "sync server storage file must not be a symlink: {}",
                path.display()
            )));
        }
        if metadata.is_file() {
            total = total.saturating_add(metadata.len());
        }
    }
    Ok(total)
}

fn sum_plain_regular_files_recursive(
    directory: &Path,
    root_canonical: &Path,
) -> Result<u64, SqliteSyncError> {
    ensure_plain_existing_directory(directory, "could not inspect sync server storage directory")?;
    canonical_contained_path(directory, root_canonical)?;
    let mut total = 0u64;
    let entries = fs::read_dir(directory).map_err(|error| SqliteSyncError::IoPath {
        action: "could not read sync server storage directory",
        path: directory.to_path_buf(),
        source: error,
    })?;
    for entry in entries {
        let entry = entry.map_err(|error| SqliteSyncError::IoPath {
            action: "could not read sync server storage directory entry",
            path: directory.to_path_buf(),
            source: error,
        })?;
        let path = entry.path();
        let metadata = fs::symlink_metadata(&path).map_err(|error| SqliteSyncError::IoPath {
            action: "could not inspect sync server storage path",
            path: path.clone(),
            source: error,
        })?;
        if metadata.file_type().is_symlink() {
            return Err(sync_storage_invalid(format!(
                "sync server storage path must not be a symlink: {}",
                path.display()
            )));
        }
        if metadata.is_dir() {
            total = total.saturating_add(sum_plain_regular_files_recursive(&path, root_canonical)?);
        } else if metadata.is_file() {
            total = total.saturating_add(metadata.len());
        }
    }
    Ok(total)
}

#[cfg(test)]
fn sqlite_deck_bundle_dir(root: &Path, deck_uuid: &SyncUuid) -> PathBuf {
    root.join(SQLITE_SYNC_BUNDLES_DIR).join(deck_uuid.as_str())
}

fn sync_storage_invalid(message: impl Into<String>) -> SqliteSyncError {
    SqliteSyncError::InvalidMetadata(message.into())
}

fn validate_sync_storage_component(value: &str, field: &str) -> Result<(), SqliteSyncError> {
    let mut components = Path::new(value).components();
    match (components.next(), components.next()) {
        (Some(Component::Normal(_)), None) => Ok(()),
        _ => Err(sync_storage_invalid(format!(
            "{field} must be a single storage path component"
        ))),
    }
}

fn ensure_plain_existing_directory(
    path: &Path,
    action: &'static str,
) -> Result<(), SqliteSyncError> {
    let metadata = fs::symlink_metadata(path).map_err(|error| SqliteSyncError::IoPath {
        action,
        path: path.to_path_buf(),
        source: error,
    })?;
    if metadata.file_type().is_symlink() {
        return Err(sync_storage_invalid(format!(
            "sync server storage directory must not be a symlink: {}",
            path.display()
        )));
    }
    if !metadata.is_dir() {
        return Err(sync_storage_invalid(format!(
            "sync server storage path must be a directory: {}",
            path.display()
        )));
    }
    Ok(())
}

fn ensure_plain_directory(path: &Path, action: &'static str) -> Result<(), SqliteSyncError> {
    match fs::symlink_metadata(path) {
        Ok(metadata) => {
            if metadata.file_type().is_symlink() {
                return Err(sync_storage_invalid(format!(
                    "sync server storage directory must not be a symlink: {}",
                    path.display()
                )));
            }
            if !metadata.is_dir() {
                return Err(sync_storage_invalid(format!(
                    "sync server storage path must be a directory: {}",
                    path.display()
                )));
            }
            Ok(())
        }
        Err(error) if error.kind() == io::ErrorKind::NotFound => {
            fs::create_dir(path).map_err(|error| SqliteSyncError::IoPath {
                action,
                path: path.to_path_buf(),
                source: error,
            })?;
            ensure_plain_existing_directory(path, action)
        }
        Err(error) => Err(SqliteSyncError::IoPath {
            action,
            path: path.to_path_buf(),
            source: error,
        }),
    }
}

fn canonical_contained_path(path: &Path, root: &Path) -> Result<PathBuf, SqliteSyncError> {
    let canonical = path
        .canonicalize()
        .map_err(|error| SqliteSyncError::IoPath {
            action: "could not resolve sync server storage path",
            path: path.to_path_buf(),
            source: error,
        })?;
    if !canonical.starts_with(root) {
        return Err(sync_storage_invalid(format!(
            "sync server storage path escapes the storage directory: {}",
            path.display()
        )));
    }
    Ok(canonical)
}

fn ensure_sqlite_sync_storage_root(root: &Path) -> Result<PathBuf, SqliteSyncError> {
    if root.exists() {
        ensure_plain_existing_directory(root, "could not inspect sync server storage directory")?;
    } else {
        fs::create_dir_all(root).map_err(|error| SqliteSyncError::IoPath {
            action: "could not create sync server storage directory",
            path: root.to_path_buf(),
            source: error,
        })?;
        ensure_plain_existing_directory(root, "could not inspect sync server storage directory")?;
    }
    root.canonicalize()
        .map_err(|error| SqliteSyncError::IoPath {
            action: "could not resolve sync server storage directory",
            path: root.to_path_buf(),
            source: error,
        })
}

fn ensure_sync_child_directory(
    root: &Path,
    root_canonical: &Path,
    parent: &Path,
    component: &str,
    create: bool,
) -> Result<PathBuf, SqliteSyncError> {
    validate_sync_storage_component(component, "sync server storage directory")?;
    let path = parent.join(component);
    if create {
        ensure_plain_directory(&path, "could not create sync server storage directory")?;
    } else {
        ensure_plain_existing_directory(&path, "could not inspect sync server storage directory")?;
    }
    let canonical = canonical_contained_path(&path, root_canonical)?;
    if !canonical.starts_with(root_canonical) || !path.starts_with(root) {
        return Err(sync_storage_invalid(format!(
            "sync server storage path escapes the storage directory: {}",
            path.display()
        )));
    }
    Ok(path)
}

fn ensure_sqlite_deck_bundle_dir(
    root: &Path,
    deck_uuid: &SyncUuid,
    create: bool,
) -> Result<PathBuf, SqliteSyncError> {
    let root_canonical = ensure_sqlite_sync_storage_root(root)?;
    let bundles_dir =
        ensure_sync_child_directory(root, &root_canonical, root, SQLITE_SYNC_BUNDLES_DIR, create)?;
    ensure_sync_child_directory(
        root,
        &root_canonical,
        &bundles_dir,
        deck_uuid.as_str(),
        create,
    )
}

fn ensure_plain_regular_file_if_exists(
    path: &Path,
    action: &'static str,
) -> Result<bool, SqliteSyncError> {
    match fs::symlink_metadata(path) {
        Ok(metadata) => {
            if metadata.file_type().is_symlink() {
                return Err(sync_storage_invalid(format!(
                    "sync server storage file must not be a symlink: {}",
                    path.display()
                )));
            }
            if !metadata.is_file() {
                return Err(sync_storage_invalid(format!(
                    "sync server storage path must be a file: {}",
                    path.display()
                )));
            }
            Ok(true)
        }
        Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(false),
        Err(error) => Err(SqliteSyncError::IoPath {
            action,
            path: path.to_path_buf(),
            source: error,
        }),
    }
}

fn sqlite_sync_database_path(root: &Path) -> Result<PathBuf, SqliteSyncError> {
    validate_sync_storage_component(SQLITE_SYNC_DATABASE_FILE, "sync server database file")?;
    let root_canonical = ensure_sqlite_sync_storage_root(root)?;
    let db_path = root.join(SQLITE_SYNC_DATABASE_FILE);
    if !ensure_plain_regular_file_if_exists(
        &db_path,
        "could not inspect sync server database file",
    )? {
        OpenOptions::new()
            .write(true)
            .create_new(true)
            .open(&db_path)
            .map_err(|error| SqliteSyncError::IoPath {
                action: "could not create sync server database file",
                path: db_path.clone(),
                source: error,
            })?;
        ensure_plain_regular_file_if_exists(
            &db_path,
            "could not inspect sync server database file",
        )?;
    }
    canonical_contained_path(&db_path, &root_canonical)?;
    Ok(db_path)
}

fn sqlite_content_bundle_file_name(revision: &str) -> String {
    format!("content-r{revision}.bundle")
}

fn sqlite_study_history_bundle_file_name(revision: &str) -> String {
    format!("study-history-r{revision}.bundle")
}

fn sqlite_delete_tombstone_name(deck_uuid: &SyncUuid) -> String {
    format!(
        "{}{}-{}-{}",
        SQLITE_SYNC_DELETE_TOMBSTONE_PREFIX,
        deck_uuid.as_str(),
        std::process::id(),
        current_unix_seconds()
    )
}

fn current_unix_seconds() -> i64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|duration| duration.as_secs().min(i64::MAX as u64) as i64)
        .unwrap_or(0)
}

struct StagedSqliteUpload {
    content: Option<StagedSqliteBundle>,
    history: Option<StagedSqliteBundle>,
}

impl StagedSqliteUpload {
    fn promote(
        &self,
        root: &Path,
        deck_uuid: &SyncUuid,
        revision: &str,
    ) -> Result<(), SqliteSyncError> {
        if let Some(content) = &self.content {
            content.promote(root, deck_uuid, &sqlite_content_bundle_file_name(revision))?;
        }
        if let Some(history) = &self.history {
            history.promote(
                root,
                deck_uuid,
                &sqlite_study_history_bundle_file_name(revision),
            )?;
        }
        Ok(())
    }

    fn remove_temporary_files(
        &self,
        root: &Path,
        deck_uuid: &SyncUuid,
    ) -> Result<(), SqliteSyncError> {
        if let Some(content) = &self.content {
            content.remove_temporary_file(root, deck_uuid)?;
        }
        if let Some(history) = &self.history {
            history.remove_temporary_file(root, deck_uuid)?;
        }
        Ok(())
    }
}

struct StagedSqliteBundle {
    temporary_file_name: String,
}

impl StagedSqliteBundle {
    fn promote(
        &self,
        root: &Path,
        deck_uuid: &SyncUuid,
        destination_file_name: &str,
    ) -> Result<(), SqliteSyncError> {
        validate_sync_storage_component(destination_file_name, "sync server bundle file")?;
        let deck_dir = ensure_sqlite_deck_bundle_dir(root, deck_uuid, false)?;
        let temporary_path = deck_dir.join(&self.temporary_file_name);
        let destination_path = deck_dir.join(destination_file_name);
        if ensure_plain_regular_file_if_exists(
            &destination_path,
            "could not inspect existing sync server bundle file",
        )? {
            fs::remove_file(&destination_path).map_err(|error| SqliteSyncError::IoPath {
                action: "could not replace existing sync server bundle file",
                path: destination_path.clone(),
                source: error,
            })?;
        }
        fs::rename(&temporary_path, &destination_path).map_err(|error| SqliteSyncError::IoPath {
            action: "could not install sync server bundle file",
            path: destination_path,
            source: error,
        })
    }

    fn remove_temporary_file(
        &self,
        root: &Path,
        deck_uuid: &SyncUuid,
    ) -> Result<(), SqliteSyncError> {
        let deck_dir = ensure_sqlite_deck_bundle_dir(root, deck_uuid, false)?;
        let temporary_path = deck_dir.join(&self.temporary_file_name);
        if ensure_plain_regular_file_if_exists(
            &temporary_path,
            "could not inspect sync server temporary bundle file",
        )? {
            fs::remove_file(&temporary_path).map_err(|error| SqliteSyncError::IoPath {
                action: "could not remove sync server temporary bundle file",
                path: temporary_path,
                source: error,
            })?;
        }
        Ok(())
    }
}

fn stage_sqlite_sync_upload_bundles(
    root: &Path,
    request: &SyncUploadRequest,
) -> Result<StagedSqliteUpload, SqliteSyncError> {
    let content = if request.study_history_bundle().is_none() {
        Some(stage_sqlite_sync_bundle_file(
            root,
            request.deck_uuid(),
            "content",
            request
                .content_bundle()
                .expect("upload request validated at least one bundle")
                .bytes(),
        )?)
    } else {
        None
    };
    let history = if let Some(bundle) = request.study_history_bundle() {
        match stage_sqlite_sync_bundle_file(
            root,
            request.deck_uuid(),
            "study-history",
            bundle.bytes(),
        ) {
            Ok(staged) => Some(staged),
            Err(error) => {
                if let Some(content) = &content {
                    let _ = content.remove_temporary_file(root, request.deck_uuid());
                }
                return Err(error);
            }
        }
    } else {
        None
    };
    Ok(StagedSqliteUpload { content, history })
}

fn stage_sqlite_sync_bundle_file(
    root: &Path,
    deck_uuid: &SyncUuid,
    kind: &str,
    bytes: &[u8],
) -> Result<StagedSqliteBundle, SqliteSyncError> {
    validate_sync_storage_component(kind, "sync server bundle kind")?;
    let deck_dir = ensure_sqlite_deck_bundle_dir(root, deck_uuid, true)?;
    for attempt in 0..1000 {
        let temporary_file_name = format!(
            ".upload-{}-{}-{kind}-{attempt}.tmp",
            std::process::id(),
            current_unix_seconds()
        );
        validate_sync_storage_component(&temporary_file_name, "sync server temporary bundle file")?;
        let temporary_path = deck_dir.join(&temporary_file_name);
        match OpenOptions::new()
            .write(true)
            .create_new(true)
            .open(&temporary_path)
        {
            Ok(mut temporary_file) => {
                if let Err(error) = temporary_file.write_all(bytes) {
                    let _ = fs::remove_file(&temporary_path);
                    return Err(SqliteSyncError::IoPath {
                        action: "could not write sync server temporary bundle file",
                        path: temporary_path,
                        source: error,
                    });
                }
                return Ok(StagedSqliteBundle {
                    temporary_file_name,
                });
            }
            Err(error) if error.kind() == io::ErrorKind::AlreadyExists => continue,
            Err(error) => {
                return Err(SqliteSyncError::IoPath {
                    action: "could not create sync server temporary bundle file",
                    path: temporary_path,
                    source: error,
                });
            }
        }
    }
    Err(sync_storage_invalid(
        "could not allocate sync server temporary bundle file name",
    ))
}

fn remove_promoted_sqlite_sync_upload_bundles(
    root: &Path,
    deck_uuid: &SyncUuid,
    revision: &str,
) -> Result<(), SqliteSyncError> {
    let deck_dir = ensure_sqlite_deck_bundle_dir(root, deck_uuid, false)?;
    for file_name in [
        sqlite_content_bundle_file_name(revision),
        sqlite_study_history_bundle_file_name(revision),
    ] {
        let path = deck_dir.join(file_name);
        if ensure_plain_regular_file_if_exists(&path, "could not inspect sync server bundle file")?
        {
            fs::remove_file(&path).map_err(|error| SqliteSyncError::IoPath {
                action: "could not remove uncommitted sync server bundle file",
                path,
                source: error,
            })?;
        }
    }
    Ok(())
}

fn reconcile_sqlite_bundle_files(
    root: &Path,
    connection: &Connection,
) -> Result<(), SqliteSyncError> {
    let root_canonical = ensure_sqlite_sync_storage_root(root)?;
    let bundles_dir = root.join(SQLITE_SYNC_BUNDLES_DIR);
    if !bundles_dir.exists() {
        return Ok(());
    }
    ensure_plain_existing_directory(
        &bundles_dir,
        "could not inspect sync server bundle directory",
    )?;
    canonical_contained_path(&bundles_dir, &root_canonical)?;

    let current_revisions = sqlite_current_bundle_revisions(connection)?;
    let entries = fs::read_dir(&bundles_dir).map_err(|error| SqliteSyncError::IoPath {
        action: "could not read sync server bundle directory",
        path: bundles_dir.clone(),
        source: error,
    })?;
    for entry in entries {
        let entry = entry.map_err(|error| SqliteSyncError::IoPath {
            action: "could not read sync server bundle directory entry",
            path: bundles_dir.clone(),
            source: error,
        })?;
        let deck_dir = entry.path();
        let metadata =
            fs::symlink_metadata(&deck_dir).map_err(|error| SqliteSyncError::IoPath {
                action: "could not inspect sync server bundle path",
                path: deck_dir.clone(),
                source: error,
            })?;
        if metadata.file_type().is_symlink() {
            return Err(sync_storage_invalid(format!(
                "sync server bundle path must not be a symlink: {}",
                deck_dir.display()
            )));
        }
        if !metadata.is_dir() {
            continue;
        }
        canonical_contained_path(&deck_dir, &root_canonical)?;
        let Some(deck_uuid) = deck_dir
            .file_name()
            .and_then(|name| name.to_str())
            .and_then(|name| name.parse::<SyncUuid>().ok())
        else {
            continue;
        };
        reconcile_sqlite_deck_bundle_files(
            &deck_dir,
            current_revisions
                .get(deck_uuid.as_str())
                .map(String::as_str),
        )?;
    }
    Ok(())
}

fn sqlite_current_bundle_revisions(
    connection: &Connection,
) -> Result<BTreeMap<String, String>, SqliteSyncError> {
    let mut statement = connection.prepare("SELECT deck_uuid, revision FROM decks")?;
    let mut rows = statement.query([])?;
    let mut revisions = BTreeMap::new();
    while let Some(row) = rows.next()? {
        let deck_uuid_text: String = row.get(0)?;
        let revision_number: i64 = row.get(1)?;
        let deck_uuid = deck_uuid_text.parse::<SyncUuid>().map_err(|error| {
            SqliteSyncError::InvalidMetadata(format!("invalid deck UUID in database: {error}"))
        })?;
        revisions.insert(deck_uuid.as_str().to_string(), revision_number.to_string());
    }
    Ok(revisions)
}

fn reconcile_sqlite_deck_bundle_files(
    deck_dir: &Path,
    current_revision: Option<&str>,
) -> Result<(), SqliteSyncError> {
    let entries = fs::read_dir(deck_dir).map_err(|error| SqliteSyncError::IoPath {
        action: "could not read sync server deck bundle directory",
        path: deck_dir.to_path_buf(),
        source: error,
    })?;
    for entry in entries {
        let entry = entry.map_err(|error| SqliteSyncError::IoPath {
            action: "could not read sync server deck bundle directory entry",
            path: deck_dir.to_path_buf(),
            source: error,
        })?;
        let path = entry.path();
        let metadata = fs::symlink_metadata(&path).map_err(|error| SqliteSyncError::IoPath {
            action: "could not inspect sync server bundle file",
            path: path.clone(),
            source: error,
        })?;
        if metadata.file_type().is_symlink() {
            return Err(sync_storage_invalid(format!(
                "sync server bundle file must not be a symlink: {}",
                path.display()
            )));
        }
        if !metadata.is_file() {
            continue;
        }
        let Some(file_name) = path.file_name().and_then(|name| name.to_str()) else {
            continue;
        };
        if should_remove_sqlite_reconciliation_file(file_name, current_revision, &metadata) {
            fs::remove_file(&path).map_err(|error| SqliteSyncError::IoPath {
                action: "could not remove stale sync server bundle file",
                path,
                source: error,
            })?;
        }
    }
    Ok(())
}

const SQLITE_SYNC_ABANDONED_UPLOAD_FILE_SECONDS: u64 = 60 * 60;

fn should_remove_sqlite_reconciliation_file(
    file_name: &str,
    current_revision: Option<&str>,
    metadata: &fs::Metadata,
) -> bool {
    if is_sqlite_staged_upload_file_name(file_name) {
        return is_stale_sqlite_staged_upload_file_name(file_name);
    }
    is_abandoned_sqlite_revision_bundle_file(file_name, current_revision, metadata)
}

fn is_sqlite_staged_upload_file_name(file_name: &str) -> bool {
    file_name.starts_with(".upload-") && file_name.ends_with(".tmp")
}

fn is_stale_sqlite_staged_upload_file_name(file_name: &str) -> bool {
    let Some(started_at) = sqlite_staged_upload_unix_seconds(file_name) else {
        return true;
    };
    current_unix_seconds()
        .checked_sub(started_at)
        .is_some_and(|age| age >= SQLITE_SYNC_ABANDONED_UPLOAD_FILE_SECONDS as i64)
}

fn sqlite_staged_upload_unix_seconds(file_name: &str) -> Option<i64> {
    let body = file_name.strip_prefix(".upload-")?.strip_suffix(".tmp")?;
    let mut parts = body.splitn(4, '-');
    let _pid = parts.next()?;
    parts.next()?.parse::<i64>().ok()
}

fn is_abandoned_sqlite_revision_bundle_file(
    file_name: &str,
    current_revision: Option<&str>,
    metadata: &fs::Metadata,
) -> bool {
    let Some(revision) = sqlite_revision_from_bundle_file_name(file_name) else {
        return false;
    };
    let Some(current_revision) = current_revision else {
        return sqlite_sync_file_age_seconds(metadata)
            .is_some_and(|age| age >= SQLITE_SYNC_ABANDONED_UPLOAD_FILE_SECONDS);
    };
    if revision == current_revision {
        return false;
    }
    let Ok(revision_number) = revision.parse::<u64>() else {
        return false;
    };
    let Ok(current_revision_number) = current_revision.parse::<u64>() else {
        return false;
    };
    if revision_number < current_revision_number {
        return true;
    }
    sqlite_sync_file_age_seconds(metadata)
        .is_some_and(|age| age >= SQLITE_SYNC_ABANDONED_UPLOAD_FILE_SECONDS)
}

fn sqlite_sync_file_age_seconds(metadata: &fs::Metadata) -> Option<u64> {
    metadata
        .modified()
        .ok()
        .and_then(|modified| SystemTime::now().duration_since(modified).ok())
        .map(|duration| duration.as_secs())
}

fn read_sqlite_sync_bundle_file(
    root: &Path,
    deck_uuid: &SyncUuid,
    file_name: &str,
) -> Result<Vec<u8>, SqliteSyncError> {
    validate_sync_storage_component(file_name, "sync server bundle file")?;
    let path = ensure_sqlite_deck_bundle_dir(root, deck_uuid, false)?.join(file_name);
    ensure_plain_regular_file_if_exists(&path, "could not inspect sync server bundle file")?;
    fs::read(&path).map_err(|error| SqliteSyncError::IoPath {
        action: "could not read sync server bundle file",
        path,
        source: error,
    })
}

fn rename_sqlite_deck_bundle_dir_for_delete(
    root: &Path,
    deck_uuid: &SyncUuid,
) -> Result<Option<PathBuf>, SqliteSyncError> {
    let root_canonical = ensure_sqlite_sync_storage_root(root)?;
    let bundles_dir = root.join(SQLITE_SYNC_BUNDLES_DIR);
    if !bundles_dir.exists() {
        return Ok(None);
    }
    ensure_plain_existing_directory(
        &bundles_dir,
        "could not inspect sync server bundle directory",
    )?;
    canonical_contained_path(&bundles_dir, &root_canonical)?;

    let deck_dir = bundles_dir.join(deck_uuid.as_str());
    if !deck_dir.exists() {
        return Ok(None);
    }
    ensure_plain_existing_directory(&deck_dir, "could not inspect sync server bundle directory")?;
    canonical_contained_path(&deck_dir, &root_canonical)?;

    let tombstone = bundles_dir.join(sqlite_delete_tombstone_name(deck_uuid));
    validate_sync_storage_component(
        tombstone
            .file_name()
            .and_then(|name| name.to_str())
            .unwrap_or_default(),
        "sync server delete tombstone directory",
    )?;
    fs::rename(&deck_dir, &tombstone).map_err(|error| SqliteSyncError::IoPath {
        action: "could not move sync server bundle directory aside for deletion",
        path: deck_dir,
        source: error,
    })?;
    Ok(Some(tombstone))
}

fn restore_sqlite_deck_bundle_dir_from_tombstone(
    root: &Path,
    deck_uuid: &SyncUuid,
    tombstone: Option<PathBuf>,
) -> Result<(), SqliteSyncError> {
    let Some(tombstone) = tombstone else {
        return Ok(());
    };
    let deck_dir = root.join(SQLITE_SYNC_BUNDLES_DIR).join(deck_uuid.as_str());
    if deck_dir.exists() {
        remove_sqlite_delete_tombstone(Some(tombstone))?;
        return Ok(());
    }
    fs::rename(&tombstone, &deck_dir).map_err(|error| SqliteSyncError::IoPath {
        action: "could not restore sync server bundle directory after failed deletion",
        path: tombstone,
        source: error,
    })
}

fn remove_sqlite_delete_tombstone(tombstone: Option<PathBuf>) -> Result<(), SqliteSyncError> {
    let Some(tombstone) = tombstone else {
        return Ok(());
    };
    if !tombstone.exists() {
        return Ok(());
    }
    remove_plain_directory_recursive(&tombstone, "could not remove sync server delete tombstone")
}

fn remove_plain_directory_recursive(
    directory: &Path,
    action: &'static str,
) -> Result<(), SqliteSyncError> {
    ensure_plain_existing_directory(directory, action)?;
    let entries = fs::read_dir(directory).map_err(|error| SqliteSyncError::IoPath {
        action,
        path: directory.to_path_buf(),
        source: error,
    })?;
    for entry in entries {
        let entry = entry.map_err(|error| SqliteSyncError::IoPath {
            action,
            path: directory.to_path_buf(),
            source: error,
        })?;
        let path = entry.path();
        let metadata = fs::symlink_metadata(&path).map_err(|error| SqliteSyncError::IoPath {
            action,
            path: path.clone(),
            source: error,
        })?;
        if metadata.file_type().is_symlink() {
            return Err(sync_storage_invalid(format!(
                "sync server storage path must not be a symlink: {}",
                path.display()
            )));
        }
        if metadata.is_dir() {
            remove_plain_directory_recursive(&path, action)?;
        } else if metadata.is_file() {
            fs::remove_file(&path).map_err(|error| SqliteSyncError::IoPath {
                action,
                path: path.clone(),
                source: error,
            })?;
        } else {
            return Err(sync_storage_invalid(format!(
                "sync server storage path must be a file or directory: {}",
                path.display()
            )));
        }
    }
    fs::remove_dir(directory).map_err(|error| SqliteSyncError::IoPath {
        action,
        path: directory.to_path_buf(),
        source: error,
    })
}

fn reconcile_sqlite_delete_tombstones(
    root: &Path,
    connection: &Connection,
) -> Result<(), SqliteSyncError> {
    let root_canonical = ensure_sqlite_sync_storage_root(root)?;
    let bundles_dir = root.join(SQLITE_SYNC_BUNDLES_DIR);
    if !bundles_dir.exists() {
        return Ok(());
    }
    ensure_plain_existing_directory(
        &bundles_dir,
        "could not inspect sync server bundle directory",
    )?;
    canonical_contained_path(&bundles_dir, &root_canonical)?;
    let entries = fs::read_dir(&bundles_dir).map_err(|error| SqliteSyncError::IoPath {
        action: "could not read sync server bundle directory",
        path: bundles_dir.clone(),
        source: error,
    })?;
    for entry in entries {
        let entry = entry.map_err(|error| SqliteSyncError::IoPath {
            action: "could not read sync server bundle directory entry",
            path: bundles_dir.clone(),
            source: error,
        })?;
        let path = entry.path();
        let Some(file_name) = path.file_name().and_then(|name| name.to_str()) else {
            continue;
        };
        let Some(deck_uuid) = deck_uuid_from_delete_tombstone_name(file_name) else {
            continue;
        };
        ensure_plain_existing_directory(
            &path,
            "could not inspect sync server delete tombstone directory",
        )?;
        canonical_contained_path(&path, &root_canonical)?;
        if read_sqlite_metadata(connection, &deck_uuid)?.is_some() {
            let deck_dir = bundles_dir.join(deck_uuid.as_str());
            if !deck_dir.exists() {
                fs::rename(&path, &deck_dir).map_err(|error| SqliteSyncError::IoPath {
                    action: "could not restore sync server bundle directory from delete tombstone",
                    path: path.clone(),
                    source: error,
                })?;
            } else {
                remove_plain_directory_recursive(
                    &path,
                    "could not remove stale sync server delete tombstone",
                )?;
            }
        } else {
            remove_plain_directory_recursive(
                &path,
                "could not remove committed sync server delete tombstone",
            )?;
        }
    }
    Ok(())
}

fn deck_uuid_from_delete_tombstone_name(name: &str) -> Option<SyncUuid> {
    let rest = name.strip_prefix(SQLITE_SYNC_DELETE_TOMBSTONE_PREFIX)?;
    let uuid = rest.get(..36)?;
    let suffix = rest.get(36..)?;
    if !suffix.starts_with('-') {
        return None;
    }
    SyncUuid::from_str(uuid).ok()
}

fn prune_old_sqlite_sync_bundle_files(
    root: &Path,
    deck_uuid: &SyncUuid,
    current_revision: &str,
) -> Result<(), SqliteSyncError> {
    let deck_dir = ensure_sqlite_deck_bundle_dir(root, deck_uuid, false)?;
    let current_content = sqlite_content_bundle_file_name(current_revision);
    let current_history = sqlite_study_history_bundle_file_name(current_revision);
    let entries = fs::read_dir(&deck_dir).map_err(|error| SqliteSyncError::IoPath {
        action: "could not read sync server bundle directory",
        path: deck_dir.clone(),
        source: error,
    })?;

    for entry in entries {
        let entry = entry.map_err(|error| SqliteSyncError::IoPath {
            action: "could not read sync server bundle directory entry",
            path: deck_dir.clone(),
            source: error,
        })?;
        let path = entry.path();
        let Some(file_name) = path.file_name().and_then(|name| name.to_str()) else {
            continue;
        };
        if !is_sqlite_revision_bundle_file_name(file_name)
            || file_name == current_content
            || file_name == current_history
        {
            continue;
        }
        if ensure_plain_regular_file_if_exists(
            &path,
            "could not inspect old sync server bundle file",
        )? {
            fs::remove_file(&path).map_err(|error| SqliteSyncError::IoPath {
                action: "could not remove old sync server bundle file",
                path: path.clone(),
                source: error,
            })?;
        }
    }
    Ok(())
}

fn is_sqlite_revision_bundle_file_name(file_name: &str) -> bool {
    sqlite_revision_from_bundle_file_name(file_name).is_some()
}

fn sqlite_revision_from_bundle_file_name(file_name: &str) -> Option<&str> {
    let revision = file_name
        .strip_prefix("content-r")
        .or_else(|| file_name.strip_prefix("study-history-r"))?
        .strip_suffix(".bundle")?;
    (!revision.is_empty() && revision.bytes().all(|byte| byte.is_ascii_digit())).then_some(revision)
}

#[cfg(test)]
mod tests {
    use super::*;
    use proptest::prelude::*;
    use std::io::Write;
    #[cfg(unix)]
    use std::os::unix::fs::symlink;
    use std::process;
    use std::time::{SystemTime, UNIX_EPOCH};

    fn uuid(value: &str) -> SyncUuid {
        value.parse().unwrap()
    }

    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-sync-{prefix}-{}-{unique}", process::id()));
        fs::create_dir_all(&path).unwrap();
        path
    }

    fn generated_uuid() -> impl Strategy<Value = SyncUuid> {
        any::<[u8; 16]>().prop_map(SyncUuid::from_random_bytes)
    }

    fn single_line_text() -> impl Strategy<Value = String> {
        "[A-Za-z0-9_.:@/-]{1,40}"
    }

    fn optional_single_line_text() -> impl Strategy<Value = Option<String>> {
        prop_oneof![Just(None), single_line_text().prop_map(Some)]
    }

    fn visibility_strategy() -> impl Strategy<Value = DeckVisibility> {
        prop_oneof![
            Just(DeckVisibility::Private),
            Just(DeckVisibility::Unlisted),
            Just(DeckVisibility::Public),
        ]
    }

    fn device() -> LocalDeviceIdentity {
        LocalDeviceIdentity::new(Some("laptop".to_string()))
    }

    fn test_bundle_payload(bytes: Vec<u8>, includes_review_database: bool) -> SyncBundlePayload {
        let bundle_hash = sync_bundle_hash(&bytes);
        SyncBundlePayload::from_parts(bytes, &bundle_hash, includes_review_database).unwrap()
    }

    fn content_bundle(label: &str) -> SyncBundlePayload {
        test_bundle_payload(
            test_zip_bytes(&[
                (
                    "deck.cosimo-deck",
                    format!("prompt {label}\n---\nresponse {label}\n===").as_bytes(),
                ),
                (
                    "deck/1.prompt.wav",
                    format!("prompt audio {label}").as_bytes(),
                ),
            ]),
            false,
        )
    }

    fn upload_request_with_payloads(
        deck_uuid: SyncUuid,
        owner_identity: &str,
        visibility: DeckVisibility,
        expected_revision: Option<&str>,
        content_bundle: SyncBundlePayload,
        study_history_bundle: Option<SyncBundlePayload>,
    ) -> SyncUploadRequest {
        SyncUploadRequest::from_parts(
            deck_uuid,
            owner_identity,
            visibility,
            expected_revision,
            device(),
            Some(content_bundle),
            study_history_bundle,
        )
        .unwrap()
    }

    fn study_history_bundle(label: &str) -> SyncBundlePayload {
        test_bundle_payload(
            test_zip_bytes(&[
                (
                    "deck.cosimo-deck",
                    format!("prompt {label}\n---\nresponse {label}\n===").as_bytes(),
                ),
                (
                    "deck.reviews.sqlite3",
                    format!("review database {label}").as_bytes(),
                ),
                (
                    "deck/1.prompt.wav",
                    format!("prompt audio {label}").as_bytes(),
                ),
            ]),
            true,
        )
    }

    fn test_zip_bytes(entries: &[(&str, &[u8])]) -> Vec<u8> {
        let mut zip = ZipWriter::new(Cursor::new(Vec::new()));
        for (name, bytes) in entries {
            zip.start_file(*name, sync_bundle_file_options()).unwrap();
            zip.write_all(bytes).unwrap();
        }
        zip.finish().unwrap().into_inner()
    }

    fn upload_request(
        deck_uuid: SyncUuid,
        visibility: DeckVisibility,
        expected_revision: Option<&str>,
    ) -> SyncUploadRequest {
        SyncUploadRequest::from_parts(
            deck_uuid,
            "acct:modulus@example.invalid",
            visibility,
            expected_revision,
            device(),
            Some(content_bundle("one")),
            Some(study_history_bundle("one")),
        )
        .unwrap()
    }

    fn other_owner_upload_request(deck_uuid: SyncUuid) -> SyncUploadRequest {
        upload_request_with_payloads(
            deck_uuid,
            "acct:other@example.invalid",
            DeckVisibility::Public,
            None,
            content_bundle("other"),
            Some(study_history_bundle("other")),
        )
    }

    fn sqlite_deck_identity_owner(root: &Path, deck_uuid: &SyncUuid) -> Option<Option<String>> {
        let connection = Connection::open(sqlite_sync_database_path(root).unwrap()).unwrap();
        let mut statement = connection
            .prepare(
                "
                SELECT owner_identity
                FROM deck_identity_owners
                WHERE deck_uuid = ?1
                ",
            )
            .unwrap();
        let mut rows = statement.query(params![deck_uuid.as_str()]).unwrap();
        rows.next().unwrap().map(|row| row.get(0).unwrap())
    }

    #[derive(Clone, Debug)]
    enum SqliteStorageOperation {
        UploadContent,
        UploadStudyHistory,
        ReopenAndReconcile,
        DeleteDeck,
        InjectFreshFutureBundle,
        InjectFreshStagedUpload,
        InjectMalformedStagedUpload,
        InjectOldRevisionBundle,
    }

    fn arbitrary_sqlite_storage_operation() -> impl Strategy<Value = SqliteStorageOperation> {
        prop_oneof![
            5 => Just(SqliteStorageOperation::UploadContent),
            5 => Just(SqliteStorageOperation::UploadStudyHistory),
            4 => Just(SqliteStorageOperation::ReopenAndReconcile),
            2 => Just(SqliteStorageOperation::DeleteDeck),
            2 => Just(SqliteStorageOperation::InjectFreshFutureBundle),
            2 => Just(SqliteStorageOperation::InjectFreshStagedUpload),
            2 => Just(SqliteStorageOperation::InjectMalformedStagedUpload),
            2 => Just(SqliteStorageOperation::InjectOldRevisionBundle),
        ]
    }

    fn run_sqlite_storage_operation_sequence(
        operations: &[SqliteStorageOperation],
    ) -> Result<(), TestCaseError> {
        let temp_dir = unique_temp_dir("sqlite-storage-property");
        let deck_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        let owner = "acct:modulus@example.invalid";
        let mut server = SqliteSyncServer::new(&temp_dir);
        let mut revision: Option<String> = None;
        let mut upload_count = 0_u64;

        for operation in operations {
            match operation {
                SqliteStorageOperation::UploadContent => {
                    upload_count += 1;
                    let content = content_bundle(&format!("content-{upload_count}"));
                    let outcome = server
                        .upload(upload_request_with_payloads(
                            deck_uuid.clone(),
                            owner,
                            DeckVisibility::Private,
                            revision.as_deref(),
                            content,
                            None,
                        ))
                        .map_err(|error| {
                            TestCaseError::fail(format!(
                                "content upload failed after {operation:?}: {error}"
                            ))
                        })?;
                    revision = Some(outcome.revision().to_string());
                }
                SqliteStorageOperation::UploadStudyHistory => {
                    upload_count += 1;
                    let label = format!("history-{upload_count}");
                    let outcome = server
                        .upload(upload_request_with_payloads(
                            deck_uuid.clone(),
                            owner,
                            DeckVisibility::Private,
                            revision.as_deref(),
                            content_bundle(&label),
                            Some(study_history_bundle(&label)),
                        ))
                        .map_err(|error| {
                            TestCaseError::fail(format!(
                                "study-history upload failed after {operation:?}: {error}"
                            ))
                        })?;
                    revision = Some(outcome.revision().to_string());
                }
                SqliteStorageOperation::ReopenAndReconcile => {
                    server = SqliteSyncServer::new(&temp_dir);
                    server.list_public_decks().map_err(|error| {
                        TestCaseError::fail(format!("reopen reconciliation failed: {error}"))
                    })?;
                }
                SqliteStorageOperation::DeleteDeck => match server.delete(&deck_uuid, owner) {
                    Ok(()) => revision = None,
                    Err(SqliteSyncError::Delete(SyncDeleteError::NotFound)) => {}
                    Err(error) => {
                        return Err(TestCaseError::fail(format!(
                            "delete failed unexpectedly: {error}"
                        )));
                    }
                },
                SqliteStorageOperation::InjectFreshFutureBundle => {
                    let future_revision = revision
                        .as_deref()
                        .and_then(|value| value.parse::<u64>().ok())
                        .unwrap_or(0)
                        + 100;
                    let bundle_dir = sqlite_deck_bundle_dir(&temp_dir, &deck_uuid);
                    fs::create_dir_all(&bundle_dir).map_err(|error| {
                        TestCaseError::fail(format!("could not create bundle dir: {error}"))
                    })?;
                    fs::write(
                        bundle_dir.join(sqlite_content_bundle_file_name(
                            &future_revision.to_string(),
                        )),
                        b"fresh future bundle",
                    )
                    .map_err(|error| {
                        TestCaseError::fail(format!("could not inject future bundle: {error}"))
                    })?;
                }
                SqliteStorageOperation::InjectFreshStagedUpload => {
                    let bundle_dir = sqlite_deck_bundle_dir(&temp_dir, &deck_uuid);
                    fs::create_dir_all(&bundle_dir).map_err(|error| {
                        TestCaseError::fail(format!("could not create bundle dir: {error}"))
                    })?;
                    let staged_file_name = format!(
                        ".upload-{}-{}-content-0.tmp",
                        process::id(),
                        current_unix_seconds()
                    );
                    fs::write(bundle_dir.join(staged_file_name), b"fresh staged upload").map_err(
                        |error| {
                            TestCaseError::fail(format!(
                                "could not inject fresh staged upload: {error}"
                            ))
                        },
                    )?;
                }
                SqliteStorageOperation::InjectMalformedStagedUpload => {
                    let bundle_dir = sqlite_deck_bundle_dir(&temp_dir, &deck_uuid);
                    fs::create_dir_all(&bundle_dir).map_err(|error| {
                        TestCaseError::fail(format!("could not create bundle dir: {error}"))
                    })?;
                    fs::write(
                        bundle_dir.join(".upload-not-from-current-format.tmp"),
                        b"malformed staged upload",
                    )
                    .map_err(|error| {
                        TestCaseError::fail(format!(
                            "could not inject malformed staged upload: {error}"
                        ))
                    })?;
                }
                SqliteStorageOperation::InjectOldRevisionBundle => {
                    let Some(current_revision) = revision
                        .as_deref()
                        .and_then(|value| value.parse::<u64>().ok())
                    else {
                        assert_sqlite_storage_invariants(
                            &temp_dir,
                            &deck_uuid,
                            revision.as_deref(),
                        )?;
                        continue;
                    };
                    let old_revision = current_revision.saturating_sub(1);
                    let bundle_dir = sqlite_deck_bundle_dir(&temp_dir, &deck_uuid);
                    fs::create_dir_all(&bundle_dir).map_err(|error| {
                        TestCaseError::fail(format!("could not create bundle dir: {error}"))
                    })?;
                    fs::write(
                        bundle_dir.join(sqlite_content_bundle_file_name(&old_revision.to_string())),
                        b"old revision bundle",
                    )
                    .map_err(|error| {
                        TestCaseError::fail(format!("could not inject old bundle: {error}"))
                    })?;
                }
            }
            assert_sqlite_storage_invariants(&temp_dir, &deck_uuid, revision.as_deref())?;
        }

        Ok(())
    }

    fn assert_sqlite_storage_invariants(
        root: &Path,
        expected_deck_uuid: &SyncUuid,
        expected_revision: Option<&str>,
    ) -> Result<(), TestCaseError> {
        let db_path = root.join(SQLITE_SYNC_DATABASE_FILE);
        if !db_path.exists() && expected_revision.is_none() {
            return Ok(());
        }
        if !db_path.exists() {
            return Err(TestCaseError::fail(format!(
                "upload model expects deck {} revision {}, but the sync database is missing",
                expected_deck_uuid.as_str(),
                expected_revision.unwrap_or("none")
            )));
        }
        let connection = Connection::open(&db_path).map_err(|error| {
            TestCaseError::fail(format!(
                "could not open sync database for invariant check: {error}"
            ))
        })?;
        let mut expected_statement = connection
            .prepare(
                "
                SELECT revision
                FROM decks
                WHERE deck_uuid = ?1
                ",
            )
            .map_err(|error| {
                TestCaseError::fail(format!("could not prepare expected deck query: {error}"))
            })?;
        let mut expected_rows = expected_statement
            .query(params![expected_deck_uuid.as_str()])
            .map_err(|error| {
                TestCaseError::fail(format!("could not run expected deck query: {error}"))
            })?;
        let actual_revision = match expected_rows.next().map_err(|error| {
            TestCaseError::fail(format!("could not read expected deck query row: {error}"))
        })? {
            Some(row) => Some(row.get::<_, i64>(0).map_err(|error| {
                TestCaseError::fail(format!("could not read expected deck revision: {error}"))
            })?),
            None => None,
        };
        match (expected_revision, actual_revision) {
            (Some(expected), Some(actual)) if expected == actual.to_string() => {}
            (Some(expected), Some(actual)) => {
                return Err(TestCaseError::fail(format!(
                    "upload model expects deck {} revision {}, but SQLite has revision {}",
                    expected_deck_uuid.as_str(),
                    expected,
                    actual
                )));
            }
            (Some(expected), None) => {
                return Err(TestCaseError::fail(format!(
                    "upload model expects deck {} revision {}, but SQLite has no deck row",
                    expected_deck_uuid.as_str(),
                    expected
                )));
            }
            (None, Some(actual)) => {
                return Err(TestCaseError::fail(format!(
                    "upload model expects no deck {}, but SQLite has revision {}",
                    expected_deck_uuid.as_str(),
                    actual
                )));
            }
            (None, None) => {}
        }
        let mut statement = connection
            .prepare(
                "
                SELECT deck_uuid, revision, study_history_bundle_hash
                FROM decks
                ",
            )
            .map_err(|error| {
                TestCaseError::fail(format!("could not prepare invariant query: {error}"))
            })?;
        let mut rows = statement.query([]).map_err(|error| {
            TestCaseError::fail(format!("could not run invariant query: {error}"))
        })?;
        while let Some(row) = rows.next().map_err(|error| {
            TestCaseError::fail(format!("could not read invariant row: {error}"))
        })? {
            let deck_uuid_text: String = row.get(0).map_err(|error| {
                TestCaseError::fail(format!("could not read invariant deck UUID: {error}"))
            })?;
            let revision_number: i64 = row.get(1).map_err(|error| {
                TestCaseError::fail(format!("could not read invariant revision: {error}"))
            })?;
            let history_hash: Option<String> = row.get(2).map_err(|error| {
                TestCaseError::fail(format!(
                    "could not read invariant study-history hash: {error}"
                ))
            })?;
            let deck_uuid = deck_uuid_text.parse::<SyncUuid>().map_err(|error| {
                TestCaseError::fail(format!("invalid invariant deck UUID: {error}"))
            })?;
            let revision = revision_number.to_string();
            let required_file = if history_hash.is_some() {
                sqlite_study_history_bundle_file_name(&revision)
            } else {
                sqlite_content_bundle_file_name(&revision)
            };
            let required_path = sqlite_deck_bundle_dir(root, &deck_uuid).join(&required_file);
            if !required_path.exists() {
                return Err(TestCaseError::fail(format!(
                    "database row for deck {} revision {} references missing bundle file {}",
                    deck_uuid.as_str(),
                    revision,
                    required_file
                )));
            }
            ensure_plain_regular_file_if_exists(
                &required_path,
                "could not inspect invariant bundle file",
            )
            .map_err(|error| {
                TestCaseError::fail(format!("invalid invariant bundle file: {error}"))
            })?;
        }
        Ok(())
    }

    #[test]
    fn parses_and_normalizes_canonical_uuid() {
        let uuid: SyncUuid = "550E8400-E29B-41D4-A716-446655440000".parse().unwrap();

        assert_eq!(uuid.as_str(), "550e8400-e29b-41d4-a716-446655440000");
        assert_eq!(uuid.to_string(), "550e8400-e29b-41d4-a716-446655440000");
    }

    #[test]
    fn rejects_invalid_uuid_strings() {
        assert_eq!("".parse::<SyncUuid>(), Err(SyncUuidParseError::WrongLength));
        assert_eq!(
            "550e8400e29b41d4a716446655440000".parse::<SyncUuid>(),
            Err(SyncUuidParseError::WrongLength)
        );
        assert_eq!(
            "550e8400_e29b-41d4-a716-446655440000".parse::<SyncUuid>(),
            Err(SyncUuidParseError::MissingHyphen)
        );
        assert_eq!(
            "550e8400-e29b-41d4-a716-44665544000z".parse::<SyncUuid>(),
            Err(SyncUuidParseError::InvalidHexDigit)
        );
        assert_eq!(
            "00000000-0000-0000-0000-000000000000".parse::<SyncUuid>(),
            Err(SyncUuidParseError::Nil)
        );
    }

    #[test]
    fn rejects_path_like_uuid_strings() {
        for value in [
            "..",
            "../550e8400-e29b-41d4-a716-446655440000",
            "550e8400-e29b-41d4-a716-446655440000/../upload",
            "%2e%2e%2f550e8400-e29b-41d4-a716-446655440000",
        ] {
            assert!(
                value.parse::<SyncUuid>().is_err(),
                "path-like UUID should be rejected: {value}"
            );
        }
    }

    #[test]
    fn generated_uuid_bytes_have_v4_shape() {
        let uuid = SyncUuid::from_random_bytes([0xff; 16]);

        assert_eq!(uuid.as_str(), "ffffffff-ffff-4fff-bfff-ffffffffffff");
    }

    #[test]
    fn deck_visibility_codes_round_trip() {
        for visibility in DeckVisibility::ALL {
            assert_eq!(
                DeckVisibility::from_code(visibility.code()),
                Some(*visibility)
            );
        }
        assert_eq!(DeckVisibility::from_code("friends"), None);
    }

    #[test]
    fn normalizes_optional_device_name() {
        assert_eq!(
            normalize_device_name(Some(" laptop ")),
            Some("laptop".to_string())
        );
        assert_eq!(normalize_device_name(Some(" ")), None);
        assert_eq!(normalize_device_name(Some("desk\ntop")), None);
        assert_eq!(normalize_device_name(None), None);
    }

    #[test]
    fn sync_auth_session_validates_bearer_token_shape() {
        assert!(
            SyncAuthSession::from_parts("acct:modulus@example.invalid", "cosimo_token").is_ok()
        );
        assert!(matches!(
            SyncAuthSession::from_parts("acct:modulus@example.invalid", "two words"),
            Err(SyncAuthError::WhitespaceBearerToken)
        ));
    }

    #[test]
    fn in_memory_sync_auth_session_maps_bearer_token_to_owner() {
        let mut server = InMemorySyncServer::new();
        let session = server
            .create_auth_session_for_verified_owner("acct:modulus@example.invalid")
            .unwrap();

        assert_eq!(
            server.authenticate_bearer_token(session.bearer_token()),
            Some("acct:modulus@example.invalid".to_string())
        );
        assert_eq!(server.authenticate_bearer_token("missing"), None);
    }

    #[test]
    fn in_memory_sync_auth_sessions_can_be_revoked() {
        let mut server = InMemorySyncServer::new();
        let session = server
            .create_auth_session_for_verified_owner("acct:modulus@example.invalid")
            .unwrap();

        assert!(server.revoke_bearer_token(session.bearer_token()));
        assert!(!server.revoke_bearer_token(session.bearer_token()));
        assert_eq!(
            server.authenticate_bearer_token(session.bearer_token()),
            None
        );
    }

    #[test]
    fn in_memory_sync_can_revoke_all_sessions_for_one_owner() {
        let mut server = InMemorySyncServer::new();
        let first = server
            .create_auth_session_for_verified_owner("acct:modulus@example.invalid")
            .unwrap();
        let second = server
            .create_auth_session_for_verified_owner("acct:modulus@example.invalid")
            .unwrap();
        let other = server
            .create_auth_session_for_verified_owner("acct:other@example.invalid")
            .unwrap();

        assert_eq!(
            server.revoke_owner_bearer_tokens("acct:modulus@example.invalid"),
            2
        );
        assert_eq!(server.authenticate_bearer_token(first.bearer_token()), None);
        assert_eq!(
            server.authenticate_bearer_token(second.bearer_token()),
            None
        );
        assert_eq!(
            server.authenticate_bearer_token(other.bearer_token()),
            Some("acct:other@example.invalid".to_string())
        );
    }

    #[test]
    fn sqlite_sync_auth_sessions_persist_across_server_instances() {
        let temp_dir = unique_temp_dir("sqlite-auth-session");
        let server = SqliteSyncServer::new(&temp_dir);
        let session = server
            .create_auth_session_for_verified_owner("acct:modulus@example.invalid")
            .unwrap();
        drop(server);

        let reopened = SqliteSyncServer::new(&temp_dir);
        assert_eq!(
            reopened
                .authenticate_bearer_token(session.bearer_token())
                .unwrap(),
            Some("acct:modulus@example.invalid".to_string())
        );
    }

    #[test]
    fn sqlite_sync_auth_sessions_store_token_verifiers_not_plaintext_tokens() {
        let temp_dir = unique_temp_dir("sqlite-auth-session-verifier");
        let server = SqliteSyncServer::new(&temp_dir);
        let session = server
            .create_auth_session_for_verified_owner("acct:modulus@example.invalid")
            .unwrap();
        let connection = Connection::open(sqlite_sync_database_path(&temp_dir).unwrap()).unwrap();
        let stored_token: String = connection
            .query_row("SELECT bearer_token FROM auth_sessions", [], |row| {
                row.get(0)
            })
            .unwrap();

        assert_ne!(stored_token, session.bearer_token());
        assert_eq!(
            stored_token,
            sync_bearer_token_verifier(session.bearer_token())
        );
        assert_eq!(
            server
                .authenticate_bearer_token(session.bearer_token())
                .unwrap(),
            Some("acct:modulus@example.invalid".to_string())
        );
    }

    #[test]
    fn sqlite_sync_connections_enable_secure_delete() {
        let temp_dir = unique_temp_dir("sqlite-secure-delete");
        let server = SqliteSyncServer::new(&temp_dir);
        let connection = server.open_connection().unwrap();
        let secure_delete: i64 = connection
            .query_row("PRAGMA secure_delete", [], |row| row.get(0))
            .unwrap();

        assert_eq!(secure_delete, 1);
    }

    #[test]
    fn sqlite_sync_auth_session_opening_migrates_legacy_plaintext_tokens() {
        let temp_dir = unique_temp_dir("sqlite-auth-session-legacy-token");
        let server = SqliteSyncServer::new(&temp_dir);
        let connection = server.open_connection().unwrap();
        connection
            .execute(
                "
                INSERT INTO auth_sessions (
                    bearer_token,
                    owner_identity,
                    created_at_unix_seconds
                )
                VALUES (?1, ?2, ?3)
                ",
                params!["cosimo_legacy_token", "acct:modulus@example.invalid", 0],
            )
            .unwrap();
        drop(connection);

        let reopened = SqliteSyncServer::new(&temp_dir);
        assert_eq!(
            reopened
                .authenticate_bearer_token("cosimo_legacy_token")
                .unwrap(),
            Some("acct:modulus@example.invalid".to_string())
        );
        let connection = Connection::open(sqlite_sync_database_path(&temp_dir).unwrap()).unwrap();
        let stored_token: String = connection
            .query_row("SELECT bearer_token FROM auth_sessions", [], |row| {
                row.get(0)
            })
            .unwrap();
        assert_eq!(
            stored_token,
            sync_bearer_token_verifier("cosimo_legacy_token")
        );
    }

    #[test]
    fn sqlite_sync_committed_upload_verification_detects_missing_bundle_file() {
        let temp_dir = unique_temp_dir("sqlite-committed-upload-missing-bundle");
        let server = SqliteSyncServer::new(&temp_dir);
        let deck_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        let owner = "acct:modulus@example.invalid";
        let content = content_bundle("verified upload");
        let content_hash = content.bundle_hash().to_string();
        let outcome = server
            .upload(upload_request_with_payloads(
                deck_uuid.clone(),
                owner,
                DeckVisibility::Private,
                None,
                content,
                None,
            ))
            .unwrap();
        let bundle_file = sqlite_content_bundle_file_name(outcome.revision());
        fs::remove_file(sqlite_deck_bundle_dir(&temp_dir, &deck_uuid).join(&bundle_file)).unwrap();

        let result = server.verify_committed_upload(ExpectedCommittedUpload {
            deck_uuid: &deck_uuid,
            owner_identity: owner,
            visibility_code: DeckVisibility::Private.code(),
            revision: outcome.revision(),
            content_bundle_hash: &content_hash,
            study_history_bundle_hash: None,
            required_bundle_file_name: &bundle_file,
        });

        assert!(matches!(
            result,
            Err(SqliteSyncError::InvalidMetadata(message))
                if message.contains("missing bundle file")
        ));
    }

    #[test]
    fn sqlite_sync_committed_upload_verification_detects_missing_database_row() {
        let temp_dir = unique_temp_dir("sqlite-committed-upload-missing-row");
        let server = SqliteSyncServer::new(&temp_dir);
        let deck_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        let owner = "acct:modulus@example.invalid";
        let content = content_bundle("verified upload");
        let content_hash = content.bundle_hash().to_string();
        let outcome = server
            .upload(upload_request_with_payloads(
                deck_uuid.clone(),
                owner,
                DeckVisibility::Private,
                None,
                content,
                None,
            ))
            .unwrap();
        let connection =
            rusqlite::Connection::open(sqlite_sync_database_path(&temp_dir).unwrap()).unwrap();
        connection
            .execute(
                "DELETE FROM decks WHERE deck_uuid = ?1",
                params![deck_uuid.as_str()],
            )
            .unwrap();

        let result = server.verify_committed_upload(ExpectedCommittedUpload {
            deck_uuid: &deck_uuid,
            owner_identity: owner,
            visibility_code: DeckVisibility::Private.code(),
            revision: outcome.revision(),
            content_bundle_hash: &content_hash,
            study_history_bundle_hash: None,
            required_bundle_file_name: &sqlite_content_bundle_file_name(outcome.revision()),
        });

        assert!(matches!(
            result,
            Err(SqliteSyncError::InvalidMetadata(message))
                if message.contains("missing deck metadata")
        ));
    }

    #[test]
    fn sqlite_sync_reconciliation_preserves_fresh_staged_upload_files() {
        let temp_dir = unique_temp_dir("sqlite-reconcile-fresh-staged-upload");
        let server = SqliteSyncServer::new(&temp_dir);
        server.list_public_decks().unwrap();
        let deck_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        let bundle_dir = sqlite_deck_bundle_dir(&temp_dir, &deck_uuid);
        fs::create_dir_all(&bundle_dir).unwrap();
        let staged_file_name = format!(
            ".upload-{}-{}-content-0.tmp",
            std::process::id(),
            current_unix_seconds()
        );
        let staged_path = bundle_dir.join(staged_file_name);
        fs::write(&staged_path, b"active upload").unwrap();
        let connection =
            rusqlite::Connection::open(sqlite_sync_database_path(&temp_dir).unwrap()).unwrap();

        reconcile_sqlite_bundle_files(&temp_dir, &connection).unwrap();

        assert!(staged_path.exists());
    }

    #[test]
    fn sqlite_sync_reconciliation_preserves_fresh_future_revision_bundle_files() {
        let temp_dir = unique_temp_dir("sqlite-reconcile-fresh-future-revision");
        let server = SqliteSyncServer::new(&temp_dir);
        let deck_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        server
            .upload(upload_request(
                deck_uuid.clone(),
                DeckVisibility::Private,
                None,
            ))
            .unwrap();
        let future_path = sqlite_deck_bundle_dir(&temp_dir, &deck_uuid)
            .join(sqlite_content_bundle_file_name("2"));
        fs::write(&future_path, b"fresh future revision").unwrap();

        let reopened = SqliteSyncServer::new(&temp_dir);
        reopened.list_public_decks().unwrap();

        assert!(future_path.exists());
    }

    #[test]
    fn sqlite_sync_reconciliation_removes_old_revision_bundles_below_current_revision() {
        let temp_dir = unique_temp_dir("sqlite-reconcile-old-revision");
        let server = SqliteSyncServer::new(&temp_dir);
        let deck_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        server
            .upload(upload_request(
                deck_uuid.clone(),
                DeckVisibility::Private,
                None,
            ))
            .unwrap();
        server
            .upload(upload_request(
                deck_uuid.clone(),
                DeckVisibility::Private,
                Some("1"),
            ))
            .unwrap();
        let old_path = sqlite_deck_bundle_dir(&temp_dir, &deck_uuid)
            .join(sqlite_content_bundle_file_name("1"));
        fs::write(&old_path, b"old revision").unwrap();

        let reopened = SqliteSyncServer::new(&temp_dir);
        reopened.list_public_decks().unwrap();

        assert!(!old_path.exists());
    }

    #[test]
    fn sqlite_staged_upload_name_staleness_uses_embedded_timestamp() {
        let fresh = format!(
            ".upload-{}-{}-content-0.tmp",
            std::process::id(),
            current_unix_seconds()
        );

        assert!(!is_stale_sqlite_staged_upload_file_name(&fresh));
        assert!(is_stale_sqlite_staged_upload_file_name(
            ".upload-1-0-content-0.tmp"
        ));
        assert!(is_stale_sqlite_staged_upload_file_name(
            ".upload-not-from-cosimo.tmp"
        ));
    }

    #[cfg(unix)]
    #[test]
    fn sqlite_sync_rejects_symlinked_storage_root_without_writing_through_it() {
        let temp_dir = unique_temp_dir("sqlite-root-symlink");
        let outside_dir = temp_dir.join("outside");
        fs::create_dir_all(&outside_dir).unwrap();
        let storage_root = temp_dir.join("storage");
        symlink(&outside_dir, &storage_root).unwrap();
        let server = SqliteSyncServer::new(&storage_root);

        let error = server
            .create_auth_session_for_verified_owner("acct:modulus@example.invalid")
            .unwrap_err();

        assert!(matches!(
            error,
            SqliteSyncError::InvalidMetadata(message)
                if message.contains("must not be a symlink")
        ));
        assert!(
            !outside_dir.join(SQLITE_SYNC_DATABASE_FILE).exists(),
            "the server must not create its database through a symlinked storage root"
        );
    }

    #[cfg(unix)]
    #[test]
    fn sqlite_sync_rejects_symlinked_database_file() {
        let temp_dir = unique_temp_dir("sqlite-db-symlink");
        let storage_root = temp_dir.join("storage");
        let outside_file = temp_dir.join("outside.sqlite3");
        fs::create_dir_all(&storage_root).unwrap();
        fs::write(&outside_file, b"not a sync database").unwrap();
        symlink(&outside_file, storage_root.join(SQLITE_SYNC_DATABASE_FILE)).unwrap();
        let server = SqliteSyncServer::new(&storage_root);

        let error = server
            .create_auth_session_for_verified_owner("acct:modulus@example.invalid")
            .unwrap_err();

        assert!(matches!(
            error,
            SqliteSyncError::InvalidMetadata(message)
                if message.contains("storage file must not be a symlink")
        ));
        assert_eq!(fs::read(&outside_file).unwrap(), b"not a sync database");
    }

    #[cfg(unix)]
    #[test]
    fn sqlite_sync_rejects_symlinked_bundle_directory_without_writing_outside() {
        let temp_dir = unique_temp_dir("sqlite-bundles-symlink");
        let storage_root = temp_dir.join("storage");
        let outside_dir = temp_dir.join("outside");
        fs::create_dir_all(&storage_root).unwrap();
        fs::create_dir_all(&outside_dir).unwrap();
        symlink(&outside_dir, storage_root.join(SQLITE_SYNC_BUNDLES_DIR)).unwrap();
        let deck_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        let server = SqliteSyncServer::new(&storage_root);

        let error = server
            .upload(upload_request(
                deck_uuid.clone(),
                DeckVisibility::Public,
                None,
            ))
            .unwrap_err();

        assert!(matches!(
            error,
            SqliteSyncError::InvalidMetadata(message)
                if message.contains("must not be a symlink")
        ));
        assert!(
            !outside_dir.join(deck_uuid.as_str()).exists(),
            "the server must not create deck bundle directories through a symlink"
        );
    }

    #[test]
    fn sqlite_sync_rejects_bundle_with_unexpected_file() {
        let temp_dir = unique_temp_dir("sqlite-unexpected-bundle-entry");
        let server = SqliteSyncServer::new(&temp_dir);
        let deck_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        let content = test_bundle_payload(
            test_zip_bytes(&[
                ("deck.cosimo-deck", b"prompt\n---\nresponse\n==="),
                ("deck/unexpected.exe", b"not deck data"),
            ]),
            false,
        );

        let result = server.upload(upload_request_with_payloads(
            deck_uuid,
            "acct:modulus@example.invalid",
            DeckVisibility::Private,
            None,
            content,
            None,
        ));

        assert!(matches!(
            result,
            Err(SqliteSyncError::Upload(SyncUploadError::InvalidBundle(_)))
        ));
    }

    #[test]
    fn sqlite_sync_accepts_supported_audio_bundle_entries() {
        let temp_dir = unique_temp_dir("sqlite-audio-bundle-entry");
        let server = SqliteSyncServer::new(&temp_dir);
        let deck_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        let content = test_bundle_payload(
            test_zip_bytes(&[
                ("deck.cosimo-deck", b"prompt\n---\nresponse\n==="),
                ("deck/1.prompt.mp3", b"mp3 audio"),
                ("deck/1.prompt.opus", b"opus audio"),
                ("deck/audio-manifest.tsv", b"version\t1\n"),
            ]),
            false,
        );

        let result = server.upload(upload_request_with_payloads(
            deck_uuid,
            "acct:modulus@example.invalid",
            DeckVisibility::Private,
            None,
            content,
            None,
        ));

        assert!(result.is_ok());
    }

    #[test]
    fn sqlite_sync_stores_deck_summary_from_uploaded_bundle() {
        let temp_dir = unique_temp_dir("sqlite-deck-summary");
        let server = SqliteSyncServer::new(&temp_dir);
        let deck_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        let content = test_bundle_payload(
            test_zip_bytes(&[
                (
                    "deck.cosimo-deck",
                    b"deck_version=2\ntitle=Indonesian\n:::\n\nair\n---\nwater\n===\nclose\n---\nmenutup\n===",
                ),
                ("deck/1.prompt.mp3", b"mp3 audio"),
                ("deck/1.prompt.opus", b"opus audio"),
                ("deck/2.response.wav", b"wav audio"),
                (
                    "deck/audio-manifest.tsv",
                    b"version\t2\t\t\n1.prompt.opus\tsha256:prompt\tonecore\tAndika\n",
                ),
            ]),
            false,
        );

        server
            .upload(upload_request_with_payloads(
                deck_uuid.clone(),
                "acct:modulus@example.invalid",
                DeckVisibility::Public,
                None,
                content,
                None,
            ))
            .unwrap();

        let public_decks = server.list_public_decks().unwrap();
        let summary = public_decks[0].summary();
        assert_eq!(public_decks[0].deck_uuid(), &deck_uuid);
        assert_eq!(summary.title(), Some("Indonesian"));
        assert_eq!(summary.deck_name(), Some("deck"));
        assert_eq!(summary.card_count(), Some(2));
        assert_eq!(summary.audio_file_count(), Some(3));
        assert_eq!(summary.managed_audio_file_count(), Some(1));
        assert_eq!(summary.manual_audio_file_count(), Some(2));
        assert_eq!(
            summary.audio_formats(),
            ["mp3".to_string(), "opus".to_string(), "wav".to_string()]
        );
    }

    #[test]
    fn sqlite_sync_rejects_content_bundle_with_review_database() {
        let temp_dir = unique_temp_dir("sqlite-content-with-review-db");
        let server = SqliteSyncServer::new(&temp_dir);
        let deck_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        let content = test_bundle_payload(
            test_zip_bytes(&[
                ("deck.cosimo-deck", b"prompt\n---\nresponse\n==="),
                ("deck.reviews.sqlite3", b"review database"),
            ]),
            false,
        );

        let result = server.upload(upload_request_with_payloads(
            deck_uuid,
            "acct:modulus@example.invalid",
            DeckVisibility::Private,
            None,
            content,
            None,
        ));

        assert!(matches!(
            result,
            Err(SqliteSyncError::Upload(SyncUploadError::InvalidBundle(_)))
        ));
    }

    #[test]
    fn sqlite_sync_enforces_max_decks_per_owner() {
        let temp_dir = unique_temp_dir("sqlite-max-decks");
        let server = SqliteSyncServer::new(&temp_dir);
        let limits = SyncServerLimits {
            max_decks_per_owner: Some(1),
            ..SyncServerLimits::default()
        };
        server
            .upload_with_limits(
                upload_request(
                    uuid("550e8400-e29b-41d4-a716-446655440000"),
                    DeckVisibility::Private,
                    None,
                ),
                limits,
            )
            .unwrap();

        let result = server.upload_with_limits(
            upload_request(
                uuid("650e8400-e29b-41d4-a716-446655440000"),
                DeckVisibility::Private,
                None,
            ),
            limits,
        );

        assert!(matches!(
            result,
            Err(SqliteSyncError::Upload(SyncUploadError::QuotaExceeded(_)))
        ));
    }

    #[test]
    fn sqlite_sync_enforces_owner_storage_quota() {
        let temp_dir = unique_temp_dir("sqlite-owner-storage-quota");
        let server = SqliteSyncServer::new(&temp_dir);
        let deck_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        let request = upload_request(deck_uuid.clone(), DeckVisibility::Private, None);
        let limit = stored_sync_upload_bytes(&request).saturating_sub(1);
        let limits = SyncServerLimits {
            max_storage_bytes_per_owner: Some(limit),
            ..SyncServerLimits::default()
        };

        let result = server.upload_with_limits(request, limits);

        assert!(matches!(
            result,
            Err(SqliteSyncError::Upload(SyncUploadError::QuotaExceeded(_)))
        ));
        let bundle_dir = sqlite_deck_bundle_dir(&temp_dir, &deck_uuid);
        if bundle_dir.exists() {
            for entry in fs::read_dir(bundle_dir).unwrap() {
                let file_name = entry.unwrap().file_name();
                let file_name = file_name.to_string_lossy();
                assert!(!file_name.starts_with(".upload-"));
                assert!(!is_sqlite_revision_bundle_file_name(&file_name));
            }
        }
    }

    #[test]
    fn sqlite_sync_reports_owner_quota_status() {
        let temp_dir = unique_temp_dir("sqlite-owner-quota-status");
        let server = SqliteSyncServer::new(&temp_dir);
        let limits = SyncServerLimits {
            max_decks_per_owner: Some(3),
            max_storage_bytes_per_owner: Some(10_000),
            max_storage_bytes_total: Some(100_000),
            ..SyncServerLimits::default()
        };
        let first = upload_request(
            uuid("550e8400-e29b-41d4-a716-446655440000"),
            DeckVisibility::Private,
            None,
        );
        let first_bytes = stored_sync_upload_bytes(&first);
        server.upload(first).unwrap();
        let second = upload_request_with_payloads(
            uuid("650e8400-e29b-41d4-a716-446655440000"),
            "acct:other@example.invalid",
            DeckVisibility::Private,
            None,
            content_bundle("two"),
            None,
        );
        server.upload(second).unwrap();

        let status = server
            .quota_status(limits, Some("acct:modulus@example.invalid"))
            .unwrap();

        assert_eq!(status.max_decks_per_owner(), Some(3));
        assert_eq!(status.max_storage_bytes_per_owner(), Some(10_000));
        assert_eq!(status.max_storage_bytes_total(), Some(100_000));
        assert_eq!(status.owner_deck_count(), Some(1));
        assert_eq!(status.owner_storage_bytes(), Some(first_bytes));
    }

    #[test]
    fn memory_sync_reports_owner_quota_status() {
        let mut server = InMemorySyncServer::new();
        let limits = SyncServerLimits {
            max_decks_per_owner: Some(2),
            max_storage_bytes_per_owner: Some(4096),
            ..SyncServerLimits::default()
        };
        let request = upload_request(
            uuid("550e8400-e29b-41d4-a716-446655440000"),
            DeckVisibility::Private,
            None,
        );
        let stored_bytes = stored_sync_upload_bytes(&request);
        server.upload(request).unwrap();

        let status = server.quota_status(limits, Some("acct:modulus@example.invalid"));

        assert_eq!(status.max_decks_per_owner(), Some(2));
        assert_eq!(status.max_storage_bytes_per_owner(), Some(4096));
        assert_eq!(status.owner_deck_count(), Some(1));
        assert_eq!(status.owner_storage_bytes(), Some(stored_bytes));
    }

    #[test]
    fn sqlite_sync_enforces_deck_file_size_limit() {
        let temp_dir = unique_temp_dir("sqlite-deck-file-size-limit");
        let server = SqliteSyncServer::new(&temp_dir);
        let deck_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        let content = test_bundle_payload(
            test_zip_bytes(&[("deck.cosimo-deck", b"prompt\n---\nresponse\n===")]),
            false,
        );
        let limits = SyncServerLimits {
            max_deck_file_bytes: Some(8),
            ..SyncServerLimits::default()
        };

        let result = server.upload_with_limits(
            upload_request_with_payloads(
                deck_uuid,
                "acct:modulus@example.invalid",
                DeckVisibility::Private,
                None,
                content,
                None,
            ),
            limits,
        );

        assert!(matches!(
            result,
            Err(SqliteSyncError::Upload(SyncUploadError::QuotaExceeded(_)))
        ));
    }

    #[test]
    fn sqlite_sync_auth_sessions_can_be_revoked() {
        let temp_dir = unique_temp_dir("sqlite-auth-session-revocation");
        let server = SqliteSyncServer::new(&temp_dir);
        let session = server
            .create_auth_session_for_verified_owner("acct:modulus@example.invalid")
            .unwrap();

        assert!(server.revoke_bearer_token(session.bearer_token()).unwrap());
        assert!(!server.revoke_bearer_token(session.bearer_token()).unwrap());
        assert_eq!(
            server
                .authenticate_bearer_token(session.bearer_token())
                .unwrap(),
            None
        );
    }

    #[test]
    fn sqlite_sync_can_revoke_all_sessions_for_one_owner() {
        let temp_dir = unique_temp_dir("sqlite-auth-session-owner-revocation");
        let server = SqliteSyncServer::new(&temp_dir);
        let first = server
            .create_auth_session_for_verified_owner("acct:modulus@example.invalid")
            .unwrap();
        let second = server
            .create_auth_session_for_verified_owner("acct:modulus@example.invalid")
            .unwrap();
        let other = server
            .create_auth_session_for_verified_owner("acct:other@example.invalid")
            .unwrap();

        assert_eq!(
            server
                .revoke_owner_bearer_tokens("acct:modulus@example.invalid")
                .unwrap(),
            2
        );
        assert_eq!(
            server
                .authenticate_bearer_token(first.bearer_token())
                .unwrap(),
            None
        );
        assert_eq!(
            server
                .authenticate_bearer_token(second.bearer_token())
                .unwrap(),
            None
        );
        assert_eq!(
            server
                .authenticate_bearer_token(other.bearer_token())
                .unwrap(),
            Some("acct:other@example.invalid".to_string())
        );
    }

    #[test]
    fn in_memory_oauth_pending_login_is_single_use() {
        let mut server = InMemorySyncServer::new();
        let pending = server.create_oauth_browser_pending_login().unwrap();
        server
            .configure_oauth_pending_login_for_browser_authorization(
                pending.login_token(),
                "https://example.invalid",
                "client id",
                "client secret",
                "https://sync.example.invalid/auth/fedi/callback",
                None,
            )
            .unwrap()
            .unwrap();

        let taken = server
            .take_oauth_pending_login(pending.login_token())
            .unwrap();

        assert_eq!(taken.instance_url(), "https://example.invalid");
        assert_eq!(taken.client_id(), "client id");
        assert_eq!(taken.client_secret(), "client secret");
        assert!(
            server
                .take_oauth_pending_login(pending.login_token())
                .is_none()
        );
    }

    #[test]
    fn sqlite_oauth_pending_login_persists_and_is_single_use() {
        let temp_dir = unique_temp_dir("sqlite-oauth-pending-login");
        let server = SqliteSyncServer::new(&temp_dir);
        let pending = server.create_oauth_browser_pending_login().unwrap();
        server
            .configure_oauth_pending_login_for_browser(
                pending.login_token(),
                "https://example.invalid",
                "client id",
                "client secret",
                "https://sync.example.invalid/auth/fedi/callback",
                None,
                "notice-v1",
                "sha256:notice",
                "consent",
            )
            .unwrap()
            .unwrap();
        drop(server);

        let reopened = SqliteSyncServer::new(&temp_dir);
        let taken = reopened
            .take_oauth_pending_login(pending.login_token())
            .unwrap()
            .unwrap();

        assert_eq!(taken.instance_url(), "https://example.invalid");
        assert_eq!(taken.client_id(), "client id");
        assert_eq!(taken.client_secret(), "client secret");
        assert_eq!(taken.accepted_notice_version(), Some("notice-v1"));
        assert_eq!(taken.accepted_notice_hash(), Some("sha256:notice"));
        assert_eq!(taken.accepted_lawful_basis(), Some("consent"));
        assert!(taken.accepted_at_unix_seconds().is_some());
        reopened
            .record_privacy_acceptance_for_verified_owner("acct:modulus@example.invalid", &taken)
            .unwrap();
        let connection = Connection::open(sqlite_sync_database_path(&temp_dir).unwrap()).unwrap();
        let acceptance_count: i64 = connection
            .query_row(
                "
                SELECT COUNT(*)
                FROM account_privacy_acceptances
                WHERE owner_identity = ?1
                  AND notice_version = ?2
                  AND notice_hash = ?3
                  AND lawful_basis = ?4
                ",
                params![
                    "acct:modulus@example.invalid",
                    "notice-v1",
                    "sha256:notice",
                    "consent"
                ],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(acceptance_count, 1);
        assert!(
            reopened
                .take_oauth_pending_login(pending.login_token())
                .unwrap()
                .is_none()
        );
    }

    #[test]
    fn sqlite_browser_oauth_login_handoff_is_single_use() {
        let temp_dir = unique_temp_dir("sqlite-browser-oauth-login");
        let server = SqliteSyncServer::new(&temp_dir);
        let pending = server.create_oauth_browser_pending_login().unwrap();
        server
            .configure_oauth_pending_login_for_browser(
                pending.login_token(),
                "https://example.invalid",
                "client id",
                "client secret",
                "https://sync.example.invalid/sync.cgi/auth/fedi/callback",
                Some("acct:modulus@example.invalid"),
                "notice-v1",
                "sha256:notice",
                "consent",
            )
            .unwrap()
            .unwrap();
        drop(server);

        let reopened = SqliteSyncServer::new(&temp_dir);
        let taken = reopened
            .take_oauth_pending_login(pending.login_token())
            .unwrap()
            .unwrap();
        assert_eq!(taken.instance_url(), "https://example.invalid");
        assert_eq!(
            taken.redirect_uri(),
            "https://sync.example.invalid/sync.cgi/auth/fedi/callback"
        );
        assert_eq!(
            taken.expected_owner_identity(),
            Some("acct:modulus@example.invalid")
        );
        assert!(taken.has_privacy_acceptance());
        reopened
            .store_completed_oauth_login(taken.login_token(), "acct:modulus@example.invalid")
            .unwrap();

        assert_eq!(
            reopened
                .take_completed_oauth_login(taken.login_token())
                .unwrap()
                .as_deref(),
            Some("acct:modulus@example.invalid")
        );
        assert!(
            reopened
                .take_completed_oauth_login(taken.login_token())
                .unwrap()
                .is_none()
        );
    }

    #[test]
    fn sqlite_oauth_start_attempts_are_limited_by_client_and_instance() {
        let temp_dir = unique_temp_dir("sqlite-oauth-start-attempt-limits");
        let server = SqliteSyncServer::new(&temp_dir);
        let client_limited = SyncServerLimits {
            max_oauth_starts_per_client_per_hour: Some(1),
            ..SyncServerLimits::default()
        };

        server
            .record_oauth_start_attempt_with_limits(
                "https://one.invalid",
                "203.0.113.10",
                client_limited,
            )
            .unwrap();
        assert!(matches!(
            server.record_oauth_start_attempt_with_limits(
                "https://two.invalid",
                "203.0.113.10",
                client_limited,
            ),
            Err(SqliteSyncError::Auth(SyncAuthError::RateLimited(_)))
        ));

        let instance_limited = SyncServerLimits {
            max_oauth_starts_per_client_per_hour: None,
            max_oauth_starts_per_instance_per_hour: Some(1),
            ..SyncServerLimits::default()
        };
        let other_server = SqliteSyncServer::new(unique_temp_dir("sqlite-oauth-instance-limit"));
        other_server
            .record_oauth_start_attempt_with_limits(
                "https://one.invalid",
                "203.0.113.10",
                instance_limited,
            )
            .unwrap();
        assert!(matches!(
            other_server.record_oauth_start_attempt_with_limits(
                "https://one.invalid",
                "203.0.113.11",
                instance_limited,
            ),
            Err(SqliteSyncError::Auth(SyncAuthError::RateLimited(_)))
        ));
    }

    #[test]
    fn sqlite_account_action_pending_logins_are_limited_per_instance() {
        let temp_dir = unique_temp_dir("sqlite-oauth-pending-limit");
        let server = SqliteSyncServer::new(&temp_dir);
        let limits = SyncServerLimits {
            max_pending_oauth_logins_per_instance: Some(1),
            ..SyncServerLimits::default()
        };

        server
            .create_oauth_account_action_pending_login_with_limits(
                "https://example.invalid",
                "client id",
                "client secret",
                "https://sync.example.invalid/auth/fedi/delete/callback",
                "acct:modulus@example.invalid",
                limits,
            )
            .unwrap();

        assert!(matches!(
            server.create_oauth_account_action_pending_login_with_limits(
                "https://example.invalid",
                "client id",
                "client secret",
                "https://sync.example.invalid/auth/fedi/delete/callback",
                "acct:modulus@example.invalid",
                limits,
            ),
            Err(SqliteSyncError::Auth(SyncAuthError::RateLimited(_)))
        ));
        assert!(
            server
                .create_oauth_account_action_pending_login_with_limits(
                    "https://other.invalid",
                    "client id",
                    "client secret",
                    "https://sync.example.invalid/auth/fedi/delete/callback",
                    "acct:modulus@other.invalid",
                    limits,
                )
                .is_ok()
        );
    }

    #[test]
    fn local_deck_sync_state_normalizes_fields() {
        let deck_uuid: SyncUuid = "550e8400-e29b-41d4-a716-446655440000".parse().unwrap();
        let state = LocalDeckSyncState::from_parts(
            deck_uuid.clone(),
            " https://sync.example.invalid ",
            Some(" acct:modulus@example.invalid "),
            Some(" 42 "),
            Some(" sha256:local "),
            Some(" sha256:remote "),
        )
        .unwrap();

        assert_eq!(state.deck_uuid(), &deck_uuid);
        assert_eq!(state.sync_server_url(), "https://sync.example.invalid");
        assert_eq!(state.owner_identity(), Some("acct:modulus@example.invalid"));
        assert_eq!(state.last_seen_remote_revision(), Some("42"));
        assert_eq!(
            state.last_uploaded_local_bundle_hash(),
            Some("sha256:local")
        );
        assert_eq!(
            state.last_downloaded_remote_bundle_hash(),
            Some("sha256:remote")
        );
    }

    #[test]
    fn local_deck_sync_state_rejects_empty_or_multiline_required_fields() {
        let deck_uuid: SyncUuid = "550e8400-e29b-41d4-a716-446655440000".parse().unwrap();

        assert_eq!(
            LocalDeckSyncState::from_parts(deck_uuid.clone(), " ", None, None, None, None),
            Err(LocalDeckSyncStateError::EmptyValue {
                field: "sync_server_url"
            })
        );
        assert_eq!(
            LocalDeckSyncState::from_parts(deck_uuid, "https://x\ninvalid", None, None, None, None),
            Err(LocalDeckSyncStateError::MultilineValue {
                field: "sync_server_url"
            })
        );
    }

    #[test]
    fn local_deck_sync_state_compares_remote_slots_without_revisions() {
        let deck_uuid: SyncUuid = "550e8400-e29b-41d4-a716-446655440000".parse().unwrap();
        let first = LocalDeckSyncState::from_parts(
            deck_uuid.clone(),
            "https://sync.example.invalid",
            Some("owner"),
            Some("1"),
            None,
            None,
        )
        .unwrap();
        let second = LocalDeckSyncState::from_parts(
            deck_uuid,
            "https://sync.example.invalid",
            Some("owner"),
            Some("2"),
            Some("sha256:local"),
            Some("sha256:remote"),
        )
        .unwrap();

        assert!(first.same_remote_slot(&second));
    }

    #[test]
    fn sync_upload_request_rejects_review_database_in_content_bundle() {
        let deck_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        let request = SyncUploadRequest::from_parts(
            deck_uuid,
            "owner",
            DeckVisibility::Private,
            None,
            device(),
            Some(study_history_bundle("not-content")),
            None,
        );

        assert_eq!(
            request,
            Err(SyncValidationError::ContentBundleContainsReviewDatabase)
        );
    }

    #[test]
    fn sync_upload_request_accepts_history_only_and_rejects_empty_payloads() {
        let deck_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");

        assert!(
            SyncUploadRequest::from_parts(
                deck_uuid.clone(),
                "owner",
                DeckVisibility::Private,
                None,
                device(),
                None,
                Some(study_history_bundle("one")),
            )
            .is_ok()
        );
        assert_eq!(
            SyncUploadRequest::from_parts(
                deck_uuid,
                "owner",
                DeckVisibility::Private,
                None,
                device(),
                None,
                None,
            ),
            Err(SyncValidationError::MissingUploadBundle)
        );
    }

    #[test]
    fn in_memory_sync_upload_and_owner_download_round_trip_study_history() {
        let deck_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        let mut server = InMemorySyncServer::new();

        let upload = server
            .upload(upload_request(
                deck_uuid.clone(),
                DeckVisibility::Private,
                None,
            ))
            .unwrap();
        let download = server
            .download(
                &deck_uuid,
                SyncDownloadKind::OwnerStudyHistory,
                Some("acct:modulus@example.invalid"),
            )
            .unwrap();

        assert_eq!(upload.revision(), "1");
        assert!(upload.bundle_hash().starts_with("sha256:"));
        assert_eq!(download.revision(), "1");
        assert!(download.bundle().includes_review_database());
        assert_eq!(download.bundle().bundle_hash(), upload.bundle_hash());
        assert_eq!(download.last_upload_device_name(), Some("laptop"));
    }

    #[test]
    fn shared_download_returns_content_only_bundle_for_public_deck() {
        let deck_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        let mut server = InMemorySyncServer::new();
        server
            .upload(upload_request(
                deck_uuid.clone(),
                DeckVisibility::Public,
                None,
            ))
            .unwrap();

        let download = server
            .download(&deck_uuid, SyncDownloadKind::SharedContent, None)
            .unwrap();

        assert_eq!(download.visibility(), DeckVisibility::Public);
        assert!(!download.bundle().includes_review_database());
        assert!(download.bundle().bundle_hash().starts_with("sha256:"));
    }

    #[test]
    fn shared_download_rejects_private_deck() {
        let deck_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        let mut server = InMemorySyncServer::new();
        server
            .upload(upload_request(
                deck_uuid.clone(),
                DeckVisibility::Private,
                None,
            ))
            .unwrap();

        assert_eq!(
            server.download(&deck_uuid, SyncDownloadKind::SharedContent, None),
            Err(SyncDownloadError::Forbidden)
        );
    }

    #[test]
    fn shared_deck_inspection_exposes_only_non_private_decks() {
        let public_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        let unlisted_uuid = uuid("650e8400-e29b-41d4-a716-446655440000");
        let private_uuid = uuid("750e8400-e29b-41d4-a716-446655440000");
        let mut server = InMemorySyncServer::new();
        server
            .upload(upload_request(
                public_uuid.clone(),
                DeckVisibility::Public,
                None,
            ))
            .unwrap();
        server
            .upload(upload_request_with_payloads(
                unlisted_uuid.clone(),
                "acct:modulus@example.invalid",
                DeckVisibility::Unlisted,
                None,
                content_bundle("unlisted"),
                None,
            ))
            .unwrap();
        server
            .upload(upload_request_with_payloads(
                private_uuid.clone(),
                "acct:modulus@example.invalid",
                DeckVisibility::Private,
                None,
                content_bundle("private"),
                None,
            ))
            .unwrap();

        let public = server.inspect_shared_deck(&public_uuid).unwrap();
        let unlisted = server.inspect_shared_deck(&unlisted_uuid).unwrap();

        assert_eq!(public.deck_uuid(), &public_uuid);
        assert_eq!(public.visibility(), DeckVisibility::Public);
        assert!(public.content_bundle_hash().starts_with("sha256:"));
        assert_eq!(unlisted.deck_uuid(), &unlisted_uuid);
        assert_eq!(unlisted.visibility(), DeckVisibility::Unlisted);
        assert_eq!(
            server.inspect_shared_deck(&private_uuid),
            Err(SyncDownloadError::Forbidden)
        );
    }

    #[test]
    fn public_deck_listing_excludes_unlisted_private_and_device_metadata() {
        let public_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        let unlisted_uuid = uuid("650e8400-e29b-41d4-a716-446655440000");
        let private_uuid = uuid("750e8400-e29b-41d4-a716-446655440000");
        let mut server = InMemorySyncServer::new();
        server
            .upload(upload_request(
                public_uuid.clone(),
                DeckVisibility::Public,
                None,
            ))
            .unwrap();
        server
            .upload(upload_request_with_payloads(
                unlisted_uuid,
                "acct:modulus@example.invalid",
                DeckVisibility::Unlisted,
                None,
                content_bundle("unlisted"),
                Some(study_history_bundle("unlisted")),
            ))
            .unwrap();
        server
            .upload(upload_request_with_payloads(
                private_uuid,
                "acct:other@example.invalid",
                DeckVisibility::Private,
                None,
                content_bundle("private"),
                Some(study_history_bundle("private")),
            ))
            .unwrap();

        let public_decks = server.list_public_decks();

        assert_eq!(public_decks.len(), 1);
        assert_eq!(public_decks[0].deck_uuid(), &public_uuid);
        assert_eq!(public_decks[0].visibility(), DeckVisibility::Public);
        assert_eq!(
            public_decks[0].owner_identity(),
            "acct:modulus@example.invalid"
        );
        assert_eq!(public_decks[0].revision(), "1");
    }

    #[test]
    fn owner_deck_listing_includes_private_and_last_upload_device() {
        let public_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        let private_uuid = uuid("650e8400-e29b-41d4-a716-446655440000");
        let other_uuid = uuid("750e8400-e29b-41d4-a716-446655440000");
        let mut server = InMemorySyncServer::new();
        server
            .upload(upload_request(
                public_uuid.clone(),
                DeckVisibility::Public,
                None,
            ))
            .unwrap();
        server
            .upload(upload_request_with_payloads(
                private_uuid.clone(),
                "acct:modulus@example.invalid",
                DeckVisibility::Private,
                None,
                content_bundle("private"),
                Some(study_history_bundle("private")),
            ))
            .unwrap();
        server
            .upload(upload_request_with_payloads(
                other_uuid,
                "acct:other@example.invalid",
                DeckVisibility::Public,
                None,
                content_bundle("other"),
                None,
            ))
            .unwrap();

        let owner_decks = server.list_owner_decks("acct:modulus@example.invalid");
        let private_status = server
            .inspect_owner_deck(&private_uuid, "acct:modulus@example.invalid")
            .unwrap();

        assert_eq!(owner_decks.len(), 2);
        assert!(owner_decks.iter().any(|status| {
            status.deck_uuid() == &public_uuid
                && status.last_upload_device_name() == Some("laptop")
                && status.has_study_history()
        }));
        assert!(owner_decks.iter().any(|status| {
            status.deck_uuid() == &private_uuid
                && status.visibility() == DeckVisibility::Private
                && status
                    .study_history_bundle_hash()
                    .is_some_and(|hash| hash.starts_with("sha256:"))
        }));
        assert_eq!(private_status.last_upload_device_name(), Some("laptop"));
        assert_eq!(
            server.inspect_owner_deck(&private_uuid, "acct:other@example.invalid"),
            Err(SyncDownloadError::Forbidden)
        );
    }

    #[test]
    fn sync_upload_rejects_blind_overwrite_and_accepts_expected_revision() {
        let deck_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        let mut server = InMemorySyncServer::new();
        let first = server
            .upload(upload_request(
                deck_uuid.clone(),
                DeckVisibility::Private,
                None,
            ))
            .unwrap();

        assert_eq!(
            server.upload(upload_request(
                deck_uuid.clone(),
                DeckVisibility::Private,
                None
            )),
            Err(SyncUploadError::Conflict {
                expected: None,
                actual: Some("1".to_string())
            })
        );

        let second = server
            .upload(upload_request(
                deck_uuid,
                DeckVisibility::Unlisted,
                Some(first.revision()),
            ))
            .unwrap();

        assert_eq!(second.revision(), "2");
    }

    #[test]
    fn sync_upload_rejects_owner_mismatch() {
        let deck_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        let mut server = InMemorySyncServer::new();
        server
            .upload(upload_request(
                deck_uuid.clone(),
                DeckVisibility::Private,
                None,
            ))
            .unwrap();
        let request = SyncUploadRequest::from_parts(
            deck_uuid,
            "acct:other@example.invalid",
            DeckVisibility::Private,
            Some("1"),
            device(),
            Some(content_bundle("two")),
            None,
        )
        .unwrap();

        assert_eq!(server.upload(request), Err(SyncUploadError::Forbidden));
    }

    #[test]
    fn in_memory_sync_delete_requires_owner_and_removes_deck() {
        let deck_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        let mut server = InMemorySyncServer::new();
        server
            .upload(upload_request(
                deck_uuid.clone(),
                DeckVisibility::Public,
                None,
            ))
            .unwrap();

        assert_eq!(
            server.delete(&deck_uuid, "acct:other@example.invalid"),
            Err(SyncDeleteError::Forbidden)
        );
        assert_eq!(
            server
                .list_owner_decks("acct:modulus@example.invalid")
                .len(),
            1
        );

        server
            .delete(&deck_uuid, "acct:modulus@example.invalid")
            .unwrap();

        assert_eq!(
            server.download(
                &deck_uuid,
                SyncDownloadKind::OwnerStudyHistory,
                Some("acct:modulus@example.invalid")
            ),
            Err(SyncDownloadError::NotFound)
        );
        assert!(server.list_public_decks().is_empty());
    }

    #[test]
    fn in_memory_sync_account_deletion_removes_owner_decks_and_sessions() {
        let deck_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        let mut server = InMemorySyncServer::new();
        let session = server
            .create_auth_session_for_verified_owner("acct:modulus@example.invalid")
            .unwrap();
        server
            .upload(upload_request(
                deck_uuid.clone(),
                DeckVisibility::Public,
                None,
            ))
            .unwrap();

        let outcome = server
            .delete_owner_account("acct:modulus@example.invalid")
            .unwrap();

        assert_eq!(outcome.decks_deleted(), 1);
        assert_eq!(outcome.sessions_revoked(), 1);
        assert_eq!(
            server.authenticate_bearer_token(session.bearer_token()),
            None
        );
        assert!(server.list_public_decks().is_empty());
        assert!(
            server
                .list_owner_decks("acct:modulus@example.invalid")
                .is_empty()
        );
    }

    #[test]
    fn sqlite_sync_restores_delete_tombstone_when_metadata_remains() {
        let temp_dir = unique_temp_dir("sqlite-delete-tombstone-restore");
        let deck_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        let server = SqliteSyncServer::new(&temp_dir);
        server
            .upload(upload_request(
                deck_uuid.clone(),
                DeckVisibility::Public,
                None,
            ))
            .unwrap();
        let bundle_dir = sqlite_deck_bundle_dir(&temp_dir, &deck_uuid);
        let tombstone = bundle_dir
            .parent()
            .unwrap()
            .join(format!(".delete-{}-test-1", deck_uuid.as_str()));
        fs::rename(&bundle_dir, &tombstone).unwrap();

        let reopened = SqliteSyncServer::new(&temp_dir);
        reopened
            .download(&deck_uuid, SyncDownloadKind::SharedContent, None)
            .unwrap();

        assert!(bundle_dir.exists());
        assert!(!tombstone.exists());
    }

    #[test]
    fn sqlite_sync_reconciliation_removes_malformed_staged_uploads_and_preserves_fresh_future_revisions()
     {
        let temp_dir = unique_temp_dir("sqlite-reconcile-stale-bundles");
        let deck_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        let server = SqliteSyncServer::new(&temp_dir);
        server
            .upload(upload_request(
                deck_uuid.clone(),
                DeckVisibility::Private,
                None,
            ))
            .unwrap();
        let bundle_dir = sqlite_deck_bundle_dir(&temp_dir, &deck_uuid);
        let current = bundle_dir.join("study-history-r1.bundle");
        let staged = bundle_dir.join(".upload-test-study-history-0.tmp");
        let fresh_future_content = bundle_dir.join("content-r2.bundle");
        let fresh_future_history = bundle_dir.join("study-history-r3.bundle");
        let unknown = bundle_dir.join("operator-note.txt");
        fs::write(&staged, b"staged").unwrap();
        fs::write(&fresh_future_content, b"fresh future content").unwrap();
        fs::write(&fresh_future_history, b"fresh future history").unwrap();
        fs::write(&unknown, b"preserve").unwrap();

        let reopened = SqliteSyncServer::new(&temp_dir);
        reopened
            .download(
                &deck_uuid,
                SyncDownloadKind::OwnerStudyHistory,
                Some("acct:modulus@example.invalid"),
            )
            .unwrap();

        assert!(current.exists());
        assert!(unknown.exists());
        assert!(!staged.exists());
        assert!(fresh_future_content.exists());
        assert!(fresh_future_history.exists());
    }

    #[cfg(unix)]
    #[test]
    fn sqlite_sync_reconciliation_rejects_symlinked_bundle_files() {
        use std::os::unix::fs::symlink;

        let temp_dir = unique_temp_dir("sqlite-reconcile-symlink");
        let deck_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        let server = SqliteSyncServer::new(&temp_dir);
        server
            .upload(upload_request(
                deck_uuid.clone(),
                DeckVisibility::Private,
                None,
            ))
            .unwrap();
        let bundle_dir = sqlite_deck_bundle_dir(&temp_dir, &deck_uuid);
        let outside = temp_dir.join("outside");
        fs::write(&outside, b"outside").unwrap();
        symlink(
            &outside,
            bundle_dir.join(".upload-test-study-history-0.tmp"),
        )
        .unwrap();

        let reopened = SqliteSyncServer::new(&temp_dir);
        let error = reopened
            .download(
                &deck_uuid,
                SyncDownloadKind::OwnerStudyHistory,
                Some("acct:modulus@example.invalid"),
            )
            .unwrap_err();

        assert!(matches!(
            error,
            SqliteSyncError::InvalidMetadata(message)
                if message.contains("must not be a symlink")
        ));
        assert!(outside.exists());
    }

    #[test]
    fn sqlite_sync_round_trips_after_reopening_backend() {
        let temp_dir = unique_temp_dir("sqlite-round-trip");
        let deck_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        let server = SqliteSyncServer::new(&temp_dir);
        let upload = server
            .upload(upload_request(
                deck_uuid.clone(),
                DeckVisibility::Public,
                None,
            ))
            .unwrap();
        drop(server);

        let reopened = SqliteSyncServer::new(&temp_dir);
        let owner_download = reopened
            .download(
                &deck_uuid,
                SyncDownloadKind::OwnerStudyHistory,
                Some("acct:modulus@example.invalid"),
            )
            .unwrap();
        let shared_download = reopened
            .download(&deck_uuid, SyncDownloadKind::SharedContent, None)
            .unwrap();

        assert_eq!(upload.revision(), "1");
        assert_eq!(owner_download.revision(), "1");
        assert!(owner_download.bundle().includes_review_database());
        assert!(owner_download.bundle().bundle_hash().starts_with("sha256:"));
        assert!(!shared_download.bundle().includes_review_database());
        assert!(
            shared_download
                .bundle()
                .bundle_hash()
                .starts_with("sha256:")
        );
        assert_ne!(
            shared_download.bundle().bundle_hash(),
            owner_download.bundle().bundle_hash()
        );
        assert_eq!(owner_download.last_upload_device_name(), Some("laptop"));
    }

    #[test]
    fn sqlite_sync_stores_only_study_history_bundle_for_owner_history_uploads() {
        let temp_dir = unique_temp_dir("sqlite-single-history");
        let deck_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        let server = SqliteSyncServer::new(&temp_dir);

        server
            .upload(upload_request(
                deck_uuid.clone(),
                DeckVisibility::Public,
                None,
            ))
            .unwrap();

        let bundle_dir = sqlite_deck_bundle_dir(&temp_dir, &deck_uuid);
        assert!(!bundle_dir.join("content-r1.bundle").exists());
        assert!(bundle_dir.join("study-history-r1.bundle").exists());

        let shared_download = server
            .download(&deck_uuid, SyncDownloadKind::SharedContent, None)
            .unwrap();
        assert!(!shared_download.bundle().includes_review_database());
        assert_ne!(
            shared_download.bundle().bytes(),
            study_history_bundle("one").bytes()
        );
    }

    #[test]
    fn sqlite_sync_prunes_old_bundle_revision_files_after_successful_upload() {
        let temp_dir = unique_temp_dir("sqlite-prune-old-revisions");
        let deck_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        let server = SqliteSyncServer::new(&temp_dir);

        let first = server
            .upload(upload_request(
                deck_uuid.clone(),
                DeckVisibility::Private,
                None,
            ))
            .unwrap();
        let second = server
            .upload(upload_request(
                deck_uuid.clone(),
                DeckVisibility::Private,
                Some(first.revision()),
            ))
            .unwrap();

        let bundle_dir = sqlite_deck_bundle_dir(&temp_dir, &deck_uuid);
        assert_eq!(second.revision(), "2");
        assert!(!bundle_dir.join("study-history-r1.bundle").exists());
        assert!(bundle_dir.join("study-history-r2.bundle").exists());
    }

    #[test]
    fn sqlite_revision_bundle_file_name_recognizes_only_server_revision_bundles() {
        assert!(is_sqlite_revision_bundle_file_name("content-r1.bundle"));
        assert!(is_sqlite_revision_bundle_file_name(
            "study-history-r123.bundle"
        ));
        assert!(!is_sqlite_revision_bundle_file_name("content-r.bundle"));
        assert!(!is_sqlite_revision_bundle_file_name(
            "study-history-rx.bundle"
        ));
        assert!(!is_sqlite_revision_bundle_file_name(
            ".study-history-r1.bundle.tmp"
        ));
        assert!(!is_sqlite_revision_bundle_file_name("notes.txt"));
    }

    #[test]
    fn sqlite_sync_persists_revision_conflicts() {
        let temp_dir = unique_temp_dir("sqlite-conflict");
        let deck_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        let server = SqliteSyncServer::new(&temp_dir);
        let first = server
            .upload(upload_request(
                deck_uuid.clone(),
                DeckVisibility::Private,
                None,
            ))
            .unwrap();
        drop(server);

        let reopened = SqliteSyncServer::new(&temp_dir);
        assert!(matches!(
            reopened.upload(upload_request(
                deck_uuid.clone(),
                DeckVisibility::Private,
                None,
            )),
            Err(SqliteSyncError::Upload(SyncUploadError::Conflict {
                expected: None,
                actual: Some(_)
            }))
        ));
        let second = reopened
            .upload(upload_request(
                deck_uuid,
                DeckVisibility::Unlisted,
                Some(first.revision()),
            ))
            .unwrap();

        assert_eq!(second.revision(), "2");
    }

    #[test]
    fn sqlite_sync_delete_requires_owner_and_removes_bundle_directory() {
        let temp_dir = unique_temp_dir("sqlite-delete");
        let deck_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        let server = SqliteSyncServer::new(&temp_dir);
        server
            .upload(upload_request(
                deck_uuid.clone(),
                DeckVisibility::Public,
                None,
            ))
            .unwrap();
        let bundle_dir = sqlite_deck_bundle_dir(&temp_dir, &deck_uuid);
        fs::write(bundle_dir.join("content-r999.bundle"), b"leftover").unwrap();

        assert!(matches!(
            server.delete(&deck_uuid, "acct:other@example.invalid"),
            Err(SqliteSyncError::Delete(SyncDeleteError::Forbidden))
        ));
        assert!(bundle_dir.exists());

        server
            .delete(&deck_uuid, "acct:modulus@example.invalid")
            .unwrap();

        assert!(!bundle_dir.exists());
        assert!(server.list_public_decks().unwrap().is_empty());
        assert!(matches!(
            server.download(
                &deck_uuid,
                SyncDownloadKind::OwnerStudyHistory,
                Some("acct:modulus@example.invalid")
            ),
            Err(SqliteSyncError::Download(SyncDownloadError::NotFound))
        ));
    }

    #[test]
    fn sqlite_sync_account_deletion_removes_owner_decks_sessions_and_bundles() {
        let temp_dir = unique_temp_dir("sqlite-account-delete");
        let deck_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        let server = SqliteSyncServer::new(&temp_dir);
        let session = server
            .create_auth_session_for_verified_owner("acct:modulus@example.invalid")
            .unwrap();
        server
            .upload(upload_request(
                deck_uuid.clone(),
                DeckVisibility::Public,
                None,
            ))
            .unwrap();
        let bundle_dir = sqlite_deck_bundle_dir(&temp_dir, &deck_uuid);
        assert!(bundle_dir.exists());

        let outcome = server
            .delete_owner_account("acct:modulus@example.invalid")
            .unwrap();

        assert_eq!(outcome.decks_deleted(), 1);
        assert_eq!(outcome.sessions_revoked(), 1);
        assert_eq!(
            server
                .authenticate_bearer_token(session.bearer_token())
                .unwrap(),
            None
        );
        assert!(!bundle_dir.exists());
        assert!(server.list_public_decks().unwrap().is_empty());
        assert!(
            server
                .list_owner_decks("acct:modulus@example.invalid")
                .unwrap()
                .is_empty()
        );
    }

    #[test]
    fn sqlite_sync_lists_public_and_owner_metadata_with_access_boundaries() {
        let temp_dir = unique_temp_dir("sqlite-listing");
        let public_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        let unlisted_uuid = uuid("650e8400-e29b-41d4-a716-446655440000");
        let private_uuid = uuid("750e8400-e29b-41d4-a716-446655440000");
        let server = SqliteSyncServer::new(&temp_dir);
        server
            .upload(upload_request(
                public_uuid.clone(),
                DeckVisibility::Public,
                None,
            ))
            .unwrap();
        server
            .upload(upload_request_with_payloads(
                unlisted_uuid.clone(),
                "acct:modulus@example.invalid",
                DeckVisibility::Unlisted,
                None,
                content_bundle("unlisted"),
                Some(study_history_bundle("unlisted")),
            ))
            .unwrap();
        server
            .upload(upload_request_with_payloads(
                private_uuid.clone(),
                "acct:other@example.invalid",
                DeckVisibility::Private,
                None,
                content_bundle("private"),
                Some(study_history_bundle("private")),
            ))
            .unwrap();

        let public_decks = server.list_public_decks().unwrap();
        let owner_decks = server
            .list_owner_decks("acct:modulus@example.invalid")
            .unwrap();
        let unlisted = server.inspect_shared_deck(&unlisted_uuid).unwrap();

        assert_eq!(public_decks.len(), 1);
        assert_eq!(public_decks[0].deck_uuid(), &public_uuid);
        assert_eq!(public_decks[0].visibility(), DeckVisibility::Public);
        assert_eq!(owner_decks.len(), 2);
        assert!(owner_decks.iter().any(|status| {
            status.deck_uuid() == &unlisted_uuid
                && status
                    .study_history_bundle_hash()
                    .is_some_and(|hash| hash.starts_with("sha256:"))
                && status.last_upload_device_name() == Some("laptop")
        }));
        assert_eq!(unlisted.visibility(), DeckVisibility::Unlisted);
        assert!(matches!(
            server.inspect_shared_deck(&private_uuid),
            Err(SqliteSyncError::Download(SyncDownloadError::Forbidden))
        ));
    }

    #[test]
    fn sqlite_sync_rejects_owner_mismatch() {
        let temp_dir = unique_temp_dir("sqlite-owner-mismatch");
        let deck_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        let server = SqliteSyncServer::new(&temp_dir);
        server
            .upload(upload_request(
                deck_uuid.clone(),
                DeckVisibility::Private,
                None,
            ))
            .unwrap();
        let request = SyncUploadRequest::from_parts(
            deck_uuid.clone(),
            "acct:other@example.invalid",
            DeckVisibility::Private,
            Some("1"),
            device(),
            Some(content_bundle("two")),
            None,
        )
        .unwrap();

        assert!(matches!(
            server.upload(request),
            Err(SqliteSyncError::Upload(SyncUploadError::Forbidden))
        ));
        assert!(matches!(
            server.download(
                &deck_uuid,
                SyncDownloadKind::OwnerStudyHistory,
                Some("acct:other@example.invalid"),
            ),
            Err(SqliteSyncError::Download(SyncDownloadError::Forbidden))
        ));
    }

    #[test]
    fn sqlite_sync_rejects_unsupported_schema_version() {
        let temp_dir = unique_temp_dir("sqlite-unsupported-version");
        let db_path = temp_dir.join(SQLITE_SYNC_DATABASE_FILE);
        let connection = rusqlite::Connection::open(db_path).unwrap();
        connection
            .execute_batch(
                "
                CREATE TABLE sync_schema_version (
                    version INTEGER PRIMARY KEY
                );
                INSERT INTO sync_schema_version(version) VALUES (99);
                ",
            )
            .unwrap();
        drop(connection);

        let server = SqliteSyncServer::new(&temp_dir);
        assert!(matches!(
            server.list_public_decks(),
            Err(SqliteSyncError::InvalidMetadata(message))
                if message.contains("unsupported sync server schema version")
        ));
    }

    #[test]
    fn sqlite_sync_migrates_version_one_device_uuid_metadata() {
        let temp_dir = unique_temp_dir("sqlite-v1-device-uuid-migration");
        let db_path = temp_dir.join(SQLITE_SYNC_DATABASE_FILE);
        let deck_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        let connection = rusqlite::Connection::open(&db_path).unwrap();
        connection
            .execute_batch(
                "
                CREATE TABLE sync_schema_version (
                    version INTEGER PRIMARY KEY CHECK (version = 1)
                );
                INSERT INTO sync_schema_version(version) VALUES (1);
                CREATE TABLE decks (
                    deck_uuid TEXT PRIMARY KEY NOT NULL,
                    owner_identity TEXT NOT NULL,
                    visibility TEXT NOT NULL,
                    revision INTEGER NOT NULL,
                    content_bundle_hash TEXT NOT NULL,
                    study_history_bundle_hash TEXT,
                    last_upload_device_uuid TEXT NOT NULL,
                    last_upload_device_name TEXT
                );
                ",
            )
            .unwrap();
        let content = content_bundle("v1");
        let bundle_dir = sqlite_deck_bundle_dir(&temp_dir, &deck_uuid);
        fs::create_dir_all(&bundle_dir).unwrap();
        fs::write(
            bundle_dir.join(sqlite_content_bundle_file_name("1")),
            content.bytes(),
        )
        .unwrap();
        connection
            .execute(
                "
                INSERT INTO decks (
                    deck_uuid,
                    owner_identity,
                    visibility,
                    revision,
                    content_bundle_hash,
                    study_history_bundle_hash,
                    last_upload_device_uuid,
                    last_upload_device_name
                )
                VALUES (?1, ?2, ?3, ?4, ?5, NULL, ?6, ?7)
                ",
                params![
                    deck_uuid.as_str(),
                    "acct:modulus@example.invalid",
                    DeckVisibility::Public.code(),
                    1_i64,
                    content.bundle_hash(),
                    "aaaaaaaa-aaaa-4aaa-aaaa-aaaaaaaaaaaa",
                    "laptop",
                ],
            )
            .unwrap();
        drop(connection);

        let server = SqliteSyncServer::new(&temp_dir);
        let download = server
            .download(&deck_uuid, SyncDownloadKind::SharedContent, None)
            .unwrap();
        let connection = rusqlite::Connection::open(&db_path).unwrap();
        let version: i64 = connection
            .query_row("SELECT version FROM sync_schema_version", [], |row| {
                row.get(0)
            })
            .unwrap();
        let device_uuid_column_count: i64 = connection
            .query_row(
                "
                SELECT COUNT(*)
                FROM pragma_table_info('decks')
                WHERE name = 'last_upload_device_uuid'
                ",
                [],
                |row| row.get(0),
            )
            .unwrap();

        assert_eq!(version, 4);
        assert_eq!(device_uuid_column_count, 0);
        assert_eq!(download.last_upload_device_name(), Some("laptop"));
    }

    #[test]
    fn sqlite_sync_migrates_version_two_summary_metadata_columns() {
        let temp_dir = unique_temp_dir("sqlite-v2-summary-migration");
        let db_path = temp_dir.join(SQLITE_SYNC_DATABASE_FILE);
        let deck_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        let connection = rusqlite::Connection::open(&db_path).unwrap();
        connection
            .execute_batch(
                "
                CREATE TABLE sync_schema_version (
                    version INTEGER PRIMARY KEY CHECK (version = 2)
                );
                INSERT INTO sync_schema_version(version) VALUES (2);
                CREATE TABLE decks (
                    deck_uuid TEXT PRIMARY KEY NOT NULL,
                    owner_identity TEXT NOT NULL,
                    visibility TEXT NOT NULL,
                    revision INTEGER NOT NULL,
                    content_bundle_hash TEXT NOT NULL,
                    study_history_bundle_hash TEXT,
                    last_upload_device_name TEXT
                );
                ",
            )
            .unwrap();
        let content = content_bundle("v2");
        let bundle_dir = sqlite_deck_bundle_dir(&temp_dir, &deck_uuid);
        fs::create_dir_all(&bundle_dir).unwrap();
        fs::write(
            bundle_dir.join(sqlite_content_bundle_file_name("1")),
            content.bytes(),
        )
        .unwrap();
        connection
            .execute(
                "
                INSERT INTO decks (
                    deck_uuid,
                    owner_identity,
                    visibility,
                    revision,
                    content_bundle_hash,
                    study_history_bundle_hash,
                    last_upload_device_name
                )
                VALUES (?1, ?2, ?3, ?4, ?5, NULL, ?6)
                ",
                params![
                    deck_uuid.as_str(),
                    "acct:modulus@example.invalid",
                    DeckVisibility::Public.code(),
                    1_i64,
                    content.bundle_hash(),
                    "laptop",
                ],
            )
            .unwrap();
        drop(connection);

        let server = SqliteSyncServer::new(&temp_dir);
        let public_decks = server.list_public_decks().unwrap();
        let connection = rusqlite::Connection::open(&db_path).unwrap();
        let version: i64 = connection
            .query_row("SELECT version FROM sync_schema_version", [], |row| {
                row.get(0)
            })
            .unwrap();
        let summary_column_count: i64 = connection
            .query_row(
                "
                SELECT COUNT(*)
                FROM pragma_table_info('decks')
                WHERE name = 'deck_audio_formats'
                ",
                [],
                |row| row.get(0),
            )
            .unwrap();

        assert_eq!(version, 4);
        assert_eq!(summary_column_count, 1);
        assert_eq!(public_decks.len(), 1);
        assert_eq!(public_decks[0].summary(), &SyncDeckSummary::unknown());
    }

    #[test]
    fn sqlite_sync_migrates_version_three_to_remove_obsolete_oauth_application_cache() {
        let temp_dir = unique_temp_dir("sqlite-v3-oauth-cache-cleanup");
        let db_path = temp_dir.join(SQLITE_SYNC_DATABASE_FILE);
        let connection = rusqlite::Connection::open(&db_path).unwrap();
        connection
            .execute_batch(
                "
                CREATE TABLE sync_schema_version (
                    version INTEGER PRIMARY KEY CHECK (version = 3)
                );
                INSERT INTO sync_schema_version(version) VALUES (3);
                CREATE TABLE decks (
                    deck_uuid TEXT PRIMARY KEY NOT NULL,
                    owner_identity TEXT NOT NULL,
                    visibility TEXT NOT NULL,
                    revision INTEGER NOT NULL,
                    content_bundle_hash TEXT NOT NULL,
                    study_history_bundle_hash TEXT,
                    last_upload_device_name TEXT,
                    deck_title TEXT,
                    deck_name TEXT,
                    deck_card_count INTEGER,
                    deck_audio_file_count INTEGER,
                    deck_managed_audio_file_count INTEGER,
                    deck_audio_formats TEXT NOT NULL DEFAULT ''
                );
                CREATE TABLE auth_sessions (
                    bearer_token TEXT PRIMARY KEY NOT NULL,
                    owner_identity TEXT NOT NULL,
                    created_at_unix_seconds INTEGER NOT NULL
                );
                CREATE TABLE oauth_pending_logins (
                    login_token TEXT PRIMARY KEY NOT NULL,
                    instance_url TEXT NOT NULL,
                    client_id TEXT NOT NULL,
                    client_secret TEXT NOT NULL,
                    redirect_uri TEXT NOT NULL DEFAULT 'urn:ietf:wg:oauth:2.0:oob',
                    expected_owner_identity TEXT,
                    created_at_unix_seconds INTEGER NOT NULL,
                    accepted_notice_version TEXT,
                    accepted_notice_hash TEXT,
                    accepted_lawful_basis TEXT,
                    accepted_at_unix_seconds INTEGER
                );
                CREATE INDEX idx_oauth_pending_logins_instance
                    ON oauth_pending_logins(instance_url);
                CREATE TABLE oauth_applications (
                    instance_url TEXT PRIMARY KEY NOT NULL,
                    client_id TEXT NOT NULL,
                    client_secret TEXT NOT NULL,
                    created_at_unix_seconds INTEGER NOT NULL
                );
                CREATE TABLE oauth_start_attempts (
                    instance_url TEXT NOT NULL,
                    client_key TEXT NOT NULL,
                    attempted_at_unix_seconds INTEGER NOT NULL
                );
                ",
            )
            .unwrap();
        connection
            .execute(
                "
                INSERT INTO oauth_pending_logins (
                    login_token,
                    instance_url,
                    client_id,
                    client_secret,
                    redirect_uri,
                    created_at_unix_seconds
                )
                VALUES (?1, ?2, ?3, ?4, ?5, ?6)
                ",
                params![
                    "pending-login",
                    "https://example.invalid",
                    "client id",
                    "client secret",
                    "https://sync.example.invalid/auth/fedi/callback",
                    123_i64,
                ],
            )
            .unwrap();
        connection
            .execute(
                "
                INSERT INTO oauth_applications (
                    instance_url,
                    client_id,
                    client_secret,
                    created_at_unix_seconds
                )
                VALUES (?1, ?2, ?3, ?4)
                ",
                params![
                    "https://example.invalid",
                    "cached client id",
                    "cached client secret",
                    123_i64,
                ],
            )
            .unwrap();
        drop(connection);

        let server = SqliteSyncServer::new(&temp_dir);
        assert!(server.list_public_decks().unwrap().is_empty());

        let connection = rusqlite::Connection::open(&db_path).unwrap();
        let version: i64 = connection
            .query_row("SELECT version FROM sync_schema_version", [], |row| {
                row.get(0)
            })
            .unwrap();
        let application_table_count: i64 = connection
            .query_row(
                "
                SELECT COUNT(*)
                FROM sqlite_master
                WHERE type = 'table' AND name = 'oauth_applications'
                ",
                [],
                |row| row.get(0),
            )
            .unwrap();
        let redirect_default: Option<String> = connection
            .query_row(
                "
                SELECT dflt_value
                FROM pragma_table_info('oauth_pending_logins')
                WHERE name = 'redirect_uri'
                ",
                [],
                |row| row.get(0),
            )
            .unwrap();
        let pending_redirect_uri: String = connection
            .query_row(
                "
                SELECT redirect_uri
                FROM oauth_pending_logins
                WHERE login_token = 'pending-login'
                ",
                [],
                |row| row.get(0),
            )
            .unwrap();

        assert_eq!(version, 4);
        assert_eq!(application_table_count, 0);
        assert_eq!(redirect_default, None);
        assert_eq!(
            pending_redirect_uri,
            "https://sync.example.invalid/auth/fedi/callback"
        );
    }

    #[test]
    fn in_memory_sync_reserves_deleted_deck_uuid_for_original_owner() {
        let deck_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        let mut server = InMemorySyncServer::new();

        server
            .upload(upload_request(
                deck_uuid.clone(),
                DeckVisibility::Public,
                None,
            ))
            .unwrap();
        server
            .delete(&deck_uuid, "acct:modulus@example.invalid")
            .unwrap();

        assert_eq!(
            server.upload(other_owner_upload_request(deck_uuid.clone())),
            Err(SyncUploadError::ReservedDeckIdentity)
        );
        server
            .upload(upload_request(
                deck_uuid.clone(),
                DeckVisibility::Public,
                None,
            ))
            .unwrap();
        assert_eq!(
            server
                .inspect_owner_deck(&deck_uuid, "acct:modulus@example.invalid")
                .unwrap()
                .owner_identity(),
            "acct:modulus@example.invalid"
        );
    }

    #[test]
    fn in_memory_sync_account_deletion_retires_deck_uuid_identity() {
        let deck_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        let mut server = InMemorySyncServer::new();

        server
            .upload(upload_request(
                deck_uuid.clone(),
                DeckVisibility::Public,
                None,
            ))
            .unwrap();
        server
            .delete_owner_account("acct:modulus@example.invalid")
            .unwrap();

        assert_eq!(
            server.upload(upload_request(
                deck_uuid.clone(),
                DeckVisibility::Public,
                None
            )),
            Err(SyncUploadError::RetiredDeckIdentity)
        );
        assert_eq!(
            server.upload(other_owner_upload_request(deck_uuid)),
            Err(SyncUploadError::RetiredDeckIdentity)
        );
    }

    #[test]
    fn in_memory_sync_account_deletion_retires_previously_deleted_deck_uuid_identity() {
        let deck_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        let mut server = InMemorySyncServer::new();

        server
            .upload(upload_request(
                deck_uuid.clone(),
                DeckVisibility::Public,
                None,
            ))
            .unwrap();
        server
            .delete(&deck_uuid, "acct:modulus@example.invalid")
            .unwrap();
        server
            .delete_owner_account("acct:modulus@example.invalid")
            .unwrap();

        assert_eq!(
            server.upload(other_owner_upload_request(deck_uuid.clone())),
            Err(SyncUploadError::RetiredDeckIdentity)
        );
        assert_eq!(
            server.upload(upload_request(deck_uuid, DeckVisibility::Public, None)),
            Err(SyncUploadError::RetiredDeckIdentity)
        );
    }

    #[test]
    fn sqlite_sync_reserves_deleted_deck_uuid_for_original_owner() {
        let temp_dir = unique_temp_dir("sqlite-deleted-deck-owner-reservation");
        let deck_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        let server = SqliteSyncServer::new(&temp_dir);

        server
            .upload(upload_request(
                deck_uuid.clone(),
                DeckVisibility::Public,
                None,
            ))
            .unwrap();
        server
            .delete(&deck_uuid, "acct:modulus@example.invalid")
            .unwrap();

        assert_eq!(
            sqlite_deck_identity_owner(&temp_dir, &deck_uuid),
            Some(Some("acct:modulus@example.invalid".to_string()))
        );
        assert!(
            server
                .download(&deck_uuid, SyncDownloadKind::SharedContent, None)
                .is_err()
        );
        assert!(matches!(
            server.upload(other_owner_upload_request(deck_uuid.clone())),
            Err(SqliteSyncError::Upload(
                SyncUploadError::ReservedDeckIdentity
            ))
        ));
        server
            .upload(upload_request(
                deck_uuid.clone(),
                DeckVisibility::Public,
                None,
            ))
            .unwrap();
        assert_eq!(
            server
                .inspect_owner_deck(&deck_uuid, "acct:modulus@example.invalid")
                .unwrap()
                .owner_identity(),
            "acct:modulus@example.invalid"
        );
    }

    #[test]
    fn sqlite_sync_account_deletion_retires_deck_uuid_identity_without_owner_pii() {
        let temp_dir = unique_temp_dir("sqlite-account-delete-deck-identity");
        let deck_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        let server = SqliteSyncServer::new(&temp_dir);

        server
            .upload(upload_request(
                deck_uuid.clone(),
                DeckVisibility::Public,
                None,
            ))
            .unwrap();
        let outcome = server
            .delete_owner_account("acct:modulus@example.invalid")
            .unwrap();

        assert_eq!(outcome.decks_deleted(), 1);
        assert_eq!(
            sqlite_deck_identity_owner(&temp_dir, &deck_uuid),
            Some(None)
        );
        assert!(matches!(
            server.upload(upload_request(
                deck_uuid.clone(),
                DeckVisibility::Public,
                None
            )),
            Err(SqliteSyncError::Upload(
                SyncUploadError::RetiredDeckIdentity
            ))
        ));
        assert!(matches!(
            server.upload(other_owner_upload_request(deck_uuid)),
            Err(SqliteSyncError::Upload(
                SyncUploadError::RetiredDeckIdentity
            ))
        ));
    }

    #[test]
    fn sqlite_sync_account_deletion_retires_previously_deleted_deck_uuid_identity() {
        let temp_dir = unique_temp_dir("sqlite-account-delete-deleted-deck-identity");
        let deck_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        let server = SqliteSyncServer::new(&temp_dir);

        server
            .upload(upload_request(
                deck_uuid.clone(),
                DeckVisibility::Public,
                None,
            ))
            .unwrap();
        server
            .delete(&deck_uuid, "acct:modulus@example.invalid")
            .unwrap();
        let outcome = server
            .delete_owner_account("acct:modulus@example.invalid")
            .unwrap();

        assert_eq!(outcome.decks_deleted(), 0);
        assert_eq!(
            sqlite_deck_identity_owner(&temp_dir, &deck_uuid),
            Some(None)
        );
        assert!(matches!(
            server.upload(other_owner_upload_request(deck_uuid.clone())),
            Err(SqliteSyncError::Upload(
                SyncUploadError::RetiredDeckIdentity
            ))
        ));
        assert!(matches!(
            server.upload(upload_request(deck_uuid, DeckVisibility::Public, None)),
            Err(SqliteSyncError::Upload(
                SyncUploadError::RetiredDeckIdentity
            ))
        ));
    }

    #[test]
    fn sqlite_sync_backfills_identity_owners_for_existing_version_four_decks() {
        let temp_dir = unique_temp_dir("sqlite-v4-identity-backfill");
        let deck_uuid = uuid("550e8400-e29b-41d4-a716-446655440000");
        fs::create_dir_all(&temp_dir).unwrap();
        let connection = Connection::open(sqlite_sync_database_path(&temp_dir).unwrap()).unwrap();
        connection
            .execute_batch(
                "
                CREATE TABLE sync_schema_version (
                    version INTEGER PRIMARY KEY CHECK (version = 4)
                );
                INSERT INTO sync_schema_version(version) VALUES (4);
                CREATE TABLE decks (
                    deck_uuid TEXT PRIMARY KEY NOT NULL,
                    owner_identity TEXT NOT NULL,
                    visibility TEXT NOT NULL,
                    revision INTEGER NOT NULL,
                    content_bundle_hash TEXT NOT NULL,
                    study_history_bundle_hash TEXT,
                    last_upload_device_name TEXT,
                    deck_title TEXT,
                    deck_name TEXT,
                    deck_card_count INTEGER,
                    deck_audio_file_count INTEGER,
                    deck_managed_audio_file_count INTEGER,
                    deck_audio_formats TEXT NOT NULL DEFAULT ''
                );
                ",
            )
            .unwrap();
        connection
            .execute(
                "
                INSERT INTO decks (
                    deck_uuid,
                    owner_identity,
                    visibility,
                    revision,
                    content_bundle_hash,
                    deck_audio_formats
                )
                VALUES (?1, ?2, 'public', 1, 'sha256:content', '')
                ",
                params![deck_uuid.as_str(), "acct:modulus@example.invalid"],
            )
            .unwrap();
        drop(connection);

        SqliteSyncServer::new(&temp_dir)
            .list_public_decks()
            .unwrap();

        assert_eq!(
            sqlite_deck_identity_owner(&temp_dir, &deck_uuid),
            Some(Some("acct:modulus@example.invalid".to_string()))
        );
    }

    proptest! {
        #![proptest_config(ProptestConfig {
            cases: 64,
            failure_persistence: None,
            ..ProptestConfig::default()
        })]

        #[test]
        fn sqlite_sync_storage_invariants_survive_generated_operation_sequences(
            operations in prop::collection::vec(arbitrary_sqlite_storage_operation(), 1..80)
        ) {
            run_sqlite_storage_operation_sequence(&operations)?;
        }

        #[test]
        fn generated_uuid_bytes_always_make_parseable_v4_uuid(bytes in any::<[u8; 16]>()) {
            let uuid = SyncUuid::from_random_bytes(bytes);
            let text = uuid.as_str();

            prop_assert_eq!(text.len(), 36);
            prop_assert_eq!(&text[14..15], "4");
            prop_assert!(matches!(text.as_bytes()[19], b'8' | b'9' | b'a' | b'b'));
            prop_assert_eq!(text.parse::<SyncUuid>().unwrap(), uuid);
        }

        #[test]
        fn local_deck_sync_state_normalizes_generated_single_line_fields(
            deck_uuid in generated_uuid(),
            server_url in single_line_text(),
            owner in optional_single_line_text(),
            revision in optional_single_line_text(),
            uploaded_hash in optional_single_line_text(),
            downloaded_hash in optional_single_line_text(),
        ) {
            let padded_server_url = format!(" {server_url} ");
            let padded_owner = owner.as_ref().map(|value| format!(" {value} "));
            let padded_revision = revision.as_ref().map(|value| format!(" {value} "));
            let padded_uploaded_hash = uploaded_hash.as_ref().map(|value| format!(" {value} "));
            let padded_downloaded_hash = downloaded_hash.as_ref().map(|value| format!(" {value} "));

            let state = LocalDeckSyncState::from_parts(
                deck_uuid.clone(),
                &padded_server_url,
                padded_owner.as_deref(),
                padded_revision.as_deref(),
                padded_uploaded_hash.as_deref(),
                padded_downloaded_hash.as_deref(),
            )
            .unwrap();

            prop_assert_eq!(state.deck_uuid(), &deck_uuid);
            prop_assert_eq!(state.sync_server_url(), server_url);
            prop_assert_eq!(state.owner_identity(), owner.as_deref());
            prop_assert_eq!(state.last_seen_remote_revision(), revision.as_deref());
            prop_assert_eq!(state.last_uploaded_local_bundle_hash(), uploaded_hash.as_deref());
            prop_assert_eq!(state.last_downloaded_remote_bundle_hash(), downloaded_hash.as_deref());
        }

        #[test]
        fn sync_bundle_payload_preserves_non_empty_bytes_hash_and_database_flag(
            bytes in prop::collection::vec(any::<u8>(), 1..128),
            includes_review_database in any::<bool>(),
        ) {
            let bundle_hash = sync_bundle_hash(&bytes);
            let payload = SyncBundlePayload::from_parts(
                bytes.clone(),
                &format!(" {bundle_hash} "),
                includes_review_database,
            )
            .unwrap();

            prop_assert_eq!(payload.bytes(), bytes.as_slice());
            prop_assert_eq!(payload.bundle_hash(), bundle_hash);
            prop_assert_eq!(payload.includes_review_database(), includes_review_database);
        }

        #[test]
        fn sync_bundle_payload_rejects_hash_mismatches(
            bytes in prop::collection::vec(any::<u8>(), 1..128),
            includes_review_database in any::<bool>(),
        ) {
            let mismatched_hash =
                "sha256:0000000000000000000000000000000000000000000000000000000000000000"
                    .to_string();
            let error = SyncBundlePayload::from_parts(
                bytes.clone(),
                &mismatched_hash,
                includes_review_database,
            )
            .unwrap_err();

            prop_assert_eq!(
                error,
                SyncValidationError::BundleHashMismatch {
                    declared: mismatched_hash,
                    actual: sync_bundle_hash(&bytes),
                }
            );
        }

        #[test]
        fn shared_download_never_returns_review_database_for_any_visibility(
            deck_uuid in generated_uuid(),
            visibility in visibility_strategy(),
            owner in single_line_text(),
            content_bytes in prop::collection::vec(any::<u8>(), 1..128),
            history_bytes in prop::collection::vec(any::<u8>(), 1..128),
            include_history in any::<bool>(),
        ) {
            let content_hash = sync_bundle_hash(&content_bytes);
            let history_hash = sync_bundle_hash(&history_bytes);
            let content = SyncBundlePayload::from_parts(content_bytes.clone(), &content_hash, false).unwrap();
            let history = SyncBundlePayload::from_parts(history_bytes.clone(), &history_hash, true).unwrap();
            let mut server = InMemorySyncServer::new();
            server
                .upload(upload_request_with_payloads(
                    deck_uuid.clone(),
                    &owner,
                    visibility,
                    None,
                    content.clone(),
                    include_history.then_some(history.clone()),
                ))
                .unwrap();

            let shared = server.download(&deck_uuid, SyncDownloadKind::SharedContent, None);
            if visibility == DeckVisibility::Private {
                prop_assert_eq!(shared, Err(SyncDownloadError::Forbidden));
            } else {
                let shared = shared.unwrap();
                prop_assert!(!shared.bundle().includes_review_database());
                prop_assert_eq!(shared.bundle().bytes(), content_bytes.as_slice());
                prop_assert_eq!(shared.bundle().bundle_hash(), content_hash.as_str());
            }

            let owner_download = server
                .download(&deck_uuid, SyncDownloadKind::OwnerStudyHistory, Some(&owner))
                .unwrap();
            prop_assert_eq!(owner_download.bundle().includes_review_database(), include_history);
            if include_history {
                prop_assert_eq!(owner_download.bundle().bytes(), history_bytes.as_slice());
                prop_assert_eq!(owner_download.bundle().bundle_hash(), history_hash);
            } else {
                prop_assert_eq!(owner_download.bundle().bytes(), content_bytes.as_slice());
                prop_assert_eq!(owner_download.bundle().bundle_hash(), content_hash.as_str());
            }
        }

        #[test]
        fn uploads_require_the_latest_seen_revision(
            deck_uuid in generated_uuid(),
            owner in single_line_text(),
            visibilities in prop::collection::vec(visibility_strategy(), 1..10),
        ) {
            let mut server = InMemorySyncServer::new();
            let mut current_revision: Option<String> = None;

            for (index, visibility) in visibilities.into_iter().enumerate() {
                let content = content_bundle(&format!("content-{index}"));
                let history = study_history_bundle(&format!("history-{index}"));

                if current_revision.is_some() {
                    let blind_result = server.upload(upload_request_with_payloads(
                        deck_uuid.clone(),
                        &owner,
                        visibility,
                        None,
                        content.clone(),
                        Some(history.clone()),
                    ));
                    let blind_result_is_conflict =
                        matches!(blind_result, Err(SyncUploadError::Conflict { .. }));
                    prop_assert!(blind_result_is_conflict);
                }

                let outcome = server
                    .upload(upload_request_with_payloads(
                        deck_uuid.clone(),
                        &owner,
                        visibility,
                        current_revision.as_deref(),
                        content,
                        Some(history),
                    ))
                    .unwrap();
                prop_assert_eq!(outcome.revision(), (index + 1).to_string());
                current_revision = Some(outcome.revision().to_string());
            }
        }

        #[test]
        fn sqlite_conflicting_uploads_do_not_publish_or_leave_staged_bundles(
            deck_uuid in generated_uuid(),
            owner in single_line_text(),
            first_text in single_line_text(),
            conflicting_text in single_line_text(),
        ) {
            let temp_dir = unique_temp_dir("sqlite-conflict-staged-upload");
            let server = SqliteSyncServer::new(&temp_dir);
            server
                .upload(upload_request_with_payloads(
                    deck_uuid.clone(),
                    &owner,
                    DeckVisibility::Private,
                    None,
                    content_bundle(&first_text),
                    None,
                ))
                .unwrap();

            let winner = server
                .upload(upload_request_with_payloads(
                    deck_uuid.clone(),
                    &owner,
                    DeckVisibility::Private,
                    Some("1"),
                    content_bundle("winning machine"),
                    None,
                ))
                .unwrap();
            prop_assert_eq!(winner.revision(), "2");

            let conflict = server.upload(upload_request_with_payloads(
                deck_uuid.clone(),
                &owner,
                DeckVisibility::Private,
                Some("1"),
                content_bundle(&conflicting_text),
                None,
            ));

            let conflict_is_revision_conflict = matches!(
                conflict,
                Err(SqliteSyncError::Upload(SyncUploadError::Conflict { .. }))
            );
            prop_assert!(conflict_is_revision_conflict);
            let download = server
                .download(&deck_uuid, SyncDownloadKind::OwnerStudyHistory, Some(&owner))
                .unwrap();
            prop_assert_eq!(download.revision(), "2");

            let bundle_dir = sqlite_deck_bundle_dir(&temp_dir, &deck_uuid);
            let entries = fs::read_dir(bundle_dir).unwrap();
            for entry in entries {
                let file_name = entry.unwrap().file_name();
                let file_name = file_name.to_string_lossy();
                prop_assert!(!file_name.starts_with(".upload-"));
                prop_assert_ne!(file_name.as_ref(), "study-history-r2.bundle");
                prop_assert_ne!(file_name.as_ref(), "content-r3.bundle");
                prop_assert_ne!(file_name.as_ref(), "study-history-r3.bundle");
            }
        }

        #[test]
        fn sync_delete_requires_owner_and_removes_remote_deck(
            deck_uuid in generated_uuid(),
            owner in single_line_text(),
            other_owner in single_line_text(),
            visibility in visibility_strategy(),
        ) {
            prop_assume!(owner != other_owner);
            let mut server = InMemorySyncServer::new();
            server
                .upload(upload_request_with_payloads(
                    deck_uuid.clone(),
                    &owner,
                    visibility,
                    None,
                    content_bundle("delete-property"),
                    Some(study_history_bundle("delete-property")),
                ))
                .unwrap();

            prop_assert_eq!(
                server.delete(&deck_uuid, &other_owner),
                Err(SyncDeleteError::Forbidden)
            );
            prop_assert!(server
                .download(&deck_uuid, SyncDownloadKind::OwnerStudyHistory, Some(&owner))
                .is_ok());

            server.delete(&deck_uuid, &owner).unwrap();

            prop_assert_eq!(
                server.download(&deck_uuid, SyncDownloadKind::OwnerStudyHistory, Some(&owner)),
                Err(SyncDownloadError::NotFound)
            );
            prop_assert_eq!(
                server.inspect_shared_deck(&deck_uuid),
                Err(SyncDownloadError::NotFound)
            );
            prop_assert!(!server
                .list_owner_decks(&owner)
                .iter()
                .any(|status| status.deck_uuid() == &deck_uuid));
        }

    }
}
