//! NatsSync — NATS JetStream `SyncBackend` (bl-031 / ADR-042).
//!
//! The first real (network) transport for the multi-actor witness seam.
//! Per ADR-042 the `SyncBackend` trait stays synchronous; NatsSync owns its
//! own current-thread tokio runtime and `block_on`s the async `platform-nats`
//! client internally, so the substrate and `FilesystemSync` remain
//! runtime-free. It MUST be driven from a blocking context (never a tokio
//! worker). Feature-gated on `nats` and an optional accelerator — the
//! zero-dependency `FilesystemSync` stays the default (ADR-042 constraints).
//!
//! Verb mapping (ADR-042): for a JetStream pull transport the announcement
//! carries the body, so [`publish_op`](NatsSync::publish_op) publishes the
//! whole signed `Operation` to `ontoref.sync.<project>.ops.<op_id>`;
//! `announced_ops` pulls the durable consumer and indexes ops by id;
//! `fetch_op` reads that index; `announce` is a no-op.

#![cfg(feature = "nats")]

use std::collections::HashMap;
use std::sync::Mutex;

use bytes::Bytes;
use ontoref_sync::{SyncBackend, SyncError};
use ontoref_types::{canonical, Operation};
use platform_nats::topology::{AckPolicy, ConsumerDef, Retention, Storage, StreamDef};
use platform_nats::{EventStream, NatsConnectionConfig, TopologyConfig};
use tokio::runtime::Runtime;

/// NATS JetStream sync backend bound to one project stream and one durable
/// consumer (per local peer). Holds its own runtime — see module docs.
pub struct NatsSync {
    rt: Runtime,
    stream: EventStream,
    subject_root: String,
    index: Mutex<HashMap<String, Vec<u8>>>,
}

impl NatsSync {
    /// Connect to `url`, ensure the project's JetStream stream + this peer's
    /// durable consumer exist, and bind for pulling. Anonymous (no NKey) —
    /// suitable for a trusted local/cluster broker.
    ///
    /// # Errors
    ///
    /// [`SyncError::Io`] wrapping any connection / topology / bind failure.
    pub fn connect(url: &str, project: &str, peer_id: &str) -> Result<Self, SyncError> {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .map_err(|e| io("building runtime", e))?;

        let subject_root = format!("ontoref.sync.{project}.ops");
        let stream_name = format!("ontoref-sync-{project}");
        let consumer_name = format!("ontoref-sync-{project}-{peer_id}");
        let subject_root_for_topo = subject_root.clone();

        let stream = rt.block_on(async {
            let conn = NatsConnectionConfig {
                url: url.to_owned(),
                nkey_seed: None,
                require_signed_messages: false,
                trusted_nkeys: vec![],
            };
            let mut s = EventStream::connect_client(&conn)
                .await
                .map_err(|e| anyhow_io("connect_client", &e))?;
            let topo = TopologyConfig {
                streams: vec![StreamDef {
                    name: stream_name.clone(),
                    subjects: vec![format!("{subject_root_for_topo}.>")],
                    retention: Retention::Limits,
                    max_age_days: 0,
                    storage: Storage::File,
                    max_messages: 0,
                    max_bytes: 0,
                }],
                consumers: vec![ConsumerDef {
                    name: consumer_name.clone(),
                    stream: stream_name.clone(),
                    ack_policy: AckPolicy::Explicit,
                    filter_subjects: vec![],
                }],
            };
            s.apply_topology(&topo)
                .await
                .map_err(|e| anyhow_io("apply_topology", &e))?;
            s.bind_consumer(&stream_name, &consumer_name)
                .await
                .map_err(|e| anyhow_io("bind_consumer", &e))?;
            Ok::<EventStream, SyncError>(s)
        })?;

        Ok(Self {
            rt,
            stream,
            subject_root,
            index: Mutex::new(HashMap::new()),
        })
    }

    /// Publish a full signed operation to the project stream — the NATS
    /// analogue of `FilesystemSync::publish_body`. The JetStream
    /// announcement carries the body (ADR-042).
    ///
    /// # Errors
    ///
    /// [`SyncError::Io`] on encode or publish failure.
    pub fn publish_op(&self, op: &Operation) -> Result<(), SyncError> {
        let id = hex(&op.id().0);
        let bytes = canonical::encode(op)
            .map_err(|e| io("encoding op", std::io::Error::other(e.to_string())))?;
        let subject = format!("{}.{id}", self.subject_root);
        // Flush after publish: async-nats only enqueues the message, and a
        // current-thread runtime does not drive the connection between
        // block_on calls — without an explicit flush the op can sit unsent.
        self.rt.block_on(async {
            self.stream
                .publish(&subject, Bytes::from(bytes))
                .await
                .map_err(|e| anyhow_io("publish", &e))?;
            self.stream
                .client()
                .flush()
                .await
                .map_err(|e| io("flushing publish", std::io::Error::other(e.to_string())))?;
            Ok::<(), SyncError>(())
        })?;
        Ok(())
    }

