//! Sync state persistence with redb
//!
//! This module provides durable storage for Automerge sync state, enabling:
//! - Persist sync heads per peer
//! - Recovery on restart (resume from last position)
//! - Periodic checkpointing
//!
//! # Storage Schema
//!
//! Uses redb with key prefixes:
//! - `sync_state:{peer_id}:{doc_key}` → serialized automerge::sync::State
//! - `checkpoint:{timestamp}` → snapshot metadata
//! - `meta:last_checkpoint` → timestamp of last checkpoint

use anyhow::{Context, Result};
use automerge::sync::State as SyncState;
use iroh::EndpointId;
use redb::{Database, TableDefinition};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::path::Path;
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};

/// Table for sync state storage
const SYNC_STATE_TABLE: TableDefinition<&[u8], &[u8]> = TableDefinition::new("sync_states");

/// Table for checkpoint storage
const CHECKPOINT_TABLE: TableDefinition<&[u8], &[u8]> = TableDefinition::new("checkpoints");

/// Table for metadata
const META_TABLE: TableDefinition<&[u8], &[u8]> = TableDefinition::new("meta");

/// Key prefixes for storage
const SYNC_STATE_PREFIX: &str = "sync_state:";
const CHECKPOINT_PREFIX: &str = "checkpoint:";
const META_LAST_CHECKPOINT: &str = "last_checkpoint";

/// Serializable wrapper for automerge::sync::State
///
/// The Automerge SyncState isn't directly serializable, so we store
/// the encoded bytes along with metadata.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PersistedSyncState {
    /// Encoded sync state bytes
    pub state_bytes: Vec<u8>,
    /// Peer ID (hex encoded for serialization)
    pub peer_id_hex: String,
    /// Document key
    pub doc_key: String,
    /// Timestamp when state was saved
    pub saved_at: u64,
    /// Number of syncs completed
    pub sync_count: u64,
}

impl PersistedSyncState {
    /// Create from SyncState and metadata
    pub fn from_sync_state(
        state: &SyncState,
        peer_id: &EndpointId,
        doc_key: &str,
        sync_count: u64,
    ) -> Self {
        Self {
            state_bytes: state.encode(),
            peer_id_hex: hex::encode(peer_id.as_bytes()),
            doc_key: doc_key.to_string(),
            saved_at: SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .expect("system clock before UNIX epoch")
                .as_secs(),
            sync_count,
        }
    }

    /// Restore SyncState from persisted data
    pub fn to_sync_state(&self) -> Result<SyncState> {
        SyncState::decode(&self.state_bytes).context("Failed to decode sync state")
    }
}

/// Decode a raw stored value into its wrapper plus the live `SyncState`.
///
/// Returns `Err` on any *deserialization* failure: a stale on-disk format
/// (e.g. an `automerge::sync::State` blob encoded by an older automerge that
/// no longer decodes after an in-place upgrade — `automerge` 0.9 added fields
/// to `sync::State`), a corrupt value, or a changed wrapper schema.
///
/// Callers treat an `Err` here as "no usable persisted sync state" and degrade
/// to re-negotiating sync from scratch (slower first sync, no data loss) rather
/// than propagating a hard error. This is deliberate: a single undecodable
/// entry must not abort recovery of the rest of the store. redb I/O errors are
/// surfaced separately at the call site and remain fatal.
fn decode_stored_entry(bytes: &[u8]) -> Result<(PersistedSyncState, SyncState)> {
    let persisted: PersistedSyncState =
        serde_json::from_slice(bytes).context("deserialize persisted sync state wrapper")?;
    let state = persisted.to_sync_state()?;
    Ok((persisted, state))
}

/// Checkpoint metadata
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Checkpoint {
    /// Timestamp of checkpoint
    pub timestamp: u64,
    /// Number of sync states saved
    pub state_count: usize,
    /// Total bytes stored
    pub total_bytes: usize,
    /// Peer IDs included
    pub peer_ids: Vec<String>,
}

