//! P2P synchronization bridge between `chronogrid_net` and `chronogrid_core`.
//!
//! This module is only compiled when the `p2p` feature is enabled.
//!
//! ## Responsibilities
//!
//! [`SyncEngine`] bridges the [`SwarmHandle`] event stream from `chronogrid_net`
//! to the local [`Database`] CRDT merge pipeline:
//!
//! * **Inbound** (`NetEvent::DeltasReceived`): deserializes Automerge changesets
//!   from each [`DeltaPacket`], merges them into the local `crdt_docs` table,
//!   and updates the SQL entity rows to reflect the merged state.
//!
//! * **Outbound** (local CRDT mutation): computes the delta between the current
//!   Automerge document state and the last-synced state, packages it as a
//!   [`DeltaPacket`], and broadcasts it via [`SwarmHandle::broadcast_event`] /
//!   [`SwarmHandle::trigger_sync`].
//!
//! ## Architecture
//!
//! ```text
//! ┌──────────────────────────────────────────────────────────────┐
//! │                       SyncEngine                             │
//! │                                                              │
//! │  SwarmHandle ──► NetEvent::DeltasReceived                    │
//! │                         │                                    │
//! │                         ▼                                    │
//! │              merge_remote_deltas()                           │
//! │                         │                                    │
//! │          ┌──────────────┼──────────────┐                     │
//! │          ▼              ▼              ▼                     │
//! │     TaskCrdt       NoteCrdt      (future types)              │
//! │          │              │                                     │
//! │          ▼              ▼                                     │
//! │     Database::save_crdt_doc()  (upsert merged bytes)         │
//! │                                                              │
//! │  Local mutation ──► compute_outbound_delta()                 │
//! │                         │                                    │
//! │                         ▼                                    │
//! │              SwarmHandle::trigger_sync()                     │
//! └──────────────────────────────────────────────────────────────┘
//! ```
//!
//! ## Thread Safety
//!
//! [`SyncEngine`] is `Send + Sync`.  The [`Database`] is accessed via a
//! `Mutex<Database>` so that the async event-loop task and the calling thread
//! can both access it safely.

use std::sync::{Arc, Mutex};
use std::time::{SystemTime, UNIX_EPOCH};

use chronogrid_net::{
    DeltaPacket, NetEvent, SwarmHandle,
    sync::VectorClock,
};
use tokio::sync::broadcast;

use crate::{
    crdt::{Hlc, NoteCrdt, TaskCrdt},
    db::Database,
    error::{CoreError, Result},
};

// ---------------------------------------------------------------------------
// SyncEngine
// ---------------------------------------------------------------------------

/// Bridges the P2P network layer to the local CRDT database.
///
/// Holds a shared reference to the [`Database`] and a [`SwarmHandle`] for
/// sending outbound sync requests.  The event-loop task (started via
/// [`SyncEngine::run`]) listens for inbound [`NetEvent`]s and merges received
/// deltas into the local database.
pub struct SyncEngine {
    /// Shared, mutex-protected database connection.
    db: Arc<Mutex<Database>>,
    /// Handle to the running P2P swarm.
    swarm: SwarmHandle,
    /// Local vector clock — tracks the HLC head for each known entity.
    local_clock: VectorClock,
}

impl SyncEngine {
    /// Create a new `SyncEngine`.
    ///
    /// # Parameters
    ///
    /// - `db` — Shared database connection wrapped in `Arc<Mutex<_>>`.
    /// - `swarm` — Handle to the running [`SwarmManager`] event loop.
    pub fn new(db: Arc<Mutex<Database>>, swarm: SwarmHandle) -> Self {
        SyncEngine {
            db,
            swarm,
            local_clock: VectorClock::new(),
        }
    }

    /// Return a clone of the current local vector clock.
    pub fn local_clock(&self) -> VectorClock {
        self.local_clock.clone()
    }

    /// Update the local vector clock for a given entity.
    ///
    /// Call this after every local CRDT mutation so the clock stays current.
    pub fn advance_clock(
        &mut self,
        entity_id: impl Into<String>,
        hlc_timestamp: impl Into<String>,
    ) {
        self.local_clock.update(entity_id, hlc_timestamp);
    }

    // ── Inbound: merge remote deltas ─────────────────────────────────────────