    /// Pull pending messages, index them by op id, and ack so they are not
    /// redelivered. Each peer has its own durable consumer, so every peer
    /// observes every published op.
    fn refresh(&self) -> Result<(), SyncError> {
        let pulled = self
            .rt
            .block_on(self.stream.pull_batch(256))
            .map_err(|e| anyhow_io("pull_batch", &e))?;
        let mut idx = self.index.lock().unwrap_or_else(|e| e.into_inner());
        for (subject, payload, msg) in pulled {
            if let Some(id) = subject.rsplit('.').next() {
                idx.insert(id.to_owned(), payload.to_vec());
            }
            let _ = self.rt.block_on(msg.ack());
        }
        Ok(())
    }
}

impl SyncBackend for NatsSync {
    fn announce(&mut self, _op_id: &str) -> Result<(), SyncError> {
        // No-op: the body is published via `publish_op` (ADR-042 — the
        // JetStream announcement carries the operation).
        Ok(())
    }

    fn discover_peers(&self) -> Vec<String> {
        // The JetStream stream is shared; peer enumeration is not modeled by
        // this transport (every peer reads the same stream).
        Vec::new()
    }

    fn fetch_op(&self, op_id: &str) -> Result<Option<Vec<u8>>, SyncError> {
        Ok(self
            .index
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .get(op_id)
            .cloned())
    }

    fn subscribe_peer(&mut self, _peer_id: &str) -> Result<(), SyncError> {
        Ok(())
    }

    fn announced_ops(&self, _peer_id: &str) -> Result<Vec<String>, SyncError> {
        self.refresh()?;
        let idx = self.index.lock().unwrap_or_else(|e| e.into_inner());
        let mut ids: Vec<String> = idx.keys().cloned().collect();
        ids.sort();
        Ok(ids)
    }
}

/// One synchronous sync round (bl-018): publish `to_publish` to the stream,
/// then pull every peer op and ingest the unseen ones into `sub`, returning
/// the count newly ingested. Pure synchronous code — MUST be driven from a
/// blocking context (e.g. `tokio::task::spawn_blocking`), never a tokio
/// worker, because [`NatsSync`] `block_on`s its own runtime (ADR-042).
///
/// # Errors
///
/// [`SyncError`] on a transport failure; [`ontoref_core::Error`] folded into
/// it on an ingest failure.
pub fn sync_round(
    nats: &NatsSync,
    sub: &mut ontoref_core::Substrate,
    to_publish: &[Operation],
) -> Result<usize, SyncError> {
    for op in to_publish {
        nats.publish_op(op)?;
    }
    let ids = nats.announced_ops("")?;
    let mut peer_ops = Vec::new();
    for id in ids {
        if let Some(bytes) = nats.fetch_op(&id)? {
            if let Ok(op) = canonical::decode::<Operation>(&bytes) {
                peer_ops.push(op);
            }
        }
    }
    sub.ingest(&peer_ops)
        .map_err(|e| io("ingesting peer ops", std::io::Error::other(e.to_string())))
}

fn hex(bytes: &[u8]) -> String {
    let mut s = String::with_capacity(bytes.len() * 2);
    for b in bytes {
        use std::fmt::Write as _;
        let _ = write!(s, "{b:02x}");
    }
    s
}

fn io(context: &str, source: std::io::Error) -> SyncError {
    SyncError::Io {
        context: context.to_owned(),
        source,
    }
}

fn anyhow_io(context: &str, e: &anyhow::Error) -> SyncError {
    SyncError::Io {
        context: format!("{context}: {e}"),
        source: std::io::Error::other(e.to_string()),
    }
}

#[cfg(test)]
mod tests {
    use ed25519_dalek::SigningKey;
    use ontoref_types::{
        operation::{OpBody, OpPayload},
        AttrId, EntityId, Hlc, PublicKey, Value,
    };

    use super::*;

    /// End-to-end round-trip against a real broker. Env-gated on
    /// `ONTOREF_NATS_URL` (e.g. `nats://0.0.0.0:4222`); skips when unset so
    /// CI without a broker stays green. Run with the broker:
    ///   ONTOREF_NATS_URL=nats://0.0.0.0:4222 cargo test -p ontoref-daemon \
    ///     --features nats nats_round_trip -- --nocapture
    #[test]
    fn nats_round_trip_publish_pull_verify() {
        let Ok(url) = std::env::var("ONTOREF_NATS_URL") else {
            eprintln!("ONTOREF_NATS_URL unset — skipping NATS integration test");
            return;
        };
        // Unique project per run isolates the stream from prior runs.
        let suffix = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_nanos())
            .unwrap_or(0);
        let project = format!("itest-{suffix}");

