//! The content-discovery seam.
//!
//! A [`DiscoveryProvider`] is a source of signed content announcements
//! the daemon may subscribe to. v1 ships [`LanDiscovery`] (backed by
//! mDNS peer sightings) and the implicit "provider" that is ticket
//! import. A future provider maps gesture-derived gossip TopicIds
//! (redoal) to announcement streams over iroh-gossip — see
//! `docs/redoal-integration.md`.
//!
//! Consent tiering: the daemon indexes every announcement on subscribed
//! topics, but only auto-fetches content announced by trusted author
//! keys, and only from providers on the peer allowlist. Verification of
//! the fetched bytes is BLAKE3 as always — trust gates participation,
//! not integrity.

use std::collections::BTreeMap;
use std::pin::Pin;
use std::sync::Arc;

use anyhow::Result;
use iroh::{EndpointAddr, EndpointId, PublicKey, SecretKey};
use iroh_blobs::Hash;
use n0_future::Stream;
use serde::{Deserialize, Serialize};
use tokio::sync::mpsc;

/// A stream of announcements, boxed for trait-object use.
pub type BoxStream<T> = Pin<Box<dyn Stream<Item = T> + Send + 'static>>;

/// Identifies an announcement channel. For the LAN provider there is a
/// single well-known topic; gossip providers derive topics from shared
/// secrets (e.g. redoal gesture keys).
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
pub struct TopicKey(pub [u8; 32]);

/// The well-known topic for local-network presence announcements.
pub const LAN_TOPIC: TopicKey = TopicKey(*b"varde/v1/lan-presence\0\0\0\0\0\0\0\0\0\0\0");

/// A content announcement: "content `root` is available from
/// `providers`", vouched for by `author`.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Announcement {
    /// The announced content (HashSeq root or single blob hash).
    pub root: Hash,
    /// ed25519 key of whoever signs this announcement.
    pub author: PublicKey,
    /// Free-form metadata (name hints, sizes, provenance...).
    pub meta: BTreeMap<String, String>,
    /// Peers believed to hold the content.
    pub providers: Vec<EndpointAddr>,
}

/// An [`Announcement`] plus the author's signature over it and its topic.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct SignedAnnouncement {
    pub announcement: Announcement,
    /// 64-byte ed25519 signature (serde can't derive for [u8; 64]).
    signature_bytes: Vec<u8>,
}

impl SignedAnnouncement {
    /// Sign `announcement` for `topic`. The signature covers the topic
    /// too, so an announcement can't be replayed onto another channel.
    /// `secret` must be the key matching `announcement.author`.
    pub fn sign(secret: &SecretKey, topic: TopicKey, announcement: Announcement) -> Result<Self> {
        anyhow::ensure!(
            secret.public() == announcement.author,
            "signing key does not match announcement author"
        );
        let payload = signing_payload(topic, &announcement)?;
        let signature = secret.sign(&payload);
        Ok(SignedAnnouncement {
            announcement,
            signature_bytes: signature.to_bytes().to_vec(),
        })
    }

    /// Verify the signature against the embedded author key and `topic`.
    pub fn verify(&self, topic: TopicKey) -> Result<()> {
        let payload = signing_payload(topic, &self.announcement)?;
        let bytes: &[u8; 64] = self
            .signature_bytes
            .as_slice()
            .try_into()
            .map_err(|_| anyhow::anyhow!("signature must be 64 bytes"))?;
        let signature = iroh::Signature::from_bytes(bytes);
        self.announcement
            .author
            .verify(&payload, &signature)
            .map_err(|e| anyhow::anyhow!("announcement signature invalid: {e}"))
    }
}

/// Deterministic bytes covered by the signature: postcard of
/// (topic, root, author, meta, providers' node ids). Provider *addresses*
/// are excluded — they are hints that relays may legitimately refresh —
/// but the provider identities are covered.
fn signing_payload(topic: TopicKey, announcement: &Announcement) -> Result<Vec<u8>> {
    let provider_ids: Vec<EndpointId> = announcement.providers.iter().map(|p| p.id).collect();
    let payload = (
        topic,
        announcement.root,
        announcement.author,
        &announcement.meta,
        provider_ids,
    );
    Ok(postcard::to_stdvec(&payload)?)
}

/// A source of content announcements the daemon may subscribe to.
pub trait DiscoveryProvider: Send + Sync {
    /// Announcements arriving on `topic`. Invalid signatures are the
    /// subscriber's problem to reject (call [`SignedAnnouncement::verify`]).
    fn subscribe(&self, topic: TopicKey) -> BoxStream<SignedAnnouncement>;

    /// Publish an announcement on `topic`.
    fn announce(&self, topic: TopicKey, ann: SignedAnnouncement) -> Result<()>;
}