/// Statistics about persisted sync state
#[derive(Debug, Clone, Default)]
pub struct PersistenceStats {
    /// Number of sync states stored
    pub state_count: usize,
    /// Total bytes used
    pub total_bytes: usize,
    /// Number of peers with stored state
    pub peer_count: usize,
    /// Last checkpoint timestamp
    pub last_checkpoint: Option<u64>,
    /// Number of checkpoints
    pub checkpoint_count: usize,
}

/// Sync state persistence manager
///
/// Provides durable storage for Automerge sync state to enable
/// fast recovery after restart without full resync.
pub struct SyncStatePersistence {
    /// redb instance
    db: Arc<Database>,
    /// Checkpoint interval (how often to create checkpoints)
    checkpoint_interval: Duration,
}

impl SyncStatePersistence {
    /// Open or create sync state storage at the given path
    pub fn open(path: impl AsRef<Path>) -> Result<Self> {
        let path = path.as_ref();

        // Ensure parent directory exists
        if let Some(parent) = path.parent() {
            std::fs::create_dir_all(parent).ok();
        }

        // redb stores in a single file, append .redb extension if it's a directory path
        let db_path = if path.is_dir() || !path.exists() {
            std::fs::create_dir_all(path).ok();
            path.join("sync_state.redb")
        } else {
            path.to_path_buf()
        };

        let db = Database::create(&db_path).context("Failed to open sync state redb")?;

        // Initialize tables
        {
            let write_txn = db
                .begin_write()
                .context("Failed to begin write transaction")?;
            let _ = write_txn.open_table(SYNC_STATE_TABLE);
            let _ = write_txn.open_table(CHECKPOINT_TABLE);
            let _ = write_txn.open_table(META_TABLE);
            write_txn
                .commit()
                .context("Failed to commit table creation")?;
        }

        Ok(Self {
            db: Arc::new(db),
            checkpoint_interval: Duration::from_secs(60), // Default: checkpoint every 60 seconds
        })
    }

    /// Open with custom checkpoint interval
    pub fn open_with_interval(
        path: impl AsRef<Path>,
        checkpoint_interval: Duration,
    ) -> Result<Self> {
        let mut persistence = Self::open(path)?;
        persistence.checkpoint_interval = checkpoint_interval;
        Ok(persistence)
    }

    /// Build storage key for sync state
    fn sync_state_key(peer_id: &EndpointId, doc_key: &str) -> String {
        format!(
            "{}{}:{}",
            SYNC_STATE_PREFIX,
            hex::encode(peer_id.as_bytes()),
            doc_key
        )
    }

    /// Save sync state for a peer and document
    pub fn save_sync_state(
        &self,
        peer_id: &EndpointId,
        doc_key: &str,
        state: &SyncState,
        sync_count: u64,
    ) -> Result<()> {
        let key = Self::sync_state_key(peer_id, doc_key);
        let persisted = PersistedSyncState::from_sync_state(state, peer_id, doc_key, sync_count);

        let value = serde_json::to_vec(&persisted).context("Failed to serialize sync state")?;

        let write_txn = self
            .db
            .begin_write()
            .context("Failed to begin write transaction")?;
        {
            let mut table = write_txn
                .open_table(SYNC_STATE_TABLE)
                .context("Failed to open sync state table")?;
            table
                .insert(key.as_bytes(), value.as_slice())
                .context("Failed to write sync state")?;
        }
        write_txn.commit().context("Failed to commit write")?;

        tracing::trace!(
            "Saved sync state for peer {} doc {}: {} bytes",
            persisted.peer_id_hex,
            doc_key,
            value.len()
        );

        Ok(())
    }

    /// Load sync state for a peer and document
    pub fn load_sync_state(
        &self,
        peer_id: &EndpointId,
        doc_key: &str,
    ) -> Result<Option<(SyncState, u64)>> {
        let key = Self::sync_state_key(peer_id, doc_key);

        let read_txn = self
            .db
            .begin_read()
            .context("Failed to begin read transaction")?;
        let table = read_txn
            .open_table(SYNC_STATE_TABLE)
            .context("Failed to open sync state table")?;

        match table.get(key.as_bytes())? {
            Some(value) => match decode_stored_entry(value.value()) {
                Ok((persisted, state)) => {
                    tracing::trace!(
                        "Loaded sync state for peer {} doc {}: sync_count={}",
                        persisted.peer_id_hex,
                        doc_key,
                        persisted.sync_count
                    );
                    Ok(Some((state, persisted.sync_count)))
                }
                Err(e) => {
                    // Stale/undecodable persisted state (e.g. an automerge
                    // version upgrade changed the `sync::State` encoding).
                    // Degrade to "no state" so the peer re-negotiates sync
                    // from scratch rather than failing the load outright.
                    tracing::warn!(
                        doc_key,
                        error = %e,
                        "discarding undecodable persisted sync state; \
                         will re-negotiate sync from scratch"
                    );
                    Ok(None)
                }
            },
            None => Ok(None),
        }
    }