    /// Merge a batch of remote [`DeltaPacket`]s into the local database.
    ///
    /// For each packet:
    /// 1. Load the existing Automerge document from `crdt_docs` (or create a
    ///    new empty document if none exists yet).
    /// 2. Merge the remote changeset bytes into the local document using
    ///    Automerge's deterministic merge algorithm.
    /// 3. Persist the merged document bytes back to `crdt_docs`.
    /// 4. Update the local vector clock with the packet's HLC timestamp.
    ///
    /// Returns the list of entity IDs that were successfully merged.
    ///
    /// # Errors
    ///
    /// Returns [`CoreError::Sync`] if the database lock cannot be acquired.
    /// Returns [`CoreError::Crdt`] if Automerge merge fails.
    /// Returns [`CoreError::UnknownEntityType`] if the entity type is not
    /// recognized (currently `"task"`, `"note"`, `"time_block"`).
    pub fn merge_remote_deltas(&mut self, deltas: Vec<DeltaPacket>) -> Result<Vec<String>> {
        let mut merged_ids = Vec::with_capacity(deltas.len());

        let db = self
            .db
            .lock()
            .map_err(|e| CoreError::Sync(format!("database lock poisoned: {e}")))?;

        for delta in &deltas {
            // Skip empty changesets (placeholder packets from Task 2.5 stub).
            if delta.changeset_bytes.is_empty() {
                tracing::debug!(
                    entity_id = %delta.entity_id,
                    entity_type = %delta.entity_type,
                    "SyncEngine: skipping empty changeset"
                );
                // Still update the clock so we know the peer has this entity.
                self.local_clock
                    .update(delta.entity_id.clone(), delta.hlc_timestamp.clone());
                continue;
            }

            // Validate entity type.
            match delta.entity_type.as_str() {
                "task" | "note" | "time_block" => {}
                other => {
                    return Err(CoreError::UnknownEntityType(other.to_string()));
                }
            }

            // Load existing CRDT document bytes (if any).
            let existing_bytes = db.load_crdt_doc(&delta.entity_id)?;

            // Merge the remote changeset into the local document.
            let merged_bytes =
                merge_automerge_changeset(existing_bytes.as_deref(), &delta.changeset_bytes)?;

            // Compute the current Unix timestamp for `updated_at`.
            let updated_at = unix_timestamp_secs();

            // Persist the merged document.
            db.save_crdt_doc(
                &delta.entity_id,
                &delta.entity_type,
                &merged_bytes,
                updated_at,
            )?;

            // Advance the local vector clock.
            self.local_clock
                .update(delta.entity_id.clone(), delta.hlc_timestamp.clone());

            merged_ids.push(delta.entity_id.clone());

            tracing::debug!(
                entity_id = %delta.entity_id,
                entity_type = %delta.entity_type,
                hlc = %delta.hlc_timestamp,
                "SyncEngine: merged remote delta"
            );
        }

        Ok(merged_ids)
    }

    // ── Outbound: compute delta for a local mutation ──────────────────────────

    /// Compute an outbound [`DeltaPacket`] for a locally mutated entity.
    ///
    /// Loads the current Automerge document from the database, saves the full
    /// document bytes as the changeset (Automerge changesets are the full
    /// serialized document for simplicity; a future optimization can extract
    /// only the incremental changes), and returns a [`DeltaPacket`] ready for
    /// transmission.
    ///
    /// # Parameters
    ///
    /// - `entity_id` — UUID of the mutated entity.
    /// - `entity_type` — Type discriminator (`"task"`, `"note"`, `"time_block"`).
    /// - `hlc_timestamp` — The HLC timestamp of the mutation.
    /// - `sequence` — Monotonically increasing sequence number for this session.
    ///
    /// # Errors
    ///
    /// Returns [`CoreError::Sync`] if the database lock cannot be acquired or
    /// if no CRDT document exists for the given entity.
    pub fn compute_outbound_delta(
        &self,
        entity_id: &str,
        entity_type: &str,
        hlc_timestamp: &str,
        sequence: u64,
    ) -> Result<DeltaPacket> {
        let db = self
            .db
            .lock()
            .map_err(|e| CoreError::Sync(format!("database lock poisoned: {e}")))?;

        let doc_bytes = db.load_crdt_doc(entity_id)?.ok_or_else(|| {
            CoreError::Sync(format!(
                "no CRDT document found for entity '{entity_id}' — cannot compute delta"
            ))
        })?;

        Ok(DeltaPacket::new(
            entity_type,
            entity_id,
            hlc_timestamp,
            doc_bytes,
            sequence,
        ))
    }