        let key = SigningKey::from_bytes(&[5u8; 32]);
        let op = Operation::sign(
            OpBody {
                actor: PublicKey::from_signing_key(&key),
                parents: vec![],
                payload: OpPayload::EntityAssert {
                    attrs: vec![(AttrId::new("v"), Value::Int(7))],
                    entity: EntityId::new("e"),
                    valid_from: None,
                },
                timestamp: Hlc::new(1, 0),
            },
            &key,
        );

        let alice = NatsSync::connect(&url, &project, "alice").expect("alice connect");
        let bob = NatsSync::connect(&url, &project, "bob").expect("bob connect");

        alice.publish_op(&op).expect("publish");

        // Pull with a few retries for JetStream propagation.
        let target = hex(&op.id().0);
        let mut fetched = None;
        for _ in 0..30 {
            let ids = bob.announced_ops("alice").expect("announced_ops");
            if ids.contains(&target) {
                fetched = bob.fetch_op(&target).expect("fetch_op");
                break;
            }
            std::thread::sleep(std::time::Duration::from_millis(100));
        }

        let bytes = fetched.expect("the op propagates to the peer over NATS JetStream");
        let decoded: Operation = canonical::decode(&bytes).expect("decode");
        decoded
            .verify_signature()
            .expect("the op verifies end-to-end after the NATS round-trip");
        assert_eq!(decoded.id(), op.id(), "same op id after transport");
    }

    /// Run sync rounds until one ingests an op or `tries` is exhausted —
    /// gives JetStream a moment to propagate. Returns the count ingested.
    fn pull_until_ingested(
        nats: &NatsSync,
        sub: &mut ontoref_core::Substrate,
        tries: u32,
    ) -> usize {
        for _ in 0..tries {
            let n = super::sync_round(nats, sub, &[]).expect("bob round");
            if n > 0 {
                return n;
            }
            std::thread::sleep(std::time::Duration::from_millis(100));
        }
        0
    }

    /// bl-018: two Substrates converge via `sync_round` over NATS, with ALL
    /// NatsSync work (connect + round, which `block_on` an owned runtime)
    /// driven from `spawn_blocking` — the daemon pattern mandated by ADR-042
    /// (never a tokio worker). Env-gated on `ONTOREF_NATS_URL`.
    #[tokio::test]
    async fn sync_round_converges_via_spawn_blocking() {
        let Ok(url) = std::env::var("ONTOREF_NATS_URL") else {
            eprintln!("ONTOREF_NATS_URL unset — skipping NATS sync-round test");
            return;
        };
        let dir = tempfile::tempdir().expect("tempdir");
        let a_path = dir.path().join("a");
        let b_path = dir.path().join("b");
        let suffix = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_nanos())
            .unwrap_or(0);
        let project = format!("round-{suffix}");

        let key = SigningKey::from_bytes(&[6u8; 32]);
        let op = Operation::sign(
            OpBody {
                actor: PublicKey::from_signing_key(&key),
                parents: vec![],
                payload: OpPayload::EntityAssert {
                    attrs: vec![(AttrId::new("v"), Value::Int(42))],
                    entity: EntityId::new("shared"),
                    valid_from: None,
                },
                timestamp: Hlc::new(1, 0),
            },
            &key,
        );

        let (alice_root, bob_root, ingested) = tokio::task::spawn_blocking(move || {
            use ontoref_core::Substrate;

            // Everything that block_on's NatsSync runs on the blocking thread.
            let mut alice_sub = Substrate::open(&a_path).expect("alice substrate");
            let mut bob_sub = Substrate::open(&b_path).expect("bob substrate");
            let alice_nats = NatsSync::connect(&url, &project, "alice").expect("alice nats");
            let bob_nats = NatsSync::connect(&url, &project, "bob").expect("bob nats");

            // Alice applies locally and publishes via a round.
            alice_sub.apply(&op).expect("alice apply");
            super::sync_round(&alice_nats, &mut alice_sub, std::slice::from_ref(&op))
                .expect("alice round");

            // Bob pulls + ingests (retry for JetStream propagation).
            let ingested = pull_until_ingested(&bob_nats, &mut bob_sub, 30);
            (alice_sub.state_root(), bob_sub.state_root(), ingested)
        })
        .await
        .expect("spawn_blocking join");

        assert_eq!(ingested, 1, "bob ingests alice's op over NATS");
        assert_eq!(
            alice_root, bob_root,
            "two substrates converge via sync_round over NATS, driven from spawn_blocking"
        );
    }
}