    /// Delete sync state for a peer and document
    pub fn delete_sync_state(&self, peer_id: &EndpointId, doc_key: &str) -> Result<()> {
        let key = Self::sync_state_key(peer_id, doc_key);

        let write_txn = self
            .db
            .begin_write()
            .context("Failed to begin write transaction")?;
        {
            let mut table = write_txn
                .open_table(SYNC_STATE_TABLE)
                .context("Failed to open sync state table")?;
            table.remove(key.as_bytes())?;
        }
        write_txn.commit().context("Failed to commit delete")?;

        Ok(())
    }

    /// Load all sync states for a peer
    pub fn load_all_for_peer(&self, peer_id: &EndpointId) -> Result<HashMap<String, SyncState>> {
        let prefix = format!("{}{}:", SYNC_STATE_PREFIX, hex::encode(peer_id.as_bytes()));
        let mut results = HashMap::new();

        let read_txn = self
            .db
            .begin_read()
            .context("Failed to begin read transaction")?;
        let table = read_txn
            .open_table(SYNC_STATE_TABLE)
            .context("Failed to open sync state table")?;

        for entry in table.range(prefix.as_bytes()..)? {
            let (key, value) = entry?;
            let key_bytes = key.value();
            let key_str = String::from_utf8_lossy(key_bytes);

            if !key_str.starts_with(&prefix) {
                break;
            }

            let (persisted, state) = match decode_stored_entry(value.value()) {
                Ok(decoded) => decoded,
                Err(e) => {
                    tracing::warn!(
                        key = %key_str,
                        error = %e,
                        "skipping undecodable persisted sync state during peer recovery"
                    );
                    continue;
                }
            };
            results.insert(persisted.doc_key.clone(), state);
        }

        Ok(results)
    }

    /// Load all sync states (for full recovery)
    pub fn load_all(&self) -> Result<HashMap<(EndpointId, String), SyncState>> {
        let mut results = HashMap::new();

        let read_txn = self
            .db
            .begin_read()
            .context("Failed to begin read transaction")?;
        let table = read_txn
            .open_table(SYNC_STATE_TABLE)
            .context("Failed to open sync state table")?;

        for entry in table.range(SYNC_STATE_PREFIX.as_bytes()..)? {
            let (key, value) = entry?;
            let key_bytes = key.value();
            let key_str = String::from_utf8_lossy(key_bytes);

            if !key_str.starts_with(SYNC_STATE_PREFIX) {
                break;
            }

            let (persisted, state) = match decode_stored_entry(value.value()) {
                Ok(decoded) => decoded,
                Err(e) => {
                    tracing::warn!(
                        key = %key_str,
                        error = %e,
                        "skipping undecodable persisted sync state during full recovery"
                    );
                    continue;
                }
            };

            // Parse peer ID from hex; skip (don't abort full recovery) on a
            // malformed key so one bad entry can't sink every other peer's state.
            let peer_id_bytes = match hex::decode(&persisted.peer_id_hex) {
                Ok(b) if b.len() == 32 => b,
                _ => {
                    tracing::warn!(
                        peer_id_hex = %persisted.peer_id_hex,
                        "skipping persisted sync state with invalid peer id"
                    );
                    continue;
                }
            };
            let mut arr = [0u8; 32];
            arr.copy_from_slice(&peer_id_bytes);
            let public_key = match iroh::PublicKey::from_bytes(&arr) {
                Ok(pk) => pk,
                Err(e) => {
                    tracing::warn!(
                        error = %e,
                        "skipping persisted sync state with invalid public key"
                    );
                    continue;
                }
            };
            let peer_id: EndpointId = public_key;

            results.insert((peer_id, persisted.doc_key.clone()), state);
        }

        tracing::info!("Loaded {} sync states from persistence", results.len());

        Ok(results)
    }