    // ── Event loop ────────────────────────────────────────────────────────────

    /// Start the async event loop that processes inbound [`NetEvent`]s.
    ///
    /// This method consumes `self` and spawns a Tokio task.  The task runs
    /// until the broadcast channel is closed (i.e., the [`SwarmManager`] shuts
    /// down) or until a fatal error occurs.
    ///
    /// # Returns
    ///
    /// A `tokio::task::JoinHandle` for the spawned task.  The caller can
    /// `.await` it to wait for the event loop to finish.
    pub fn run(mut self) -> tokio::task::JoinHandle<()> {
        let mut event_rx: broadcast::Receiver<NetEvent> = self.swarm.subscribe_events();

        tokio::spawn(async move {
            tracing::info!("SyncEngine: event loop started");

            loop {
                match event_rx.recv().await {
                    Ok(NetEvent::DeltasReceived { from_peer, deltas }) => {
                        tracing::debug!(
                            %from_peer,
                            count = deltas.len(),
                            "SyncEngine: received deltas from peer"
                        );
                        match self.merge_remote_deltas(deltas) {
                            Ok(ids) => {
                                tracing::info!(
                                    %from_peer,
                                    merged_count = ids.len(),
                                    "SyncEngine: merged remote deltas"
                                );
                            }
                            Err(e) => {
                                tracing::warn!(
                                    %from_peer,
                                    error = %e,
                                    "SyncEngine: delta merge failed"
                                );
                            }
                        }
                    }

                    Ok(NetEvent::SyncComplete { peer_id }) => {
                        tracing::debug!(%peer_id, "SyncEngine: sync complete with peer");
                    }

                    Ok(NetEvent::SyncError { peer_id, reason }) => {
                        tracing::warn!(
                            %peer_id,
                            %reason,
                            "SyncEngine: sync error from peer"
                        );
                    }

                    Ok(NetEvent::GossipReceived {
                        entity_type,
                        entity_id,
                        hlc_timestamp,
                        origin_peer,
                    }) => {
                        tracing::debug!(
                            %entity_type,
                            %entity_id,
                            %hlc_timestamp,
                            %origin_peer,
                            "SyncEngine: gossip notification received — triggering sync"
                        );
                        // A gossip notification means a peer has new data.
                        // Trigger a full sync to pull the delta.
                        let clock = self.local_clock.clone();
                        if let Err(e) = self.swarm.trigger_sync(clock).await {
                            tracing::warn!(error = %e, "SyncEngine: trigger_sync failed");
                        }
                    }

                    Err(broadcast::error::RecvError::Lagged(n)) => {
                        tracing::warn!(
                            skipped = n,
                            "SyncEngine: event receiver lagged — some events were dropped"
                        );
                        // Continue processing; lagging is non-fatal.
                    }

                    Err(broadcast::error::RecvError::Closed) => {
                        tracing::info!("SyncEngine: event channel closed — shutting down");
                        break;
                    }
                }
            }

            tracing::info!("SyncEngine: event loop stopped");
        })
    }
}

// ---------------------------------------------------------------------------
// Automerge merge helper
// ---------------------------------------------------------------------------

/// Merge a remote Automerge changeset into an existing local document.
///
/// If `existing_bytes` is `None`, the remote changeset is treated as the
/// initial document state (i.e., the local node has never seen this entity).
///
/// Returns the serialized bytes of the merged document.
///
/// # Implementation note
///
/// Automerge's `doc.merge()` performs a deterministic, conflict-free merge of
/// two document states.  The result is the same regardless of merge order
/// (commutativity) and is idempotent for repeated merges of the same changeset.
fn merge_automerge_changeset(
    existing_bytes: Option<&[u8]>,
    remote_bytes: &[u8],
) -> Result<Vec<u8>> {
    use automerge::{AutoCommit, AutomergeError};

    // Load or create the local document.
    let mut local_doc: AutoCommit = match existing_bytes {
        Some(bytes) => AutoCommit::load(bytes)
            .map_err(|e: AutomergeError| CoreError::Crdt(format!("load local doc: {e}")))?,
        None => AutoCommit::new(),
    };

    // Load the remote document from the received bytes.
    let mut remote_doc: AutoCommit = AutoCommit::load(remote_bytes)
        .map_err(|e: AutomergeError| CoreError::Crdt(format!("load remote doc: {e}")))?;

    // Merge the remote document into the local document.
    local_doc
        .merge(&mut remote_doc)
        .map_err(|e: AutomergeError| CoreError::Crdt(format!("merge failed: {e}")))?;

    // Serialize the merged document.
    Ok(local_doc.save())
}

