use std::collections::{HashMap, HashSet};
use std::env;
use std::fmt;
use std::fs;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::time::{SystemTime, UNIX_EPOCH};

use rusqlite::{Connection, OptionalExtension, params};
use serde::{Deserialize, Serialize};
use serde_json::Value;

use crate::error::{MegaError, Result};
use crate::fs::upload_state::UploadState;
use crate::fs::{Node, NodeType};

pub(crate) const ENGINE_STATE_SCHEMA_VERSION: u32 = 1;
pub(crate) const SQLITE_PERSISTENCE_BACKEND_SCHEMA_VERSION: u32 = 1;
pub(crate) const SQLITE_TRANSFER_CACHE_SCHEMA_VERSION: u32 = 1;

const SQLITE_META_SCHEMA_VERSION_KEY: &str = "backend_schema_version";
const SQLITE_TRANSFER_CACHE_META_SCHEMA_VERSION_KEY: &str = "transfer_cache_schema_version";
const SQLITE_ENGINE_STATE_SINGLETON_SLOT: i64 = 0;

#[derive(Debug)]
enum SqliteSchemaInitError {
    RecycleRequired(String),
    Fatal(MegaError),
}

#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub(crate) struct PersistenceScope {
    pub(crate) account_handle: String,
}

impl PersistenceScope {
    pub(crate) fn new(account_handle: impl Into<String>) -> Self {
        Self {
            account_handle: account_handle.into(),
        }
    }
}

fn default_persistence_root() -> Result<PathBuf> {
    #[cfg(target_os = "windows")]
    {
        let base = env::var_os("LOCALAPPDATA").ok_or_else(|| {
            MegaError::Custom("LOCALAPPDATA is not set for persistence root resolution".to_string())
        })?;
        return Ok(PathBuf::from(base).join("megalib"));
    }

    #[cfg(target_os = "macos")]
    {
        let home = env::var_os("HOME").ok_or_else(|| {
            MegaError::Custom("HOME is not set for persistence root resolution".to_string())
        })?;
        Ok(PathBuf::from(home)
            .join("Library")
            .join("Application Support")
            .join("megalib"))
    }

    #[cfg(not(any(target_os = "windows", target_os = "macos")))]
    {
        if let Some(xdg_state_home) = env::var_os("XDG_STATE_HOME") {
            return Ok(PathBuf::from(xdg_state_home).join("megalib"));
        }

        let home = env::var_os("HOME").ok_or_else(|| {
            MegaError::Custom(
                "HOME is not set for persistence root resolution and XDG_STATE_HOME is absent"
                    .to_string(),
            )
        })?;
        Ok(PathBuf::from(home)
            .join(".local")
            .join("state")
            .join("megalib"))
    }
}

#[cfg_attr(not(test), allow(dead_code))]
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub(crate) enum TransferCheckpointKind {
    Upload,
    Download,
}

#[cfg_attr(not(test), allow(dead_code))]
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub(crate) enum TransferCheckpointKey {
    Upload {
        source_fingerprint: String,
        local_path: PathBuf,
    },
    Download {
        source_fingerprint: String,
        source_node_handle: String,
    },
}

#[cfg_attr(not(test), allow(dead_code))]
impl TransferCheckpointKey {
    pub(crate) fn upload(source_fingerprint: impl Into<String>, local_path: PathBuf) -> Self {
        Self::Upload {
            source_fingerprint: source_fingerprint.into(),
            local_path,
        }
    }

    pub(crate) fn download(
        source_fingerprint: impl Into<String>,
        source_node_handle: impl Into<String>,
    ) -> Self {
        Self::Download {
            source_fingerprint: source_fingerprint.into(),
            source_node_handle: source_node_handle.into(),
        }
    }

    fn kind(&self) -> TransferCheckpointKind {
        match self {
            Self::Upload { .. } => TransferCheckpointKind::Upload,
            Self::Download { .. } => TransferCheckpointKind::Download,
        }
    }

    fn source_fingerprint(&self) -> &str {
        match self {
            Self::Upload {
                source_fingerprint, ..
            }
            | Self::Download {
                source_fingerprint, ..
            } => source_fingerprint,
        }
    }