    /// Create a checkpoint
    pub fn create_checkpoint(&self) -> Result<Checkpoint> {
        let timestamp = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .expect("system clock before UNIX epoch")
            .as_millis() as u64;

        // Count current states
        let mut state_count = 0;
        let mut total_bytes = 0;
        let mut peer_ids = std::collections::HashSet::new();

        {
            let read_txn = self
                .db
                .begin_read()
                .context("Failed to begin read transaction")?;
            let table = read_txn
                .open_table(SYNC_STATE_TABLE)
                .context("Failed to open sync state table")?;

            for entry in table.range(SYNC_STATE_PREFIX.as_bytes()..)? {
                let (key, value) = entry?;
                let key_bytes = key.value();
                let key_str = String::from_utf8_lossy(key_bytes);

                if !key_str.starts_with(SYNC_STATE_PREFIX) {
                    break;
                }

                state_count += 1;
                total_bytes += value.value().len();

                // Extract peer ID from key
                if let Some(rest) = key_str.strip_prefix(SYNC_STATE_PREFIX) {
                    if let Some(peer_id) = rest.split(':').next() {
                        peer_ids.insert(peer_id.to_string());
                    }
                }
            }
        }

        let checkpoint = Checkpoint {
            timestamp,
            state_count,
            total_bytes,
            peer_ids: peer_ids.into_iter().collect(),
        };

        // Save checkpoint
        let checkpoint_key = format!("{}{}", CHECKPOINT_PREFIX, timestamp);
        let checkpoint_bytes = serde_json::to_vec(&checkpoint)?;

        let write_txn = self
            .db
            .begin_write()
            .context("Failed to begin write transaction")?;
        {
            let mut table = write_txn
                .open_table(CHECKPOINT_TABLE)
                .context("Failed to open checkpoint table")?;
            table.insert(checkpoint_key.as_bytes(), checkpoint_bytes.as_slice())?;
        }
        {
            // Update last checkpoint timestamp
            let mut meta_table = write_txn
                .open_table(META_TABLE)
                .context("Failed to open meta table")?;
            meta_table.insert(
                META_LAST_CHECKPOINT.as_bytes(),
                &timestamp.to_be_bytes()[..],
            )?;
        }
        write_txn.commit().context("Failed to commit checkpoint")?;

        tracing::info!(
            "Created checkpoint: {} states, {} bytes, {} peers",
            state_count,
            total_bytes,
            checkpoint.peer_ids.len()
        );

        Ok(checkpoint)
    }

    /// Get the last checkpoint
    pub fn get_last_checkpoint(&self) -> Result<Option<Checkpoint>> {
        let read_txn = self
            .db
            .begin_read()
            .context("Failed to begin read transaction")?;

        // Get last checkpoint timestamp
        let meta_table = read_txn
            .open_table(META_TABLE)
            .context("Failed to open meta table")?;

        let timestamp_bytes = match meta_table.get(META_LAST_CHECKPOINT.as_bytes())? {
            Some(value) => value.value().to_vec(),
            None => return Ok(None),
        };

        if timestamp_bytes.len() != 8 {
            return Ok(None);
        }

        let mut arr = [0u8; 8];
        arr.copy_from_slice(&timestamp_bytes);
        let timestamp = u64::from_be_bytes(arr);

        // Load checkpoint
        let checkpoint_key = format!("{}{}", CHECKPOINT_PREFIX, timestamp);
        let checkpoint_table = read_txn
            .open_table(CHECKPOINT_TABLE)
            .context("Failed to open checkpoint table")?;

        match checkpoint_table.get(checkpoint_key.as_bytes())? {
            Some(value) => {
                let checkpoint: Checkpoint = serde_json::from_slice(value.value())?;
                Ok(Some(checkpoint))
            }
            None => Ok(None),
        }
    }

