//! `Node` — the application-level mesh document layer.
//!
//! `Node` is the canonical home for application-level mesh document state.
//! It wraps a [`DataSyncBackend`] (production: peat-protocol's `AutomergeBackend`;
//! tests: [`InMemoryBackend`]) and provides a thin, transport-agnostic surface
//! for publishing, querying, and observing documents.
//!
//! Per-doc-type convenience surfaces (`send_chat`, `send_emergency`, …) will
//! live in sibling modules under `node::` in subsequent commits, each as a
//! thin layer over [`Node::publish`] / [`Node::observe`].
//!
//! Cross-transport behavior:
//! - iroh-bound peers sync natively over Automerge via the backend's own sync
//!   engine.
//! - BLE-bound peers go through `peat-protocol::sync::ble_translation::BleTranslator`
//!   at the gateway boundary — wired in by callers (peat-ffi, gateway nodes)
//!   that compose `Node` with a BLE transport. The wiring is *not* part of
//!   `Node` itself, to keep this crate free of transport-specific knowledge.
//!
//! Why so thin: the backend trait stack (`DocumentStore` / `PeerDiscovery` /
//! `SyncEngine` / `DataSyncBackend`) per ADR-005 is already correctly factored
//! and already dispatches change events through the backend's own observer.
//! Synthesizing events here on `publish`/`delete` would deliver duplicates
//! alongside the backend stream and would force the synthesizer to second-guess
//! the backend's deletion policy (ADR-034); both are wrong. The right design
//! is to delegate to the backend's existing observer and let it be the single
//! source of truth.
//!
//! Naming note: `peat-mesh::PeatMesh` (in `mesh.rs`) is the transport
//! orchestrator. `peat-mesh::Node` is the document layer. Compose them: a
//! `PeatMesh` typically holds a `Node`.

use std::sync::Arc;

use crate::sync::traits::DataSyncBackend;
use crate::sync::types::{ChangeStream, Document, DocumentId, Query};

/// Application-level mesh document layer.
///
/// Holds an [`Arc<dyn DataSyncBackend>`] as the canonical doc store.
/// Cloning a `Node` shares the underlying backend.
#[derive(Clone)]
pub struct Node {
    backend: Arc<dyn DataSyncBackend>,
}

impl Node {
    /// Construct a new `Node` over the given backend.
    pub fn new(backend: Arc<dyn DataSyncBackend>) -> Self {
        Self { backend }
    }

    /// The underlying backend, for callers that need lifecycle control
    /// (`initialize`, `shutdown`) or backend-specific extensions.
    pub fn backend(&self) -> &Arc<dyn DataSyncBackend> {
        &self.backend
    }

    /// Publish a document into a collection. Returns the assigned id.
    ///
    /// The backend emits a [`ChangeEvent`] on its observer stream as part of
    /// the upsert; subscribers attached via [`Node::observe`] receive it
    /// asynchronously. `Node` does not synthesize a separate event — that
    /// would deliver a duplicate alongside the backend's own stream.
    ///
    /// [`ChangeEvent`]: crate::sync::types::ChangeEvent
    pub async fn publish(
        &self,
        collection: &str,
        document: Document,
    ) -> anyhow::Result<DocumentId> {
        self.publish_with_origin(collection, document, None).await
    }

    /// Publish a document with a per-call origin hint per ADR-059.
    ///
    /// Used by transport bridges to mark a document as "ingested via
    /// transport `X`" so that `TransportManager`'s observer-driven
    /// fan-out can suppress the same-node echo loop (`X → Node →
    /// observer → X`). The origin rides on the resulting
    /// `ChangeEvent.origin`; it is not stored on the document, not
    /// serialized to the wire, and not replicated to remote peers.
    ///
    /// Locally-published documents (application code calling `publish`
    /// without specifying an origin) and Automerge-sync-delivered
    /// documents both produce events with `origin = None` — the right
    /// answer for both: locally-published has no transport-of-origin,
    /// and sync-delivered has no local-transport-of-origin (only
    /// remote nodes' transports were involved, which is invisible to
    /// this node by design).
    pub async fn publish_with_origin(
        &self,
        collection: &str,
        document: Document,
        origin: Option<String>,
    ) -> anyhow::Result<DocumentId> {
        self.backend
            .document_store()
            .upsert_with_origin(collection, document, origin)
            .await
    }

    /// Get a document by id.
    pub async fn get(&self, collection: &str, id: &DocumentId) -> anyhow::Result<Option<Document>> {
        self.backend.document_store().get(collection, id).await
    }

    /// Query documents in a collection.
    pub async fn query(&self, collection: &str, query: &Query) -> anyhow::Result<Vec<Document>> {
        self.backend.document_store().query(collection, query).await
    }

    /// Delete a document by id, applying the backend's deletion policy.
    ///
    /// Returns the [`DeleteResult`] from the backend. The actual change event
    /// the backend emits depends on the policy (per ADR-034): under
    /// [`DeletionPolicy::SoftDelete`] (the default) the backend emits
    /// `ChangeEvent::Updated` with `_deleted=true`; under `ImmediatePurge` it
    /// emits `ChangeEvent::Removed`. `Node` does *not* synthesize a `Removed`
    /// event — that would misrepresent the soft-delete case to subscribers.
    ///
    /// [`DeleteResult`]: crate::qos::DeleteResult
    /// [`DeletionPolicy::SoftDelete`]: crate::qos::DeletionPolicy::SoftDelete
    pub async fn delete(
        &self,
        collection: &str,
        id: &DocumentId,
        reason: Option<&str>,
    ) -> anyhow::Result<crate::qos::DeleteResult> {
        self.backend
            .document_store()
            .delete(collection, id, reason)
            .await
    }