    fn storage_disambiguator(&self) -> String {
        match self {
            Self::Upload { local_path, .. } => local_path.to_string_lossy().into_owned(),
            Self::Download {
                source_node_handle, ..
            } => source_node_handle.clone(),
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub(crate) struct TransferCheckpointRecord {
    pub(crate) key: TransferCheckpointKey,
    pub(crate) committed_bytes: u64,
    pub(crate) temp_urls: Vec<String>,
    pub(crate) payload: TransferCheckpointPayload,
}

impl TransferCheckpointRecord {
    pub(crate) fn from_upload_state(source_path: &Path, state: &UploadState) -> Self {
        Self {
            key: TransferCheckpointKey::upload(state.file_hash.clone(), source_path.to_path_buf()),
            committed_bytes: state.offset,
            temp_urls: vec![state.upload_url.clone()],
            payload: TransferCheckpointPayload::Upload(
                PersistedUploadCheckpoint::from_upload_state(state),
            ),
        }
    }

    pub(crate) fn into_upload_state(self) -> Result<UploadState> {
        let key = self.key;
        let committed_bytes = self.committed_bytes;
        let temp_urls = self.temp_urls;
        match (key, self.payload) {
            (
                TransferCheckpointKey::Upload {
                    source_fingerprint, ..
                },
                TransferCheckpointPayload::Upload(payload),
            ) => payload.into_upload_state(source_fingerprint, committed_bytes, temp_urls),
            _ => Err(MegaError::Custom(
                "Transfer checkpoint did not contain an upload payload".to_string(),
            )),
        }
    }

    fn matches_key(&self, key: &TransferCheckpointKey) -> bool {
        if &self.key != key {
            return false;
        }

        matches!(
            (&self.key, &self.payload),
            (
                TransferCheckpointKey::Upload { .. },
                TransferCheckpointPayload::Upload(_)
            ) | (
                TransferCheckpointKey::Download { .. },
                TransferCheckpointPayload::Download(_)
            )
        )
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub(crate) enum TransferCheckpointPayload {
    Upload(PersistedUploadCheckpoint),
    Download(PersistedDownloadCheckpoint),
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub(crate) struct PersistedUploadCheckpoint {
    pub(crate) file_key: [u8; 16],
    pub(crate) nonce: [u8; 8],
    pub(crate) chunk_macs: Vec<[u8; 16]>,
    pub(crate) file_size: u64,
    pub(crate) file_name: String,
    pub(crate) parent_handle: String,
    pub(crate) source_fingerprint: String,
    pub(crate) created_at: i64,
    pub(crate) upload_token: Option<String>,
}

impl PersistedUploadCheckpoint {
    fn from_upload_state(state: &UploadState) -> Self {
        Self {
            file_key: state.file_key,
            nonce: state.nonce,
            chunk_macs: state.chunk_macs.clone(),
            file_size: state.file_size,
            file_name: state.file_name.clone(),
            parent_handle: state.parent_handle.clone(),
            source_fingerprint: state.file_hash.clone(),
            created_at: state.created_at,
            upload_token: None,
        }
    }

    fn into_upload_state(
        self,
        source_fingerprint: String,
        committed_bytes: u64,
        temp_urls: Vec<String>,
    ) -> Result<UploadState> {
        if self.source_fingerprint != source_fingerprint {
            return Err(MegaError::Custom(
                "Transfer checkpoint fingerprint did not match its storage key".to_string(),
            ));
        }

        if committed_bytes > self.file_size {
            return Err(MegaError::Custom(format!(
                "Transfer checkpoint progress {committed_bytes} exceeds file size {}",
                self.file_size
            )));
        }

        let upload_url = match temp_urls.as_slice() {
            [upload_url] => upload_url.clone(),
            _ => {
                return Err(MegaError::Custom(
                    "Upload checkpoint did not contain exactly one temp URL".to_string(),
                ));
            }
        };

        Ok(UploadState {
            upload_url,
            file_key: self.file_key,
            nonce: self.nonce,
            chunk_macs: self.chunk_macs,
            offset: committed_bytes,
            file_size: self.file_size,
            file_name: self.file_name,
            parent_handle: self.parent_handle,
            file_hash: self.source_fingerprint,
            created_at: self.created_at,
        })
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub(crate) struct PersistedDownloadCheckpoint {
    pub(crate) source_node_handle: String,
    pub(crate) temp_path: PathBuf,
    pub(crate) target_path: PathBuf,
    pub(crate) file_size: u64,
    pub(crate) temp_file_size: u64,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct PersistedEngineState {
    pub(crate) schema_version: u32,
    pub(crate) sc: PersistedScState,
    pub(crate) alerts: PersistedAlertsState,
    pub(crate) tree: Option<PersistedTreeState>,
}

impl PersistedEngineState {
    pub(crate) fn validate_schema_version(&self) -> Result<()> {
        if self.schema_version == ENGINE_STATE_SCHEMA_VERSION {
            Ok(())
        } else {
            Err(MegaError::Custom(format!(
                "Unsupported persistence schema version: {}",
                self.schema_version
            )))
        }
    }
}

#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub(crate) struct PersistedScState {
    pub(crate) scsn: Option<String>,
}

#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub(crate) struct PersistedAlertsState {
    pub(crate) alerts_catchup_pending: bool,
    pub(crate) user_alert_lsn: Option<String>,
    pub(crate) user_alerts: Vec<Value>,
}

#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub(crate) struct PersistedTreeState {
    pub(crate) nodes: Vec<PersistedNodeRecord>,
    pub(crate) pending_nodes: Vec<Value>,
    pub(crate) outshares: HashMap<String, HashSet<String>>,
    pub(crate) pending_outshares: HashMap<String, HashSet<String>>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct PersistedNodeRecord {
    pub(crate) name: String,
    pub(crate) handle: String,
    pub(crate) parent_handle: Option<String>,
    pub(crate) node_type: NodeType,
    pub(crate) size: u64,
    pub(crate) timestamp: i64,
    pub(crate) key: Vec<u8>,
    pub(crate) link: Option<String>,
    pub(crate) file_attr: Option<String>,
    pub(crate) share_key: Option<[u8; 16]>,
    pub(crate) share_handle: Option<String>,
    pub(crate) is_inshare: bool,
    pub(crate) is_outshare: bool,
    pub(crate) share_access: Option<i32>,
}

impl From<&Node> for PersistedNodeRecord {
    fn from(node: &Node) -> Self {
        Self {
            name: node.name.clone(),
            handle: node.handle.clone(),
            parent_handle: node.parent_handle.clone(),
            node_type: node.node_type,
            size: node.size,
            timestamp: node.timestamp,
            key: node.key.clone(),
            link: node.link.clone(),
            file_attr: node.file_attr.clone(),
            share_key: node.share_key,
            share_handle: node.share_handle.clone(),
            is_inshare: node.is_inshare,
            is_outshare: node.is_outshare,
            share_access: node.share_access,
        }
    }
}

impl PersistedNodeRecord {
    pub(crate) fn into_node(self) -> Node {
        Node {
            name: self.name,
            handle: self.handle,
            parent_handle: self.parent_handle,
            node_type: self.node_type,
            size: self.size,
            timestamp: self.timestamp,
            key: self.key,
            path: None,
            link: self.link,
            file_attr: self.file_attr,
            share_key: self.share_key,
            share_handle: self.share_handle,
            is_inshare: self.is_inshare,
            is_outshare: self.is_outshare,
            share_access: self.share_access,
        }
    }
}

#[cfg_attr(not(test), allow(dead_code))]
#[derive(Debug, Clone)]
pub(crate) struct SqlitePersistenceBackend {
    root: PathBuf,
}

#[cfg_attr(not(test), allow(dead_code))]
impl SqlitePersistenceBackend {
    pub(crate) fn new(root: PathBuf) -> Self {
        Self { root }
    }

    fn engine_db_path(&self, scope: &PersistenceScope) -> PathBuf {
        self.root.join(format!(
            "acct-{}.sqlite3",
            hex::encode(scope.account_handle.as_bytes())
        ))
    }

    fn transfer_db_path(&self, scope: &PersistenceScope) -> PathBuf {
        self.root.join(format!(
            "acct-{}.transfers.sqlite3",
            hex::encode(scope.account_handle.as_bytes())
        ))
    }

    fn open_engine_connection(&self, scope: &PersistenceScope) -> Result<Connection> {
        self.ensure_root_dir()?;
        let path = self.engine_db_path(scope);
        match self.open_connection_once(&path) {
            Ok(mut conn) => match self.ensure_engine_schema(&mut conn, &path) {
                Ok(()) => Ok(conn),
                Err(SqliteSchemaInitError::RecycleRequired(reason)) => {
                    drop(conn);
                    self.recycle_database(&path, &reason)?;
                    let mut recycled = self.open_connection_once(&path)?;
                    self.ensure_engine_schema(&mut recycled, &path)
                        .map_err(Self::schema_init_error_to_mega_error)?;
                    Ok(recycled)
                }
                Err(SqliteSchemaInitError::Fatal(err)) => Err(err),
            },
            Err(err) => Err(err),
        }
    }

    fn open_transfer_connection(&self, scope: &PersistenceScope) -> Result<Connection> {
        self.ensure_root_dir()?;
        let path = self.transfer_db_path(scope);
        match self.open_connection_once(&path) {
            Ok(mut conn) => match self.ensure_transfer_schema(&mut conn, &path) {
                Ok(()) => Ok(conn),
                Err(SqliteSchemaInitError::RecycleRequired(reason)) => {
                    drop(conn);
                    self.recycle_database(&path, &reason)?;
                    let mut recycled = self.open_connection_once(&path)?;
                    self.ensure_transfer_schema(&mut recycled, &path)
                        .map_err(Self::schema_init_error_to_mega_error)?;
                    Ok(recycled)
                }
                Err(SqliteSchemaInitError::Fatal(err)) => Err(err),
            },
            Err(err) => Err(err),
        }
    }

    fn ensure_root_dir(&self) -> Result<()> {
        fs::create_dir_all(&self.root).map_err(|err| {
            MegaError::Custom(format!(
                "Failed to create persistence root {}: {}",
                self.root.display(),
                err
            ))
        })
    }

    fn open_connection_once(&self, path: &Path) -> Result<Connection> {
        Connection::open(path).map_err(|err| {
            MegaError::Custom(format!(
                "Failed to open persistence database {}: {}",
                path.display(),
                err
            ))
        })
    }

    fn ensure_engine_schema(
        &self,
        conn: &mut Connection,
        path: &Path,
    ) -> std::result::Result<(), SqliteSchemaInitError> {
        let tx = conn.transaction().map_err(|err| {
            SqliteSchemaInitError::Fatal(MegaError::Custom(format!(
                "Failed to start persistence schema transaction for {}: {}",
                path.display(),
                err
            )))
        })?;
        tx.execute_batch(
            "CREATE TABLE IF NOT EXISTS meta (
                key TEXT PRIMARY KEY,
                value TEXT NOT NULL
            );
            CREATE TABLE IF NOT EXISTS engine_state (
                slot INTEGER PRIMARY KEY CHECK(slot = 0),
                json TEXT NOT NULL
            );",
        )
        .map_err(|err| {
            SqliteSchemaInitError::Fatal(MegaError::Custom(format!(
                "Failed to initialize persistence schema for {}: {}",
                path.display(),
                err
            )))
        })?;

        let existing_schema_version: Option<String> = tx
            .query_row(
                "SELECT value FROM meta WHERE key = ?1",
                [SQLITE_META_SCHEMA_VERSION_KEY],
                |row| row.get(0),
            )
            .optional()
            .map_err(|err| {
                SqliteSchemaInitError::Fatal(MegaError::Custom(format!(
                    "Failed to read persistence schema version from {}: {}",
                    path.display(),
                    err
                )))
            })?;

        match existing_schema_version {
            Some(version) => {
                let parsed = version.parse::<u32>().map_err(|err| {
                    SqliteSchemaInitError::RecycleRequired(format!(
                        "Invalid persistence backend schema version in {}: {}",
                        path.display(),
                        err
                    ))
                })?;
                if parsed != SQLITE_PERSISTENCE_BACKEND_SCHEMA_VERSION {
                    return Err(SqliteSchemaInitError::RecycleRequired(format!(
                        "Unsupported persistence backend schema version: {}",
                        parsed
                    )));
                }
            }
            None => {
                tx.execute(
                    "INSERT INTO meta (key, value) VALUES (?1, ?2)",
                    params![
                        SQLITE_META_SCHEMA_VERSION_KEY,
                        SQLITE_PERSISTENCE_BACKEND_SCHEMA_VERSION.to_string()
                    ],
                )
                .map_err(|err| {
                    SqliteSchemaInitError::Fatal(MegaError::Custom(format!(
                        "Failed to write persistence schema version to {}: {}",
                        path.display(),
                        err
                    )))
                })?;
            }
        }

        tx.commit().map_err(|err| {
            SqliteSchemaInitError::Fatal(MegaError::Custom(format!(
                "Failed to commit persistence schema transaction for {}: {}",
                path.display(),
                err
            )))
        })?;

        Ok(())
    }

    fn ensure_transfer_schema(
        &self,
        conn: &mut Connection,
        path: &Path,
    ) -> std::result::Result<(), SqliteSchemaInitError> {
        let tx = conn.transaction().map_err(|err| {
            SqliteSchemaInitError::Fatal(MegaError::Custom(format!(
                "Failed to start transfer cache schema transaction for {}: {}",
                path.display(),
                err
            )))
        })?;
        tx.execute_batch(
            "CREATE TABLE IF NOT EXISTS meta (
                key TEXT PRIMARY KEY,
                value TEXT NOT NULL
            );
            CREATE TABLE IF NOT EXISTS transfer_checkpoint (
                kind TEXT NOT NULL,
                source_fingerprint TEXT NOT NULL,
                disambiguator TEXT NOT NULL,
                json TEXT NOT NULL,
                PRIMARY KEY(kind, source_fingerprint, disambiguator)
            );",
        )
        .map_err(|err| {
            SqliteSchemaInitError::Fatal(MegaError::Custom(format!(
                "Failed to initialize transfer cache schema for {}: {}",
                path.display(),
                err
            )))
        })?;

        let existing_schema_version: Option<String> = tx
            .query_row(
                "SELECT value FROM meta WHERE key = ?1",
                [SQLITE_TRANSFER_CACHE_META_SCHEMA_VERSION_KEY],
                |row| row.get(0),
            )
            .optional()
            .map_err(|err| {
                SqliteSchemaInitError::Fatal(MegaError::Custom(format!(
                    "Failed to read transfer cache schema version from {}: {}",
                    path.display(),
                    err
                )))
            })?;

        match existing_schema_version {
            Some(version) => {
                let parsed = version.parse::<u32>().map_err(|err| {
                    SqliteSchemaInitError::RecycleRequired(format!(
                        "Invalid transfer cache schema version in {}: {}",
                        path.display(),
                        err
                    ))
                })?;
                if parsed != SQLITE_TRANSFER_CACHE_SCHEMA_VERSION {
                    return Err(SqliteSchemaInitError::RecycleRequired(format!(
                        "Unsupported transfer cache schema version: {}",
                        parsed
                    )));
                }
            }
            None => {
                tx.execute(
                    "INSERT INTO meta (key, value) VALUES (?1, ?2)",
                    params![
                        SQLITE_TRANSFER_CACHE_META_SCHEMA_VERSION_KEY,
                        SQLITE_TRANSFER_CACHE_SCHEMA_VERSION.to_string()
                    ],
                )
                .map_err(|err| {
                    SqliteSchemaInitError::Fatal(MegaError::Custom(format!(
                        "Failed to write transfer cache schema version to {}: {}",
                        path.display(),
                        err
                    )))
                })?;
            }
        }

        tx.commit().map_err(|err| {
            SqliteSchemaInitError::Fatal(MegaError::Custom(format!(
                "Failed to commit transfer cache schema transaction for {}: {}",
                path.display(),
                err
            )))
        })?;

        Ok(())
    }

    fn transfer_kind_storage_value(kind: TransferCheckpointKind) -> &'static str {
        match kind {
            TransferCheckpointKind::Upload => "upload",
            TransferCheckpointKind::Download => "download",
        }
    }

    fn recycle_database(&self, path: &Path, reason: &str) -> Result<()> {
        if !path.exists() {
            return Ok(());
        }

        let timestamp = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map(|duration| duration.as_nanos())
            .unwrap_or_default();
        let recycled_path = path.with_extension(format!("sqlite3.recycled-{timestamp}"));

        fs::rename(path, &recycled_path).map_err(|err| {
            MegaError::Custom(format!(
                "Failed to recycle persistence database {} after {}: {}",
                path.display(),
                reason,
                err
            ))
        })
    }

    fn schema_init_error_to_mega_error(err: SqliteSchemaInitError) -> MegaError {
        match err {
            SqliteSchemaInitError::RecycleRequired(reason) => MegaError::Custom(reason),
            SqliteSchemaInitError::Fatal(err) => err,
        }
    }
}

#[cfg_attr(not(test), allow(dead_code))]
pub(crate) trait PersistenceBackend: Send + Sync {
    fn load_engine_state(&self, scope: &PersistenceScope) -> Result<Option<PersistedEngineState>>;
    fn save_engine_state(
        &self,
        scope: &PersistenceScope,
        state: &PersistedEngineState,
    ) -> Result<()>;
    fn clear_engine_state(&self, scope: &PersistenceScope) -> Result<()>;

    fn load_transfer_checkpoint(
        &self,
        scope: &PersistenceScope,
        key: &TransferCheckpointKey,
    ) -> Result<Option<TransferCheckpointRecord>>;
    fn save_transfer_checkpoint(
        &self,
        scope: &PersistenceScope,
        record: &TransferCheckpointRecord,
    ) -> Result<()>;
    fn clear_transfer_checkpoint(
        &self,
        scope: &PersistenceScope,
        key: &TransferCheckpointKey,
    ) -> Result<()>;
}

#[derive(Clone)]
pub(crate) struct PersistenceRuntime {
    backend: Arc<dyn PersistenceBackend>,
}

impl fmt::Debug for PersistenceRuntime {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("PersistenceRuntime").finish_non_exhaustive()
    }
}

#[cfg_attr(not(test), allow(dead_code))]
impl PersistenceRuntime {
    pub(crate) fn disabled() -> Self {
        Self::new(Arc::new(NoopPersistenceBackend))
    }

    pub(crate) fn production_default() -> Result<Self> {
        Ok(Self::production_at(default_persistence_root()?))
    }

    #[cfg_attr(not(test), allow(dead_code))]
    pub(crate) fn production_at(root: PathBuf) -> Self {
        Self::new(Arc::new(SqlitePersistenceBackend::new(root)))
    }

    pub(crate) fn new(backend: Arc<dyn PersistenceBackend>) -> Self {
        Self { backend }
    }

    pub(crate) fn load_engine_state(
        &self,
        scope: &PersistenceScope,
    ) -> Result<Option<PersistedEngineState>> {
        let state = self.backend.load_engine_state(scope)?;
        if let Some(state) = state {
            state.validate_schema_version()?;
            Ok(Some(state))
        } else {
            Ok(None)
        }
    }

    pub(crate) fn save_engine_state(
        &self,
        scope: &PersistenceScope,
        state: &PersistedEngineState,
    ) -> Result<()> {
        state.validate_schema_version()?;
        self.backend.save_engine_state(scope, state)
    }

    pub(crate) fn clear_engine_state(&self, scope: &PersistenceScope) -> Result<()> {
        self.backend.clear_engine_state(scope)
    }

    pub(crate) fn load_transfer_checkpoint(
        &self,
        scope: &PersistenceScope,
        key: &TransferCheckpointKey,
    ) -> Result<Option<TransferCheckpointRecord>> {
        self.backend.load_transfer_checkpoint(scope, key)
    }

    pub(crate) fn save_transfer_checkpoint(
        &self,
        scope: &PersistenceScope,
        record: &TransferCheckpointRecord,
    ) -> Result<()> {
        self.backend.save_transfer_checkpoint(scope, record)
    }

    pub(crate) fn clear_transfer_checkpoint(
        &self,
        scope: &PersistenceScope,
        key: &TransferCheckpointKey,
    ) -> Result<()> {
        self.backend.clear_transfer_checkpoint(scope, key)
    }
}

#[derive(Debug, Default)]
pub(crate) struct NoopPersistenceBackend;

impl PersistenceBackend for NoopPersistenceBackend {
    fn load_engine_state(&self, _scope: &PersistenceScope) -> Result<Option<PersistedEngineState>> {
        Ok(None)
    }

    fn save_engine_state(
        &self,
        _scope: &PersistenceScope,
        _state: &PersistedEngineState,
    ) -> Result<()> {
        Ok(())
    }

    fn clear_engine_state(&self, _scope: &PersistenceScope) -> Result<()> {
        Ok(())
    }

    fn load_transfer_checkpoint(
        &self,
        _scope: &PersistenceScope,
        _key: &TransferCheckpointKey,
    ) -> Result<Option<TransferCheckpointRecord>> {
        Ok(None)
    }

    fn save_transfer_checkpoint(
        &self,
        _scope: &PersistenceScope,
        _record: &TransferCheckpointRecord,
    ) -> Result<()> {
        Ok(())
    }

    fn clear_transfer_checkpoint(
        &self,
        _scope: &PersistenceScope,
        _key: &TransferCheckpointKey,
    ) -> Result<()> {
        Ok(())
    }
}

#[cfg_attr(not(test), allow(dead_code))]
#[derive(Debug, Default)]
pub(crate) struct MemoryPersistenceBackend {
    engine_states: Mutex<HashMap<PersistenceScope, PersistedEngineState>>,
    transfer_checkpoints:
        Mutex<HashMap<(PersistenceScope, TransferCheckpointKey), TransferCheckpointRecord>>,
}

#[cfg_attr(not(test), allow(dead_code))]
impl MemoryPersistenceBackend {
    pub(crate) fn new() -> Self {
        Self::default()
    }

    fn lock_engine_states(
        &self,
    ) -> Result<std::sync::MutexGuard<'_, HashMap<PersistenceScope, PersistedEngineState>>> {
        self.engine_states
            .lock()
            .map_err(|_| MegaError::Custom("engine state memory store is poisoned".to_string()))
    }

    fn lock_transfer_checkpoints(
        &self,
    ) -> Result<
        std::sync::MutexGuard<
            '_,
            HashMap<(PersistenceScope, TransferCheckpointKey), TransferCheckpointRecord>,
        >,
    > {
        self.transfer_checkpoints.lock().map_err(|_| {
            MegaError::Custom("transfer checkpoint memory store is poisoned".to_string())
        })
    }
}

impl PersistenceBackend for MemoryPersistenceBackend {
    fn load_engine_state(&self, scope: &PersistenceScope) -> Result<Option<PersistedEngineState>> {
        Ok(self.lock_engine_states()?.get(scope).cloned())
    }

    fn save_engine_state(
        &self,
        scope: &PersistenceScope,
        state: &PersistedEngineState,
    ) -> Result<()> {
        self.lock_engine_states()?
            .insert(scope.clone(), state.clone());
        Ok(())
    }

    fn clear_engine_state(&self, scope: &PersistenceScope) -> Result<()> {
        self.lock_engine_states()?.remove(scope);
        Ok(())
    }

    fn load_transfer_checkpoint(
        &self,
        scope: &PersistenceScope,
        key: &TransferCheckpointKey,
    ) -> Result<Option<TransferCheckpointRecord>> {
        Ok(self
            .lock_transfer_checkpoints()?
            .get(&(scope.clone(), key.clone()))
            .cloned())
    }

    fn save_transfer_checkpoint(
        &self,
        scope: &PersistenceScope,
        record: &TransferCheckpointRecord,
    ) -> Result<()> {
        self.lock_transfer_checkpoints()?
            .insert((scope.clone(), record.key.clone()), record.clone());
        Ok(())
    }

    fn clear_transfer_checkpoint(
        &self,
        scope: &PersistenceScope,
        key: &TransferCheckpointKey,
    ) -> Result<()> {
        self.lock_transfer_checkpoints()?
            .remove(&(scope.clone(), key.clone()));
        Ok(())
    }
}

impl PersistenceBackend for SqlitePersistenceBackend {
    fn load_engine_state(&self, scope: &PersistenceScope) -> Result<Option<PersistedEngineState>> {
        let conn = self.open_engine_connection(scope)?;
        let db_path = self.engine_db_path(scope);
        let json: Option<String> = conn
            .query_row(
                "SELECT json FROM engine_state WHERE slot = ?1",
                [SQLITE_ENGINE_STATE_SINGLETON_SLOT],
                |row| row.get(0),
            )
            .optional()
            .map_err(|err| {
                MegaError::Custom(format!(
                    "Failed to load engine state from {}: {}",
                    db_path.display(),
                    err
                ))
            })?;

        match json {
            Some(json) => match serde_json::from_str::<PersistedEngineState>(&json) {
                Ok(state) => Ok(Some(state)),
                Err(_) => {
                    conn.execute(
                        "DELETE FROM engine_state WHERE slot = ?1",
                        [SQLITE_ENGINE_STATE_SINGLETON_SLOT],
                    )
                    .map_err(|err| {
                        MegaError::Custom(format!(
                            "Failed to clear malformed engine state from {}: {}",
                            db_path.display(),
                            err
                        ))
                    })?;
                    Ok(None)
                }
            },
            None => Ok(None),
        }
    }

    fn save_engine_state(
        &self,
        scope: &PersistenceScope,
        state: &PersistedEngineState,
    ) -> Result<()> {
        let conn = self.open_engine_connection(scope)?;
        let json = serde_json::to_string(state)?;
        conn.execute(
            "INSERT INTO engine_state (slot, json) VALUES (?1, ?2)
             ON CONFLICT(slot) DO UPDATE SET json = excluded.json",
            params![SQLITE_ENGINE_STATE_SINGLETON_SLOT, json],
        )
        .map_err(|err| {
            MegaError::Custom(format!(
                "Failed to save engine state to {}: {}",
                self.engine_db_path(scope).display(),
                err
            ))
        })?;
        Ok(())
    }

    fn clear_engine_state(&self, scope: &PersistenceScope) -> Result<()> {
        let conn = self.open_engine_connection(scope)?;
        conn.execute(
            "DELETE FROM engine_state WHERE slot = ?1",
            [SQLITE_ENGINE_STATE_SINGLETON_SLOT],
        )
        .map_err(|err| {
            MegaError::Custom(format!(
                "Failed to clear engine state from {}: {}",
                self.engine_db_path(scope).display(),
                err
            ))
        })?;
        Ok(())
    }

    fn load_transfer_checkpoint(
        &self,
        scope: &PersistenceScope,
        key: &TransferCheckpointKey,
    ) -> Result<Option<TransferCheckpointRecord>> {
        let conn = self.open_transfer_connection(scope)?;
        let db_path = self.transfer_db_path(scope);
        let kind = Self::transfer_kind_storage_value(key.kind());
        let disambiguator = key.storage_disambiguator();
        let json: Option<String> = conn
            .query_row(
                "SELECT json FROM transfer_checkpoint
                 WHERE kind = ?1 AND source_fingerprint = ?2 AND disambiguator = ?3",
                params![kind, key.source_fingerprint(), disambiguator],
                |row| row.get(0),
            )
            .optional()
            .map_err(|err| {
                MegaError::Custom(format!(
                    "Failed to load transfer checkpoint from {}: {}",
                    db_path.display(),
                    err
                ))
            })?;

        match json {
            Some(json) => match serde_json::from_str::<TransferCheckpointRecord>(&json) {
                Ok(record) if record.matches_key(key) => Ok(Some(record)),
                Err(_) => {
                    conn.execute(
                        "DELETE FROM transfer_checkpoint
                         WHERE kind = ?1 AND source_fingerprint = ?2 AND disambiguator = ?3",
                        params![kind, key.source_fingerprint(), disambiguator],
                    )
                    .map_err(|err| {
                        MegaError::Custom(format!(
                            "Failed to clear malformed transfer checkpoint from {}: {}",
                            db_path.display(),
                            err
                        ))
                    })?;
                    Ok(None)
                }
                Ok(_) => {
                    conn.execute(
                        "DELETE FROM transfer_checkpoint
                         WHERE kind = ?1 AND source_fingerprint = ?2 AND disambiguator = ?3",
                        params![kind, key.source_fingerprint(), disambiguator],
                    )
                    .map_err(|err| {
                        MegaError::Custom(format!(
                            "Failed to clear mismatched transfer checkpoint from {}: {}",
                            db_path.display(),
                            err
                        ))
                    })?;
                    Ok(None)
                }
            },
            None => Ok(None),
        }
    }

    fn save_transfer_checkpoint(
        &self,
        scope: &PersistenceScope,
        record: &TransferCheckpointRecord,
    ) -> Result<()> {
        let conn = self.open_transfer_connection(scope)?;
        let json = serde_json::to_string(record)?;
        let kind = Self::transfer_kind_storage_value(record.key.kind());
        let disambiguator = record.key.storage_disambiguator();
        conn.execute(
            "INSERT INTO transfer_checkpoint (kind, source_fingerprint, disambiguator, json)
             VALUES (?1, ?2, ?3, ?4)
             ON CONFLICT(kind, source_fingerprint, disambiguator)
             DO UPDATE SET json = excluded.json",
            params![kind, record.key.source_fingerprint(), disambiguator, json],
        )
        .map_err(|err| {
            MegaError::Custom(format!(
                "Failed to save transfer checkpoint to {}: {}",
                self.transfer_db_path(scope).display(),
                err
            ))
        })?;
        Ok(())
    }

    fn clear_transfer_checkpoint(
        &self,
        scope: &PersistenceScope,
        key: &TransferCheckpointKey,
    ) -> Result<()> {
        let conn = self.open_transfer_connection(scope)?;
        let kind = Self::transfer_kind_storage_value(key.kind());
        let disambiguator = key.storage_disambiguator();
        conn.execute(
            "DELETE FROM transfer_checkpoint
             WHERE kind = ?1 AND source_fingerprint = ?2 AND disambiguator = ?3",
            params![kind, key.source_fingerprint(), disambiguator],
        )
        .map_err(|err| {
            MegaError::Custom(format!(
                "Failed to clear transfer checkpoint from {}: {}",
                self.transfer_db_path(scope).display(),
                err
            ))
        })?;
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::fs::Node;
    use std::path::{Path, PathBuf};
    use std::time::{SystemTime, UNIX_EPOCH};

    struct TestDir {
        path: PathBuf,
    }

    impl TestDir {
        fn new(label: &str) -> Self {
            let unique = SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .expect("system clock should be after unix epoch")
                .as_nanos();
            let path = std::env::temp_dir().join(format!(
                "megalib-persistence-{}-{}-{}",
                label,
                std::process::id(),
                unique
            ));
            Self { path }
        }

        fn path(&self) -> &Path {
            &self.path
        }
    }

    impl Drop for TestDir {
        fn drop(&mut self) {
            let _ = fs::remove_dir_all(&self.path);
        }
    }

    fn sample_scope() -> PersistenceScope {
        PersistenceScope::new("account-handle")
    }

    fn sample_engine_state() -> PersistedEngineState {
        PersistedEngineState {
            schema_version: ENGINE_STATE_SCHEMA_VERSION,
            sc: PersistedScState {
                scsn: Some("scsn-123".to_string()),
            },
            alerts: PersistedAlertsState {
                alerts_catchup_pending: true,
                user_alert_lsn: Some("lsn-123".to_string()),
                user_alerts: vec![serde_json::json!({"id": 1})],
            },
            tree: None,
        }
    }

    fn sample_upload_state() -> UploadState {
        UploadState::new(
            "https://example.invalid/upload".to_string(),
            [1u8; 16],
            [2u8; 8],
            1024,
            "example.bin".to_string(),
            "parent-handle".to_string(),
            "fingerprint".to_string(),
        )
    }

    fn sample_upload_checkpoint_record() -> TransferCheckpointRecord {
        TransferCheckpointRecord::from_upload_state(
            Path::new("/tmp/example.bin"),
            &sample_upload_state(),
        )
    }

    fn sample_download_checkpoint_record() -> TransferCheckpointRecord {
        TransferCheckpointRecord {
            key: TransferCheckpointKey::download("download-fingerprint", "node-handle"),
            committed_bytes: 2048,
            temp_urls: vec!["https://example.invalid/download".to_string()],
            payload: TransferCheckpointPayload::Download(PersistedDownloadCheckpoint {
                source_node_handle: "node-handle".to_string(),
                temp_path: PathBuf::from("/tmp/.megatmp.node-handle"),
                target_path: PathBuf::from("/tmp/example-download.bin"),
                file_size: 4096,
                temp_file_size: 2048,
            }),
        }
    }

    fn sample_node() -> Node {
        Node {
            name: "example.txt".to_string(),
            handle: "node-handle".to_string(),
            parent_handle: Some("parent-handle".to_string()),
            node_type: NodeType::File,
            size: 1024,
            timestamp: 12345,
            key: vec![1, 2, 3, 4],
            path: Some("/Root/example.txt".to_string()),
            link: Some("public-link".to_string()),
            file_attr: Some("924:1*thumb".to_string()),
            share_key: Some([7u8; 16]),
            share_handle: Some("share-root".to_string()),
            is_inshare: true,
            is_outshare: false,
            share_access: Some(1),
        }
    }

    fn sqlite_backend(label: &str) -> (TestDir, SqlitePersistenceBackend) {
        let dir = TestDir::new(label);
        let backend = SqlitePersistenceBackend::new(dir.path().to_path_buf());
        (dir, backend)
    }

    #[test]
    fn disabled_runtime_behaves_like_empty_store() {
        let runtime = PersistenceRuntime::disabled();
        let scope = sample_scope();
        let upload_record = sample_upload_checkpoint_record();

        assert!(
            runtime
                .load_engine_state(&scope)
                .expect("load engine state should succeed")
                .is_none()
        );
        assert!(
            runtime
                .load_transfer_checkpoint(&scope, &upload_record.key)
                .expect("load transfer checkpoint should succeed")
                .is_none()
        );
        runtime
            .save_engine_state(&scope, &sample_engine_state())
            .expect("save engine state should succeed");
        runtime
            .save_transfer_checkpoint(&scope, &upload_record)
            .expect("save transfer checkpoint should succeed");
        runtime
            .clear_engine_state(&scope)
            .expect("clear engine state should succeed");
        runtime
            .clear_transfer_checkpoint(&scope, &upload_record.key)
            .expect("clear transfer checkpoint should succeed");
    }

    #[test]
    fn memory_backend_round_trips_engine_and_transfer_checkpoints() {
        let runtime = PersistenceRuntime::new(Arc::new(MemoryPersistenceBackend::new()));
        let scope = sample_scope();
        let engine_state = sample_engine_state();
        let upload_record = sample_upload_checkpoint_record();
        let download_record = sample_download_checkpoint_record();

        runtime
            .save_engine_state(&scope, &engine_state)
            .expect("save engine state should succeed");
        runtime
            .save_transfer_checkpoint(&scope, &upload_record)
            .expect("save upload checkpoint should succeed");
        runtime
            .save_transfer_checkpoint(&scope, &download_record)
            .expect("save download checkpoint should succeed");

        let loaded_engine_state = runtime
            .load_engine_state(&scope)
            .expect("load engine state should succeed")
            .expect("engine state should exist");
        let loaded_upload_record = runtime
            .load_transfer_checkpoint(&scope, &upload_record.key)
            .expect("load upload checkpoint should succeed")
            .expect("upload checkpoint should exist");
        let loaded_download_record = runtime
            .load_transfer_checkpoint(&scope, &download_record.key)
            .expect("load download checkpoint should succeed")
            .expect("download checkpoint should exist");

        assert_eq!(loaded_engine_state.sc.scsn.as_deref(), Some("scsn-123"));
        assert_eq!(
            loaded_engine_state.alerts.user_alert_lsn.as_deref(),
            Some("lsn-123")
        );
        assert_eq!(loaded_engine_state.alerts.user_alerts.len(), 1);
        assert_eq!(loaded_upload_record, upload_record);
        assert_eq!(loaded_download_record, download_record);

        runtime
            .clear_engine_state(&scope)
            .expect("clear engine state should succeed");
        runtime
            .clear_transfer_checkpoint(&scope, &upload_record.key)
            .expect("clear upload checkpoint should succeed");
        runtime
            .clear_transfer_checkpoint(&scope, &download_record.key)
            .expect("clear download checkpoint should succeed");

        assert!(
            runtime
                .load_engine_state(&scope)
                .expect("load engine state after clear should succeed")
                .is_none()
        );
        assert!(
            runtime
                .load_transfer_checkpoint(&scope, &upload_record.key)
                .expect("load upload checkpoint after clear should succeed")
                .is_none()
        );
        assert!(
            runtime
                .load_transfer_checkpoint(&scope, &download_record.key)
                .expect("load download checkpoint after clear should succeed")
                .is_none()
        );
    }

    #[test]
    fn runtime_rejects_unsupported_schema_version() {
        let runtime = PersistenceRuntime::new(Arc::new(MemoryPersistenceBackend::new()));
        let scope = sample_scope();
        let mut engine_state = sample_engine_state();
        engine_state.schema_version = ENGINE_STATE_SCHEMA_VERSION + 1;

        let err = runtime
            .save_engine_state(&scope, &engine_state)
            .expect_err("save should reject unsupported schema version");

        assert!(
            err.to_string()
                .contains("Unsupported persistence schema version")
        );
    }

    #[test]
    fn memory_backend_scope_isolation_is_preserved() {
        let runtime = PersistenceRuntime::new(Arc::new(MemoryPersistenceBackend::new()));
        let scope_a = PersistenceScope::new("account-a");
        let scope_b = PersistenceScope::new("account-b");

        runtime
            .save_engine_state(&scope_a, &sample_engine_state())
            .expect("save for scope A should succeed");

        assert!(
            runtime
                .load_engine_state(&scope_b)
                .expect("load for scope B should succeed")
                .is_none()
        );
    }

    #[test]
    fn persisted_node_record_round_trips_runtime_fields_except_path() {
        let node = sample_node();

        let record = PersistedNodeRecord::from(&node);
        let restored = record.into_node();

        assert_eq!(restored.name, node.name);
        assert_eq!(restored.handle, node.handle);
        assert_eq!(restored.parent_handle, node.parent_handle);
        assert_eq!(restored.node_type, node.node_type);
        assert_eq!(restored.size, node.size);
        assert_eq!(restored.timestamp, node.timestamp);
        assert_eq!(restored.key, node.key);
        assert_eq!(restored.link, node.link);
        assert_eq!(restored.file_attr, node.file_attr);
        assert_eq!(restored.share_key, node.share_key);
        assert_eq!(restored.share_handle, node.share_handle);
        assert_eq!(restored.is_inshare, node.is_inshare);
        assert_eq!(restored.is_outshare, node.is_outshare);
        assert_eq!(restored.share_access, node.share_access);
        assert_eq!(restored.path, None);
    }

    #[test]
    fn sqlite_backend_round_trips_engine_and_transfer_checkpoints() {
        let (_dir, backend) = sqlite_backend("roundtrip");
        let scope = sample_scope();
        let engine_state = sample_engine_state();
        let upload_record = sample_upload_checkpoint_record();
        let download_record = sample_download_checkpoint_record();

        backend
            .save_engine_state(&scope, &engine_state)
            .expect("sqlite backend should save engine state");
        backend
            .save_transfer_checkpoint(&scope, &upload_record)
            .expect("sqlite backend should save upload checkpoint");
        backend
            .save_transfer_checkpoint(&scope, &download_record)
            .expect("sqlite backend should save download checkpoint");

        let reloaded = SqlitePersistenceBackend::new(backend.root.clone());

        let loaded_engine_state = reloaded
            .load_engine_state(&scope)
            .expect("sqlite backend should load engine state")
            .expect("engine state should exist");
        let loaded_upload_record = reloaded
            .load_transfer_checkpoint(&scope, &upload_record.key)
            .expect("sqlite backend should load upload checkpoint")
            .expect("upload checkpoint should exist");
        let loaded_download_record = reloaded
            .load_transfer_checkpoint(&scope, &download_record.key)
            .expect("sqlite backend should load download checkpoint")
            .expect("download checkpoint should exist");

        assert_eq!(loaded_engine_state.sc.scsn, engine_state.sc.scsn);
        assert_eq!(
            loaded_engine_state.alerts.user_alert_lsn,
            engine_state.alerts.user_alert_lsn
        );
        assert_eq!(loaded_upload_record, upload_record);
        assert_eq!(loaded_download_record, download_record);
    }

    #[test]
    fn sqlite_backend_uses_scope_qualified_db_paths() {
        let (_dir, backend) = sqlite_backend("scope-paths");
        let scope_a = PersistenceScope::new("account-a");
        let scope_b = PersistenceScope::new("account-b");

        let path_a = backend.engine_db_path(&scope_a);
        let path_b = backend.engine_db_path(&scope_b);
        let transfer_path_a = backend.transfer_db_path(&scope_a);
        let transfer_path_b = backend.transfer_db_path(&scope_b);

        assert_ne!(path_a, path_b);
        assert_ne!(transfer_path_a, transfer_path_b);
        assert!(path_a.starts_with(&backend.root));
        assert!(path_b.starts_with(&backend.root));
        assert!(transfer_path_a.starts_with(&backend.root));
        assert!(transfer_path_b.starts_with(&backend.root));
        assert_eq!(
            path_a.extension().and_then(|ext| ext.to_str()),
            Some("sqlite3")
        );
        assert_eq!(
            path_b.extension().and_then(|ext| ext.to_str()),
            Some("sqlite3")
        );
        assert_eq!(
            transfer_path_a.extension().and_then(|ext| ext.to_str()),
            Some("sqlite3")
        );
        assert_eq!(
            transfer_path_b.extension().and_then(|ext| ext.to_str()),
            Some("sqlite3")
        );
    }

    #[test]
    fn sqlite_backend_initializes_schema_on_first_write() {
        let (_dir, backend) = sqlite_backend("schema-init");
        let scope = sample_scope();
        let db_path = backend.engine_db_path(&scope);

        assert!(!db_path.exists());

        backend
            .save_engine_state(&scope, &sample_engine_state())
            .expect("sqlite backend should initialize schema during first save");

        assert!(db_path.exists());
    }

    #[test]
    fn sqlite_backend_recycles_unsupported_backend_schema_version() {
        let (_dir, backend) = sqlite_backend("schema-version");
        let scope = sample_scope();

        backend
            .save_engine_state(&scope, &sample_engine_state())
            .expect("initial save should succeed");

        let conn = Connection::open(backend.engine_db_path(&scope))
            .expect("test should be able to reopen sqlite database");
        conn.execute(
            "UPDATE meta SET value = ?1 WHERE key = ?2",
            params!["999", SQLITE_META_SCHEMA_VERSION_KEY],
        )
        .expect("test should be able to mutate schema version");

        let loaded = backend
            .load_engine_state(&scope)
            .expect("unsupported backend schema should recycle cleanly");

        assert!(loaded.is_none());
        assert!(backend.engine_db_path(&scope).exists());
    }

    #[test]
    fn sqlite_backend_treats_malformed_engine_state_as_cache_miss() {
        let (_dir, backend) = sqlite_backend("malformed-engine");
        let scope = sample_scope();

        backend
            .save_engine_state(&scope, &sample_engine_state())
            .expect("initial save should succeed");

        let conn = Connection::open(backend.engine_db_path(&scope))
            .expect("test should be able to reopen sqlite database");
        conn.execute(
            "UPDATE engine_state SET json = ?1 WHERE slot = ?2",
            params!["{not-json", SQLITE_ENGINE_STATE_SINGLETON_SLOT],
        )
        .expect("test should be able to corrupt engine state row");

        let loaded = backend
            .load_engine_state(&scope)
            .expect("malformed engine state should be treated as cache miss");

        assert!(loaded.is_none());
    }

    #[test]
    fn sqlite_backend_treats_malformed_transfer_checkpoint_as_cache_miss() {
        let (_dir, backend) = sqlite_backend("malformed-transfer");
        let scope = sample_scope();
        let upload_record = sample_upload_checkpoint_record();

        backend
            .save_transfer_checkpoint(&scope, &upload_record)
            .expect("initial save should succeed");

        let conn = Connection::open(backend.transfer_db_path(&scope))
            .expect("test should be able to reopen sqlite database");
        conn.execute(
            "UPDATE transfer_checkpoint SET json = ?1
             WHERE kind = ?2 AND source_fingerprint = ?3 AND disambiguator = ?4",
            params![
                "{not-json",
                SqlitePersistenceBackend::transfer_kind_storage_value(upload_record.key.kind()),
                upload_record.key.source_fingerprint(),
                upload_record.key.storage_disambiguator()
            ],
        )
        .expect("test should be able to corrupt transfer checkpoint row");

        let loaded = backend
            .load_transfer_checkpoint(&scope, &upload_record.key)
            .expect("malformed transfer checkpoint should be treated as cache miss");

        assert!(loaded.is_none());
    }

    #[test]
    fn sqlite_backend_treats_malformed_download_checkpoint_as_cache_miss() {
        let (_dir, backend) = sqlite_backend("malformed-download-transfer");
        let scope = sample_scope();
        let download_record = sample_download_checkpoint_record();

        backend
            .save_transfer_checkpoint(&scope, &download_record)
            .expect("initial save should succeed");

        let conn = Connection::open(backend.transfer_db_path(&scope))
            .expect("test should be able to reopen transfer cache database");
        conn.execute(
            "UPDATE transfer_checkpoint SET json = ?1
             WHERE kind = ?2 AND source_fingerprint = ?3 AND disambiguator = ?4",
            params![
                "{not-json",
                SqlitePersistenceBackend::transfer_kind_storage_value(download_record.key.kind()),
                download_record.key.source_fingerprint(),
                download_record.key.storage_disambiguator()
            ],
        )
        .expect("test should be able to corrupt download checkpoint row");

        let loaded = backend
            .load_transfer_checkpoint(&scope, &download_record.key)
            .expect("malformed download checkpoint should be treated as cache miss");

        assert!(loaded.is_none());
    }

    #[test]
    fn sqlite_backend_recycles_transfer_cache_without_touching_engine_state() {
        let (_dir, backend) = sqlite_backend("transfer-recycle");
        let scope = sample_scope();
        let upload_record = sample_upload_checkpoint_record();
        let engine_state = sample_engine_state();

        backend
            .save_engine_state(&scope, &engine_state)
            .expect("initial engine state save should succeed");
        backend
            .save_transfer_checkpoint(&scope, &upload_record)
            .expect("initial transfer checkpoint save should succeed");

        let conn = Connection::open(backend.transfer_db_path(&scope))
            .expect("test should be able to reopen transfer cache database");
        conn.execute(
            "UPDATE meta SET value = ?1 WHERE key = ?2",
            params!["999", SQLITE_TRANSFER_CACHE_META_SCHEMA_VERSION_KEY],
        )
        .expect("test should be able to mutate transfer cache schema version");

        let recycled_transfer = backend
            .load_transfer_checkpoint(&scope, &upload_record.key)
            .expect("transfer cache recycle should behave like an empty cache");
        let preserved_engine = backend
            .load_engine_state(&scope)
            .expect("engine state load should still succeed")
            .expect("engine state should survive transfer cache recycle");

        assert!(recycled_transfer.is_none());
        assert_eq!(preserved_engine.sc.scsn, engine_state.sc.scsn);
        assert!(backend.engine_db_path(&scope).exists());
        assert!(backend.transfer_db_path(&scope).exists());
    }

    #[test]
    fn production_runtime_uses_sqlite_backend_round_trip() {
        let dir = TestDir::new("production-runtime");
        let runtime = PersistenceRuntime::production_at(dir.path().to_path_buf());
        let scope = sample_scope();
        let state = sample_engine_state();

        runtime
            .save_engine_state(&scope, &state)
            .expect("production runtime should save engine state");

        let loaded = runtime
            .load_engine_state(&scope)
            .expect("production runtime should load engine state")
            .expect("engine state should exist");

        assert_eq!(loaded.sc.scsn, state.sc.scsn);
    }
}