    /// Get persistence statistics
    pub fn stats(&self) -> Result<PersistenceStats> {
        let mut stats = PersistenceStats::default();
        let mut peer_ids = std::collections::HashSet::new();

        let read_txn = self
            .db
            .begin_read()
            .context("Failed to begin read transaction")?;

        // Count sync states
        {
            let table = read_txn
                .open_table(SYNC_STATE_TABLE)
                .context("Failed to open sync state table")?;

            for entry in table.range(SYNC_STATE_PREFIX.as_bytes()..)? {
                let (key, value) = entry?;
                let key_bytes = key.value();
                let key_str = String::from_utf8_lossy(key_bytes);

                if !key_str.starts_with(SYNC_STATE_PREFIX) {
                    break;
                }

                stats.state_count += 1;
                stats.total_bytes += value.value().len();

                if let Some(rest) = key_str.strip_prefix(SYNC_STATE_PREFIX) {
                    if let Some(peer_id) = rest.split(':').next() {
                        peer_ids.insert(peer_id.to_string());
                    }
                }
            }
        }

        stats.peer_count = peer_ids.len();

        // Count checkpoints
        {
            let checkpoint_table = read_txn
                .open_table(CHECKPOINT_TABLE)
                .context("Failed to open checkpoint table")?;

            for entry in checkpoint_table.range(CHECKPOINT_PREFIX.as_bytes()..)? {
                let (key, _) = entry?;
                if !key.value().starts_with(CHECKPOINT_PREFIX.as_bytes()) {
                    break;
                }
                stats.checkpoint_count += 1;
            }
        }

        // Get last checkpoint timestamp
        if let Ok(Some(checkpoint)) = self.get_last_checkpoint() {
            stats.last_checkpoint = Some(checkpoint.timestamp);
        }

        Ok(stats)
    }

    /// Clean up old checkpoints, keeping only the last N
    pub fn cleanup_old_checkpoints(&self, keep_count: usize) -> Result<usize> {
        let mut checkpoints: Vec<u64> = Vec::new();

        {
            let read_txn = self
                .db
                .begin_read()
                .context("Failed to begin read transaction")?;
            let table = read_txn
                .open_table(CHECKPOINT_TABLE)
                .context("Failed to open checkpoint table")?;

            for entry in table.range(CHECKPOINT_PREFIX.as_bytes()..)? {
                let (key, _) = entry?;
                let key_bytes = key.value();
                let key_str = String::from_utf8_lossy(key_bytes);

                if !key_str.starts_with(CHECKPOINT_PREFIX) {
                    break;
                }

                if let Some(ts_str) = key_str.strip_prefix(CHECKPOINT_PREFIX) {
                    if let Ok(ts) = ts_str.parse::<u64>() {
                        checkpoints.push(ts);
                    }
                }
            }
        }

        // Sort descending (newest first)
        checkpoints.sort_by(|a, b| b.cmp(a));

        // Delete old ones
        let mut deleted = 0;
        let to_delete: Vec<_> = checkpoints.iter().skip(keep_count).cloned().collect();

        if !to_delete.is_empty() {
            let write_txn = self
                .db
                .begin_write()
                .context("Failed to begin write transaction")?;
            {
                let mut table = write_txn
                    .open_table(CHECKPOINT_TABLE)
                    .context("Failed to open checkpoint table")?;

                for ts in to_delete {
                    let key = format!("{}{}", CHECKPOINT_PREFIX, ts);
                    table.remove(key.as_bytes())?;
                    deleted += 1;
                }
            }
            write_txn.commit().context("Failed to commit cleanup")?;
        }

        if deleted > 0 {
            tracing::info!("Cleaned up {} old checkpoints", deleted);
        }

        Ok(deleted)
    }