// ---------------------------------------------------------------------------
// Utility
// ---------------------------------------------------------------------------

/// Return the current Unix timestamp in seconds.
fn unix_timestamp_secs() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_secs()
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::*;
    use crate::crdt::{Hlc, TaskCrdt};
    use crate::db::Database;
    use chronogrid_net::swarm_manager::SwarmHandle;
    use chronogrid_net::PeerId;
    use std::sync::{Arc, Mutex};
    use tokio::sync::broadcast;

    /// A deterministic 32-byte key used exclusively in tests.
    const TEST_KEY: [u8; 32] = [0xAA; 32];

    /// Create a shared in-memory test database.
    fn make_db() -> Arc<Mutex<Database>> {
        Arc::new(Mutex::new(
            Database::open_in_memory(&TEST_KEY).expect("open_in_memory"),
        ))
    }

    /// Create a `SwarmHandle` backed by a disconnected channel for unit tests.
    ///
    /// Returns `(handle, event_tx)` so tests can inject [`NetEvent`]s.
    fn make_test_handle() -> (SwarmHandle, broadcast::Sender<NetEvent>) {
        let (event_tx, _) = broadcast::channel(16);
        let handle = SwarmHandle::new_for_test(event_tx.clone(), PeerId::random());
        (handle, event_tx)
    }

    /// Create a `SyncEngine` with a fake `SwarmHandle` for unit testing.
    ///
    /// Returns `(engine, event_tx)` so tests can inject `NetEvent`s.
    fn make_test_engine() -> (SyncEngine, broadcast::Sender<NetEvent>) {
        let (handle, event_tx) = make_test_handle();
        let db = make_db();
        let engine = SyncEngine::new(db, handle);
        (engine, event_tx)
    }

    // ── merge_automerge_changeset ─────────────────────────────────────────────

    #[test]
    fn merge_into_empty_local_doc() {
        // Create a remote document with a task.
        let mut hlc = Hlc::new("node-remote");
        let mut remote_task = TaskCrdt::new(&mut hlc);
        remote_task.set_title(&mut hlc, "Remote Task Title");
        let remote_bytes = remote_task.save();

        // Merge into a non-existent local doc (None).
        let merged = merge_automerge_changeset(None, &remote_bytes)
            .expect("merge into empty must succeed");

        // The merged doc must be loadable and contain the remote data.
        let loaded = TaskCrdt::load(&merged).expect("load merged doc");
        assert_eq!(loaded.title(), Some("Remote Task Title"));
    }

    #[test]
    fn merge_two_concurrent_edits_converges() {
        // Simulate two peers editing different fields of the same task.
        let mut hlc_a = Hlc::new("node-a");
        let mut task_a = TaskCrdt::new(&mut hlc_a);
        task_a.set_title(&mut hlc_a, "Title from A");

        let mut hlc_b = Hlc::new("node-b");
        let mut task_b = TaskCrdt::new(&mut hlc_b);
        task_b.set_title(&mut hlc_b, "Title from B");
        task_b.set_status(&mut hlc_b, "in_progress");

        let bytes_a = task_a.save();
        let bytes_b = task_b.save();

        // Merge B into A.
        let merged_a_b =
            merge_automerge_changeset(Some(&bytes_a), &bytes_b).expect("merge A←B must succeed");

        // Merge A into B.
        let merged_b_a =
            merge_automerge_changeset(Some(&bytes_b), &bytes_a).expect("merge B←A must succeed");

        // Both merged documents must be loadable.
        let doc_a_b = TaskCrdt::load(&merged_a_b).expect("load A←B");
        let doc_b_a = TaskCrdt::load(&merged_b_a).expect("load B←A");

        // Status must be "in_progress" in both (only B set it).
        assert_eq!(doc_a_b.status(), Some("in_progress"));
        assert_eq!(doc_b_a.status(), Some("in_progress"));
    }

    #[test]
    fn merge_idempotent_same_changeset() {
        let mut hlc = Hlc::new("node-x");
        let mut task = TaskCrdt::new(&mut hlc);
        task.set_title(&mut hlc, "Idempotent Task");
        let bytes = task.save();

        // Merge the same bytes twice.
        let merged1 = merge_automerge_changeset(None, &bytes).expect("first merge");
        let merged2 =
            merge_automerge_changeset(Some(&merged1), &bytes).expect("second merge");

        let doc1 = TaskCrdt::load(&merged1).expect("load merged1");
        let doc2 = TaskCrdt::load(&merged2).expect("load merged2");

        // Title must be the same after idempotent merge.
        assert_eq!(doc1.title(), doc2.title());
    }

    #[test]
    fn merge_invalid_remote_bytes_returns_error() {
        let result = merge_automerge_changeset(None, b"not-valid-automerge-bytes");
        assert!(result.is_err(), "invalid bytes must return an error");
        match result.unwrap_err() {
            CoreError::Crdt(_) => {}
            other => panic!("expected CoreError::Crdt, got: {other:?}"),
        }
    }

    // ── SyncEngine::merge_remote_deltas ──────────────────────────────────────

    /// Build a minimal `DeltaPacket` from a `TaskCrdt` for testing.
    fn make_task_delta(entity_id: &str, hlc: &str, task: &TaskCrdt, seq: u64) -> DeltaPacket {
        DeltaPacket::new("task", entity_id, hlc, task.save(), seq)
    }

    /// Build a minimal `DeltaPacket` from a `NoteCrdt` for testing.
    fn make_note_delta(entity_id: &str, hlc: &str, note: &NoteCrdt, seq: u64) -> DeltaPacket {
        DeltaPacket::new("note", entity_id, hlc, note.save(), seq)
    }

    #[test]
    fn merge_remote_deltas_empty_list_returns_empty() {
        let (mut engine, _event_tx) = make_test_engine();
        let result = engine.merge_remote_deltas(vec![]).expect("empty merge");
        assert!(result.is_empty());
    }

    #[test]
    fn merge_remote_deltas_task_persisted_to_db() {
        let (handle, _event_tx) = make_test_handle();
        let db = make_db();
        let mut engine = SyncEngine::new(Arc::clone(&db), handle);

        // Create a remote task delta.
        let mut hlc = Hlc::new("remote-node");
        let mut task = TaskCrdt::new(&mut hlc);
        task.set_title(&mut hlc, "Synced Task");
        let hlc_str = hlc.tick().to_string();
        let delta = make_task_delta("task-uuid-001", &hlc_str, &task, 0);

        let merged = engine.merge_remote_deltas(vec![delta]).expect("merge");
        assert_eq!(merged, vec!["task-uuid-001"]);

        // Verify the CRDT doc was persisted.
        let stored = db
            .lock()
            .unwrap()
            .load_crdt_doc("task-uuid-001")
            .expect("load")
            .expect("must exist");

        let loaded_task = TaskCrdt::load(&stored).expect("load task");
        assert_eq!(loaded_task.title(), Some("Synced Task"));
    }

    #[test]
    fn merge_remote_deltas_note_persisted_to_db() {
        let (handle, _event_tx) = make_test_handle();
        let db = make_db();
        let mut engine = SyncEngine::new(Arc::clone(&db), handle);

        let mut hlc = Hlc::new("remote-node");
        let mut note = NoteCrdt::new(&mut hlc);
        note.set_title(&mut hlc, "Synced Note");
        let hlc_str = hlc.tick().to_string();
        let delta = make_note_delta("note-uuid-001", &hlc_str, &note, 0);

        let merged = engine.merge_remote_deltas(vec![delta]).expect("merge");
        assert_eq!(merged, vec!["note-uuid-001"]);

        let stored = db
            .lock()
            .unwrap()
            .load_crdt_doc("note-uuid-001")
            .expect("load")
            .expect("must exist");

        let loaded_note = NoteCrdt::load(&stored).expect("load note");
        assert_eq!(loaded_note.title(), Some("Synced Note"));
    }

    #[test]
    fn merge_remote_deltas_updates_local_clock() {
        let (handle, _event_tx) = make_test_handle();
        let db = make_db();
        let mut engine = SyncEngine::new(db, handle);

        let mut hlc = Hlc::new("remote-node");
        let mut task = TaskCrdt::new(&mut hlc);
        task.set_title(&mut hlc, "Clock Test Task");
        let hlc_str = "1724000000000-0001-ab12".to_string();
        let delta = make_task_delta("task-clock-001", &hlc_str, &task, 0);

        engine.merge_remote_deltas(vec![delta]).expect("merge");

        // The local clock must now track this entity.
        assert_eq!(
            engine.local_clock().get("task-clock-001"),
            Some("1724000000000-0001-ab12")
        );
    }

    #[test]
    fn merge_remote_deltas_empty_changeset_skipped() {
        let (handle, _event_tx) = make_test_handle();
        let db = make_db();
        let mut engine = SyncEngine::new(Arc::clone(&db), handle);

        // Empty changeset (placeholder from Task 2.5 stub).
        let delta =
            DeltaPacket::new("task", "task-empty-001", "1724000000000-0001-ab12", vec![], 0);

        let merged = engine.merge_remote_deltas(vec![delta]).expect("merge");
        // Empty changesets are skipped — not in the merged list.
        assert!(merged.is_empty());

        // But the clock must still be updated.
        assert_eq!(
            engine.local_clock().get("task-empty-001"),
            Some("1724000000000-0001-ab12")
        );

        // And no CRDT doc must have been written.
        let stored = db
            .lock()
            .unwrap()
            .load_crdt_doc("task-empty-001")
            .expect("load");
        assert!(stored.is_none(), "empty changeset must not write to DB");
    }

    #[test]
    fn merge_remote_deltas_unknown_entity_type_returns_error() {
        let (handle, _event_tx) = make_test_handle();
        let db = make_db();
        let mut engine = SyncEngine::new(db, handle);

        // Non-empty changeset with unknown entity type.
        let delta =
            DeltaPacket::new("unknown_type", "entity-001", "ts-001", vec![0xDE, 0xAD], 0);

        let result = engine.merge_remote_deltas(vec![delta]);
        assert!(result.is_err());
        match result.unwrap_err() {
            CoreError::UnknownEntityType(t) => assert_eq!(t, "unknown_type"),
            other => panic!("expected UnknownEntityType, got: {other:?}"),
        }
    }

    #[test]
    fn merge_remote_deltas_multiple_entities() {
        let (handle, _event_tx) = make_test_handle();
        let db = make_db();
        let mut engine = SyncEngine::new(Arc::clone(&db), handle);

        let mut hlc = Hlc::new("remote-node");

        let mut task1 = TaskCrdt::new(&mut hlc);
        task1.set_title(&mut hlc, "Task One");
        let hlc1 = hlc.tick().to_string();

        let mut task2 = TaskCrdt::new(&mut hlc);
        task2.set_title(&mut hlc, "Task Two");
        let hlc2 = hlc.tick().to_string();

        let deltas = vec![
            make_task_delta("task-multi-001", &hlc1, &task1, 0),
            make_task_delta("task-multi-002", &hlc2, &task2, 1),
        ];

        let merged = engine.merge_remote_deltas(deltas).expect("merge");
        assert_eq!(merged.len(), 2);
        assert!(merged.contains(&"task-multi-001".to_string()));
        assert!(merged.contains(&"task-multi-002".to_string()));

        // Both docs must be in the DB.
        let db_guard = db.lock().unwrap();
        assert!(db_guard.load_crdt_doc("task-multi-001").unwrap().is_some());
        assert!(db_guard.load_crdt_doc("task-multi-002").unwrap().is_some());
    }

    // ── SyncEngine::compute_outbound_delta ────────────────────────────────────

    #[test]
    fn compute_outbound_delta_returns_packet() {
        let (handle, _event_tx) = make_test_handle();
        let db = make_db();
        let engine = SyncEngine::new(Arc::clone(&db), handle);

        // Pre-populate a CRDT doc in the DB.
        let mut hlc = Hlc::new("local-node");
        let mut task = TaskCrdt::new(&mut hlc);
        task.set_title(&mut hlc, "Outbound Task");
        let doc_bytes = task.save();
        let hlc_str = hlc.tick().to_string();

        db.lock()
            .unwrap()
            .save_crdt_doc("task-out-001", "task", &doc_bytes, 1000)
            .expect("save");

        let packet = engine
            .compute_outbound_delta("task-out-001", "task", &hlc_str, 0)
            .expect("compute_outbound_delta");

        assert_eq!(packet.entity_id, "task-out-001");
        assert_eq!(packet.entity_type, "task");
        assert_eq!(packet.hlc_timestamp, hlc_str);
        assert_eq!(packet.sequence, 0);
        assert!(!packet.changeset_bytes.is_empty());
    }

    #[test]
    fn compute_outbound_delta_missing_entity_returns_error() {
        let (handle, _event_tx) = make_test_handle();
        let db = make_db();
        let engine = SyncEngine::new(db, handle);

        let result = engine.compute_outbound_delta("nonexistent", "task", "ts-001", 0);
        assert!(result.is_err());
        match result.unwrap_err() {
            CoreError::Sync(_) => {}
            other => panic!("expected CoreError::Sync, got: {other:?}"),
        }
    }

    // ── advance_clock ─────────────────────────────────────────────────────────

    #[test]
    fn advance_clock_updates_local_clock() {
        let (handle, _event_tx) = make_test_handle();
        let db = make_db();
        let mut engine = SyncEngine::new(db, handle);

        engine.advance_clock("entity-001", "1724000000000-0001-ab12");
        assert_eq!(
            engine.local_clock().get("entity-001"),
            Some("1724000000000-0001-ab12")
        );

        // Advancing with a newer timestamp must update.
        engine.advance_clock("entity-001", "1724000000001-0001-ab12");
        assert_eq!(
            engine.local_clock().get("entity-001"),
            Some("1724000000001-0001-ab12")
        );

        // Advancing with an older timestamp must NOT update.
        engine.advance_clock("entity-001", "1724000000000-0001-ab12");
        assert_eq!(
            engine.local_clock().get("entity-001"),
            Some("1724000000001-0001-ab12"),
            "older timestamp must not overwrite newer"
        );
    }

    // ── SyncEngine::run event loop ────────────────────────────────────────────

    #[tokio::test]
    async fn run_event_loop_processes_deltas_received() {
        let (handle, event_tx) = make_test_handle();
        let db = make_db();
        let engine = SyncEngine::new(Arc::clone(&db), handle);

        // Prepare a real task delta.
        let mut hlc = Hlc::new("remote-node");
        let mut task = TaskCrdt::new(&mut hlc);
        task.set_title(&mut hlc, "Event Loop Task");
        let hlc_str = hlc.tick().to_string();
        let delta = DeltaPacket::new("task", "task-loop-001", &hlc_str, task.save(), 0);

        // Start the event loop.
        let join = engine.run();

        // Inject a DeltasReceived event.
        event_tx
            .send(NetEvent::DeltasReceived {
                from_peer: PeerId::random(),
                deltas: vec![delta],
            })
            .expect("send event");

        // Give the event loop time to process.
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

        // Close the channel to stop the event loop.
        drop(event_tx);
        let _ = join.await;

        // Verify the delta was merged into the DB.
        let stored = db
            .lock()
            .unwrap()
            .load_crdt_doc("task-loop-001")
            .expect("load")
            .expect("must exist after event loop merge");

        let loaded = TaskCrdt::load(&stored).expect("load task");
        assert_eq!(loaded.title(), Some("Event Loop Task"));
    }

    #[tokio::test]
    async fn run_event_loop_stops_on_channel_close() {
        let (handle, event_tx) = make_test_handle();
        let db = make_db();
        let engine = SyncEngine::new(db, handle);

        let join = engine.run();

        // Close the channel immediately.
        drop(event_tx);

        // The event loop must terminate cleanly.
        tokio::time::timeout(std::time::Duration::from_secs(1), join)
            .await
            .expect("event loop must stop within 1 second")
            .expect("join handle must not panic");
    }
}