/// The v1 LAN provider: turns mDNS peer sightings into announcements.
///
/// When a peer appears on the LAN, we synthesize one announcement per
/// locally pinned root, authored (and signed) by *this* daemon: "I saw
/// peer N; N may hold root R". The daemon's own trust policy then
/// decides whether to dial N. Outbound `announce` is a no-op: presence
/// on the LAN is already advertised by mDNS itself, and varde never
/// broadcasts its content list.
pub struct LanDiscovery {
    secret: SecretKey,
    sightings: std::sync::Mutex<Option<mpsc::Receiver<EndpointId>>>,
    /// Snapshot of locally pinned roots, queried per sighting.
    pins: Arc<dyn Fn() -> Vec<Hash> + Send + Sync>,
}

impl LanDiscovery {
    pub fn new(
        secret: SecretKey,
        sightings: mpsc::Receiver<EndpointId>,
        pins: Arc<dyn Fn() -> Vec<Hash> + Send + Sync>,
    ) -> LanDiscovery {
        LanDiscovery {
            secret,
            sightings: std::sync::Mutex::new(Some(sightings)),
            pins,
        }
    }
}

impl DiscoveryProvider for LanDiscovery {
    fn subscribe(&self, topic: TopicKey) -> BoxStream<SignedAnnouncement> {
        if topic != LAN_TOPIC {
            // The LAN has exactly one channel; other topics are empty.
            return Box::pin(n0_future::stream::empty());
        }
        let Some(mut sightings) = self.sightings.lock().expect("sightings lock").take() else {
            return Box::pin(n0_future::stream::empty());
        };
        let secret = self.secret.clone();
        let pins = self.pins.clone();
        let (tx, rx) = mpsc::channel(64);
        tokio::spawn(async move {
            while let Some(node_id) = sightings.recv().await {
                for root in pins() {
                    let announcement = Announcement {
                        root,
                        author: secret.public(),
                        meta: BTreeMap::from([("source".to_string(), "mdns".to_string())]),
                        providers: vec![EndpointAddr::new(node_id)],
                    };
                    match SignedAnnouncement::sign(&secret, LAN_TOPIC, announcement) {
                        Ok(signed) => {
                            if tx.send(signed).await.is_err() {
                                return;
                            }
                        }
                        Err(e) => tracing::warn!(error = %e, "signing lan announcement"),
                    }
                }
            }
        });
        Box::pin(tokio_stream_from(rx))
    }

    fn announce(&self, _topic: TopicKey, _ann: SignedAnnouncement) -> Result<()> {
        // Presence is already advertised by mDNS; varde does not
        // broadcast content lists on the LAN.
        Ok(())
    }
}

fn tokio_stream_from<T: Send + 'static>(mut rx: mpsc::Receiver<T>) -> impl Stream<Item = T> {
    n0_future::stream::poll_fn(move |cx| rx.poll_recv(cx))
}

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

    fn keypair() -> SecretKey {
        SecretKey::generate()
    }

    fn sample(author: PublicKey) -> Announcement {
        Announcement {
            root: Hash::new(b"content"),
            author,
            meta: BTreeMap::from([("name".to_string(), "demo".to_string())]),
            providers: vec![EndpointAddr::new(keypair().public())],
        }
    }

    #[test]
    fn sign_verify_round_trip() {
        let secret = keypair();
        let signed = SignedAnnouncement::sign(&secret, LAN_TOPIC, sample(secret.public())).unwrap();
        signed.verify(LAN_TOPIC).expect("valid signature verifies");
    }

    #[test]
    fn tampered_root_fails() {
        let secret = keypair();
        let mut signed =
            SignedAnnouncement::sign(&secret, LAN_TOPIC, sample(secret.public())).unwrap();
        signed.announcement.root = Hash::new(b"evil");
        assert!(signed.verify(LAN_TOPIC).is_err());
    }

    #[test]
    fn wrong_topic_fails() {
        let secret = keypair();
        let signed = SignedAnnouncement::sign(&secret, LAN_TOPIC, sample(secret.public())).unwrap();
        assert!(signed.verify(TopicKey([9u8; 32])).is_err());
    }

    #[test]
    fn forged_author_fails() {
        let secret = keypair();
        let mut signed =
            SignedAnnouncement::sign(&secret, LAN_TOPIC, sample(secret.public())).unwrap();
        // Claim a different author without their key.
        signed.announcement.author = keypair().public();
        assert!(signed.verify(LAN_TOPIC).is_err());
    }

    #[test]
    fn signing_with_mismatched_key_is_rejected() {
        let secret = keypair();
        let other = keypair();
        assert!(SignedAnnouncement::sign(&secret, LAN_TOPIC, sample(other.public())).is_err());
    }

    #[test]
    fn survives_serde_round_trip() {
        let secret = keypair();
        let signed = SignedAnnouncement::sign(&secret, LAN_TOPIC, sample(secret.public())).unwrap();
        let json = serde_json::to_string(&signed).unwrap();
        let back: SignedAnnouncement = serde_json::from_str(&json).unwrap();
        back.verify(LAN_TOPIC).expect("signature survives serde");
        assert_eq!(back, signed);
    }
}