    /// Delete all sync state for a peer (when peer is removed from mesh)
    pub fn delete_peer(&self, peer_id: &EndpointId) -> Result<usize> {
        let prefix = format!("{}{}:", SYNC_STATE_PREFIX, hex::encode(peer_id.as_bytes()));
        let mut keys_to_delete = Vec::new();

        // First, collect keys to delete
        {
            let read_txn = self
                .db
                .begin_read()
                .context("Failed to begin read transaction")?;
            let table = read_txn
                .open_table(SYNC_STATE_TABLE)
                .context("Failed to open sync state table")?;

            for entry in table.range(prefix.as_bytes()..)? {
                let (key, _) = entry?;
                let key_bytes = key.value();
                if !key_bytes.starts_with(prefix.as_bytes()) {
                    break;
                }
                keys_to_delete.push(key_bytes.to_vec());
            }
        }

        // Then delete them
        let deleted = keys_to_delete.len();
        if !keys_to_delete.is_empty() {
            let write_txn = self
                .db
                .begin_write()
                .context("Failed to begin write transaction")?;
            {
                let mut table = write_txn
                    .open_table(SYNC_STATE_TABLE)
                    .context("Failed to open sync state table")?;

                for key in keys_to_delete {
                    table.remove(key.as_slice())?;
                }
            }
            write_txn.commit().context("Failed to commit delete")?;
        }

        if deleted > 0 {
            tracing::info!(
                "Deleted {} sync states for peer {}",
                deleted,
                hex::encode(peer_id.as_bytes())
            );
        }

        Ok(deleted)
    }