    /// Observe document changes for a collection.
    ///
    /// Returns a [`ChangeStream`] backed by the underlying backend's observer
    /// (`DocumentStore::observe`). This is the *single* path through which
    /// in-process subscribers receive events for both local-publish-driven
    /// updates and peer-sync-driven updates.
    ///
    /// Bounded-channel migration note: `ChangeStream` today wraps an
    /// `mpsc::UnboundedReceiver` (per `sync::types`), so a slow consumer can
    /// grow memory under burst sync (e.g. partition heal). Switching to a
    /// bounded broadcast (matching `PeatMesh`'s `broadcast::Receiver<PeatMeshEvent>`
    /// pattern) is a separate change against `sync::types::ChangeStream`
    /// itself, since it ripples through every `DocumentStore` impl.
    pub fn observe(&self, collection: &str, query: &Query) -> anyhow::Result<ChangeStream> {
        self.backend.document_store().observe(collection, query)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::sync::in_memory::InMemoryBackend;
    use crate::sync::types::ChangeEvent;
    use serde_json::Value;
    use std::collections::HashMap;

    fn doc(text: &str) -> Document {
        let mut fields = HashMap::new();
        fields.insert("text".to_string(), Value::String(text.to_string()));
        Document::new(fields)
    }

    /// A subscriber attached via `Node::observe` receives exactly one
    /// `ChangeEvent::Updated` per `publish` — not two. Guards against
    /// re-introducing synthetic in-`Node` dispatch alongside the backend's
    /// own observer (which would deliver duplicates).
    #[tokio::test]
    async fn publish_emits_exactly_one_event() {
        let backend: Arc<dyn DataSyncBackend> = Arc::new(InMemoryBackend::new_initialized());
        let node = Node::new(backend);

        let mut stream = node.observe("chats", &Query::All).expect("observe");

        let id = node.publish("chats", doc("hello")).await.expect("publish");

        let event = stream.receiver.recv().await.expect("event");
        match event {
            ChangeEvent::Updated {
                collection,
                document,
                origin,
            } => {
                assert_eq!(collection, "chats");
                assert_eq!(document.id.as_deref(), Some(id.as_str()));
                // `publish` is the no-origin convenience; the resulting
                // event must carry `origin = None` per ADR-059.
                assert_eq!(origin, None);
            }
            other => panic!("unexpected event: {:?}", other),
        }

        // Nothing else queued: the publish path didn't synthesize a duplicate.
        let timeout = tokio::time::sleep(std::time::Duration::from_millis(50));
        tokio::select! {
            extra = stream.receiver.recv() => panic!("got duplicate event: {:?}", extra),
            _ = timeout => {}
        }
    }

    /// Under the default `SoftDelete` policy, the backend emits
    /// `ChangeEvent::Updated` (with `_deleted=true`), not `Removed`. Verify
    /// that's what subscribers see, AND that exactly one event is delivered
    /// (no synthetic in-`Node` dispatch sitting alongside the backend's).
    #[tokio::test]
    async fn delete_default_policy_does_not_lie_about_event_variant() {
        let backend: Arc<dyn DataSyncBackend> = Arc::new(InMemoryBackend::new_initialized());
        let node = Node::new(backend);

        let id = node
            .publish("chats", doc("delete-me"))
            .await
            .expect("publish");

        let mut stream = node.observe("chats", &Query::All).expect("observe");

        node.delete("chats", &id, None).await.expect("delete");

        let event = stream.receiver.recv().await.expect("event");
        // The default policy is SoftDelete; the backend signals it via
        // `ChangeEvent::Updated`. `Node` does not turn that into `Removed`.
        match event {
            ChangeEvent::Updated { collection, .. } => {
                assert_eq!(collection, "chats");
            }
            ChangeEvent::Removed { .. } => {
                panic!(
                    "node emitted Removed under default SoftDelete policy — \
                     should defer to backend, which emits Updated for soft-delete"
                );
            }
            other => panic!("unexpected event: {:?}", other),
        }

        // Same duplicate-guard as the publish test: nothing else queued.
        let timeout = tokio::time::sleep(std::time::Duration::from_millis(50));
        tokio::select! {
            extra = stream.receiver.recv() => panic!("got duplicate event after delete: {:?}", extra),
            _ = timeout => {}
        }
    }

    /// Sanity: `query` returns documents `publish` put in.
    #[tokio::test]
    async fn query_returns_published_docs() {
        let backend: Arc<dyn DataSyncBackend> = Arc::new(InMemoryBackend::new_initialized());
        let node = Node::new(backend);

        node.publish("chats", doc("first")).await.expect("publish");
        node.publish("chats", doc("second")).await.expect("publish");

        let results = node.query("chats", &Query::All).await.expect("query");
        assert_eq!(results.len(), 2);
    }

    /// Observers are scoped to their collection.
    #[tokio::test]
    async fn collection_isolation() {
        let backend: Arc<dyn DataSyncBackend> = Arc::new(InMemoryBackend::new_initialized());
        let node = Node::new(backend);

        let mut chats = node.observe("chats", &Query::All).expect("observe chats");
        let mut markers = node
            .observe("markers", &Query::All)
            .expect("observe markers");

        node.publish("chats", doc("c1"))
            .await
            .expect("publish chat");

        // chats observer sees the chat.
        let _ = chats.receiver.recv().await.expect("chat event");

        // markers observer should NOT see the chat event.
        let timeout = tokio::time::sleep(std::time::Duration::from_millis(50));
        tokio::select! {
            _ = markers.receiver.recv() => panic!("markers observer received unrelated chat event"),
            _ = timeout => {}
        }
    }
}