    /// Get checkpoint interval
    pub fn checkpoint_interval(&self) -> Duration {
        self.checkpoint_interval
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use tempfile::TempDir;

    fn create_test_persistence() -> (SyncStatePersistence, TempDir) {
        let temp_dir = TempDir::new().unwrap();
        let persistence = SyncStatePersistence::open(temp_dir.path()).unwrap();
        (persistence, temp_dir)
    }

    fn create_test_peer_id() -> EndpointId {
        use iroh::SecretKey;
        SecretKey::generate().public()
    }

    #[test]
    fn test_save_and_load_sync_state() {
        let (persistence, _temp) = create_test_persistence();
        let peer_id = create_test_peer_id();
        let state = SyncState::new();

        // Save
        persistence
            .save_sync_state(&peer_id, "doc1", &state, 5)
            .unwrap();

        // Load
        let (loaded_state, sync_count) = persistence
            .load_sync_state(&peer_id, "doc1")
            .unwrap()
            .expect("State should exist");

        assert_eq!(sync_count, 5);
        // Sync states should be functionally equivalent (both empty/initial)
        assert_eq!(loaded_state.encode(), state.encode());
    }

    #[test]
    fn test_load_nonexistent_state() {
        let (persistence, _temp) = create_test_persistence();
        let peer_id = create_test_peer_id();

        let result = persistence
            .load_sync_state(&peer_id, "nonexistent")
            .unwrap();
        assert!(result.is_none());
    }

    #[test]
    fn test_delete_sync_state() {
        let (persistence, _temp) = create_test_persistence();
        let peer_id = create_test_peer_id();
        let state = SyncState::new();

        persistence
            .save_sync_state(&peer_id, "doc1", &state, 1)
            .unwrap();
        assert!(persistence
            .load_sync_state(&peer_id, "doc1")
            .unwrap()
            .is_some());

        persistence.delete_sync_state(&peer_id, "doc1").unwrap();
        assert!(persistence
            .load_sync_state(&peer_id, "doc1")
            .unwrap()
            .is_none());
    }

    #[test]
    fn test_load_all_for_peer() {
        let (persistence, _temp) = create_test_persistence();
        let peer_id = create_test_peer_id();
        let peer_id2 = create_test_peer_id();
        let state = SyncState::new();

        // Save states for peer 1
        persistence
            .save_sync_state(&peer_id, "doc1", &state, 1)
            .unwrap();
        persistence
            .save_sync_state(&peer_id, "doc2", &state, 2)
            .unwrap();

        // Save state for peer 2
        persistence
            .save_sync_state(&peer_id2, "doc1", &state, 3)
            .unwrap();

        // Load all for peer 1
        let peer1_states = persistence.load_all_for_peer(&peer_id).unwrap();
        assert_eq!(peer1_states.len(), 2);
        assert!(peer1_states.contains_key("doc1"));
        assert!(peer1_states.contains_key("doc2"));

        // Load all for peer 2
        let peer2_states = persistence.load_all_for_peer(&peer_id2).unwrap();
        assert_eq!(peer2_states.len(), 1);
    }

    #[test]
    fn test_load_all() {
        let (persistence, _temp) = create_test_persistence();
        let peer_id1 = create_test_peer_id();
        let peer_id2 = create_test_peer_id();
        let state = SyncState::new();

        persistence
            .save_sync_state(&peer_id1, "doc1", &state, 1)
            .unwrap();
        persistence
            .save_sync_state(&peer_id2, "doc2", &state, 2)
            .unwrap();

        let all_states = persistence.load_all().unwrap();
        assert_eq!(all_states.len(), 2);
    }

    /// Inject a raw value directly under a sync-state key, bypassing the
    /// `save_sync_state` encoding path — used to simulate a stale/undecodable
    /// persisted entry (e.g. one written by an older automerge whose
    /// `sync::State` encoding no longer decodes after an in-place upgrade).
    fn inject_raw_sync_state(persistence: &SyncStatePersistence, key: &str, bytes: &[u8]) {
        let write_txn = persistence.db.begin_write().unwrap();
        {
            let mut table = write_txn.open_table(SYNC_STATE_TABLE).unwrap();
            table.insert(key.as_bytes(), bytes).unwrap();
        }
        write_txn.commit().unwrap();
    }

    /// A persisted `sync::State` blob that no longer decodes (cross-version
    /// format drift) must degrade to `Ok(None)` — the peer re-negotiates sync
    /// from scratch — not bubble up a hard error from `load_sync_state`.
    #[test]
    fn load_sync_state_discards_undecodable_state_as_none() {
        let (persistence, _temp) = create_test_persistence();
        let peer_id = create_test_peer_id();

        // Valid wrapper, but state_bytes are not a decodable SyncState.
        let bad = PersistedSyncState {
            state_bytes: vec![0xFF, 0x00, 0xDE, 0xAD, 0xBE, 0xEF],
            peer_id_hex: hex::encode(peer_id.as_bytes()),
            doc_key: "doc1".to_string(),
            saved_at: 0,
            sync_count: 7,
        };
        let key = SyncStatePersistence::sync_state_key(&peer_id, "doc1");
        inject_raw_sync_state(&persistence, &key, &serde_json::to_vec(&bad).unwrap());

        // Tolerant: no error, treated as no persisted state.
        let result = persistence.load_sync_state(&peer_id, "doc1").unwrap();
        assert!(
            result.is_none(),
            "undecodable persisted state must load as None, not error"
        );
    }

    /// A corrupt wrapper (not even valid JSON for `PersistedSyncState`) is
    /// likewise tolerated by `load_sync_state`.
    #[test]
    fn load_sync_state_discards_corrupt_wrapper_as_none() {
        let (persistence, _temp) = create_test_persistence();
        let peer_id = create_test_peer_id();

        let key = SyncStatePersistence::sync_state_key(&peer_id, "doc1");
        inject_raw_sync_state(&persistence, &key, b"this is not json");

        let result = persistence.load_sync_state(&peer_id, "doc1").unwrap();
        assert!(result.is_none());
    }

    /// One undecodable entry must not abort full recovery: `load_all` skips the
    /// bad entry and still returns every good one. This is the real severity of
    /// the pre-hardening behavior — a single stale blob (e.g. after an automerge
    /// upgrade) would otherwise sink every peer's persisted sync state on restart.
    #[test]
    fn load_all_skips_undecodable_entry_and_keeps_good_ones() {
        let (persistence, _temp) = create_test_persistence();
        let good_peer = create_test_peer_id();
        let bad_peer = create_test_peer_id();
        let state = SyncState::new();

        // One good, persisted normally.
        persistence
            .save_sync_state(&good_peer, "good-doc", &state, 1)
            .unwrap();

        // One bad: valid key + wrapper, undecodable state bytes.
        let bad = PersistedSyncState {
            state_bytes: vec![0x00, 0x01, 0x02, 0x03],
            peer_id_hex: hex::encode(bad_peer.as_bytes()),
            doc_key: "bad-doc".to_string(),
            saved_at: 0,
            sync_count: 0,
        };
        let bad_key = SyncStatePersistence::sync_state_key(&bad_peer, "bad-doc");
        inject_raw_sync_state(&persistence, &bad_key, &serde_json::to_vec(&bad).unwrap());

        let all = persistence.load_all().unwrap();
        assert_eq!(all.len(), 1, "bad entry skipped, good entry retained");
        assert!(all.contains_key(&(good_peer, "good-doc".to_string())));
    }

    #[test]
    fn test_checkpoint() {
        let (persistence, _temp) = create_test_persistence();
        let peer_id = create_test_peer_id();
        let state = SyncState::new();

        // Save some states
        persistence
            .save_sync_state(&peer_id, "doc1", &state, 1)
            .unwrap();
        persistence
            .save_sync_state(&peer_id, "doc2", &state, 2)
            .unwrap();

        // Create checkpoint
        let checkpoint = persistence.create_checkpoint().unwrap();
        assert_eq!(checkpoint.state_count, 2);
        assert_eq!(checkpoint.peer_ids.len(), 1);

        // Get last checkpoint
        let loaded = persistence.get_last_checkpoint().unwrap().unwrap();
        assert_eq!(loaded.timestamp, checkpoint.timestamp);
        assert_eq!(loaded.state_count, 2);
    }

    #[test]
    fn test_stats() {
        let (persistence, _temp) = create_test_persistence();
        let peer_id1 = create_test_peer_id();
        let peer_id2 = create_test_peer_id();
        let state = SyncState::new();

        persistence
            .save_sync_state(&peer_id1, "doc1", &state, 1)
            .unwrap();
        persistence
            .save_sync_state(&peer_id2, "doc2", &state, 2)
            .unwrap();

        let stats = persistence.stats().unwrap();
        assert_eq!(stats.state_count, 2);
        assert_eq!(stats.peer_count, 2);
        assert!(stats.total_bytes > 0);
    }

    #[test]
    fn test_cleanup_old_checkpoints() {
        let (persistence, _temp) = create_test_persistence();
        let peer_id = create_test_peer_id();
        let state = SyncState::new();

        persistence
            .save_sync_state(&peer_id, "doc1", &state, 1)
            .unwrap();

        // Create multiple checkpoints
        for _ in 0..5 {
            persistence.create_checkpoint().unwrap();
            std::thread::sleep(std::time::Duration::from_millis(10));
        }

        let stats_before = persistence.stats().unwrap();
        assert_eq!(stats_before.checkpoint_count, 5);

        // Keep only 2
        let deleted = persistence.cleanup_old_checkpoints(2).unwrap();
        assert_eq!(deleted, 3);

        let stats_after = persistence.stats().unwrap();
        assert_eq!(stats_after.checkpoint_count, 2);
    }

    #[test]
    fn test_delete_peer() {
        let (persistence, _temp) = create_test_persistence();
        let peer_id1 = create_test_peer_id();
        let peer_id2 = create_test_peer_id();
        let state = SyncState::new();

        // Save states for both peers
        persistence
            .save_sync_state(&peer_id1, "doc1", &state, 1)
            .unwrap();
        persistence
            .save_sync_state(&peer_id1, "doc2", &state, 2)
            .unwrap();
        persistence
            .save_sync_state(&peer_id2, "doc1", &state, 3)
            .unwrap();

        // Delete peer 1
        let deleted = persistence.delete_peer(&peer_id1).unwrap();
        assert_eq!(deleted, 2);

        // Verify peer 1 states are gone
        assert!(persistence
            .load_sync_state(&peer_id1, "doc1")
            .unwrap()
            .is_none());
        assert!(persistence
            .load_sync_state(&peer_id1, "doc2")
            .unwrap()
            .is_none());

        // Verify peer 2 state remains
        assert!(persistence
            .load_sync_state(&peer_id2, "doc1")
            .unwrap()
            .is_some());
    }

    #[test]
    fn test_persisted_sync_state_roundtrip() {
        let peer_id = create_test_peer_id();
        let state = SyncState::new();

        let persisted = PersistedSyncState::from_sync_state(&state, &peer_id, "test_doc", 42);

        assert_eq!(persisted.doc_key, "test_doc");
        assert_eq!(persisted.sync_count, 42);
        assert!(!persisted.state_bytes.is_empty());

        let restored = persisted.to_sync_state().unwrap();
        assert_eq!(restored.encode(), state.encode());
    }
}
