//! Blob provider gossip — the `peat/blob-announce/1` ALPN.
//!
//! ## Why a separate ALPN
//!
//! Multi-hop blob delivery needs "who holds blob X" to propagate across the
//! mesh, not just along direct dials. The natural place to land that knowledge
//! is [`NetworkedIrohBlobStore`]'s `BlobPeerIndex` — which `fetch_blob` already
//! consults — via `advertise_blob`. The question was only how to carry the
//! announcement on the wire.
//!
//! It rides a dedicated ALPN rather than a new `SyncMessageType` because the
//! Automerge sync decoder *hard-errors* on an unrecognized message-type tag
//! (`automerge_sync.rs`: `bail!("Unknown sync message type")`). A new tag would
//! therefore break sync against any older node during a mixed-version rollout.
//! A separate ALPN is compat-safe: iroh negotiates ALPNs, so a node that does
//! not speak `peat/blob-announce/1` simply never opens the announce stream —
//! graceful degradation to today's direct-peer behavior — and all of this code
//! stays isolated from the proven sync path. See peat-node#170.
//!
//! ## Wire format (version 1)
//!
//! ```text
//! byte  0       version (= ANNOUNCE_VERSION)
//! byte  1       ttl — remaining relay hops (0 = do not re-broadcast)
//! bytes 2..34   holder EndpointId (32 bytes)
//! byte  34      addr_count (u8)
//!   repeated    [len: u8][utf8 socket-addr bytes]
//! bytes ..      hash_count (u16 LE)
//!   repeated    [len: u16 LE][utf8 blob-hash (hex) bytes]
//! ```
//!
//! The holder's socket addresses are carried so a receiver can populate its
//! iroh `MemoryLookup` and dial a holder it is not yet directly peered with
//! (relay/holepunch permitting). `ttl` bounds transitive re-broadcast.

use std::sync::{Arc, OnceLock, Weak};
use std::time::Duration;

use iroh::endpoint::Connection;
use iroh::protocol::AcceptError;
use iroh::{Endpoint, EndpointId};
use tokio::time::timeout;
use tracing::{debug, warn};

use super::blob_traits::BlobHash;
use super::iroh_blob_store::NetworkedIrohBlobStore;

/// ALPN protocol identifier for blob provider gossip.
pub const CAP_BLOB_ANNOUNCE_ALPN: &[u8] = b"peat/blob-announce/1";

/// Wire-format version for [`BlobAnnounce`]. Bump on any incompatible change;
/// `decode` rejects mismatches so a future version can't be misread.
pub const ANNOUNCE_VERSION: u8 = 1;

/// Default time-to-live (relay hops) for a freshly-originated announce. Small
/// on purpose: epidemic re-broadcast plus the holder-announces-on-acquire loop
/// converge without deep flooding, and a low TTL bounds amplification.
pub const DEFAULT_ANNOUNCE_TTL: u8 = 3;

/// Upper bound on hashes carried in a single announce frame, so a malformed or
/// hostile peer can't force an unbounded allocation on decode.
const MAX_HASHES: usize = 4096;
/// Upper bound on holder addresses in a single announce frame.
const MAX_ADDRS: usize = 16;

/// An announcement that `holder` holds one or more blobs, plus the addresses at
/// which `holder` can be reached.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct BlobAnnounce {
    /// The endpoint that holds the announced blobs.
    pub holder: EndpointId,
    /// Socket addresses (`host:port`) at which `holder` accepts connections.
    /// May be empty for a relayed announce whose origin address is unknown.
    pub addrs: Vec<String>,
    /// Hashes of the blobs `holder` is announcing.
    pub hashes: Vec<BlobHash>,
    /// Remaining relay hops. A receiver re-broadcasts with `ttl - 1` while
    /// `ttl > 0`, and does not re-broadcast at `0`.
    pub ttl: u8,
}

/// Error decoding a [`BlobAnnounce`] off the wire.
#[derive(Debug, thiserror::Error, PartialEq, Eq)]
pub enum AnnounceDecodeError {
    /// Buffer ended before a field was fully read.
    #[error("truncated blob-announce frame (need {need} bytes at offset {at}, have {have})")]
    Truncated { at: usize, need: usize, have: usize },
    /// Version byte did not match [`ANNOUNCE_VERSION`].
    #[error("unsupported blob-announce version: {0} (expected {ANNOUNCE_VERSION})")]
    Version(u8),
    /// A declared count exceeded the per-frame safety bound.
    #[error("blob-announce {field} count {count} exceeds maximum {max}")]
    TooMany {
        field: &'static str,
        count: usize,
        max: usize,
    },
    /// A length-prefixed string was not valid UTF-8.
    #[error("blob-announce {field} is not valid UTF-8")]
    Utf8 { field: &'static str },
    /// The holder EndpointId bytes were not a valid key.
    #[error("blob-announce holder id is not a valid EndpointId")]
    HolderId,
}

impl BlobAnnounce {
    /// Decrement the TTL, returning `None` once it reaches zero (do not relay).
    pub fn relayed(&self) -> Option<BlobAnnounce> {
        if self.ttl == 0 {
            return None;
        }
        Some(BlobAnnounce {
            holder: self.holder,
            addrs: self.addrs.clone(),
            hashes: self.hashes.clone(),
            ttl: self.ttl - 1,
        })
    }

    /// Serialize to the version-1 wire format.
    pub fn encode(&self) -> Vec<u8> {
        let mut out = Vec::with_capacity(2 + 32 + 1 + self.hashes.len() * 34);
        out.push(ANNOUNCE_VERSION);
        out.push(self.ttl);
        out.extend_from_slice(self.holder.as_bytes());

        // addrs (count bounded to u8 / MAX_ADDRS on encode for symmetry).
        let addr_count = self.addrs.len().min(MAX_ADDRS);
        out.push(addr_count as u8);
        for addr in self.addrs.iter().take(addr_count) {
            let bytes = addr.as_bytes();
            // Addr strings are short (host:port); clamp to u8 length.
            let len = bytes.len().min(u8::MAX as usize);
            out.push(len as u8);
            out.extend_from_slice(&bytes[..len]);
        }

        // hashes
        let hash_count = self.hashes.len().min(MAX_HASHES);
        out.extend_from_slice(&(hash_count as u16).to_le_bytes());
        for hash in self.hashes.iter().take(hash_count) {
            let bytes = hash.as_hex().as_bytes();
            let len = bytes.len().min(u16::MAX as usize);
            out.extend_from_slice(&(len as u16).to_le_bytes());
            out.extend_from_slice(&bytes[..len]);
        }
        out
    }

    /// Parse a version-1 wire frame.
    pub fn decode(buf: &[u8]) -> Result<BlobAnnounce, AnnounceDecodeError> {
        let mut cur = Cursor::new(buf);

        let version = cur.u8()?;
        if version != ANNOUNCE_VERSION {
            return Err(AnnounceDecodeError::Version(version));
        }
        let ttl = cur.u8()?;
        let holder_bytes = cur.array32()?;
        let holder =
            EndpointId::from_bytes(&holder_bytes).map_err(|_| AnnounceDecodeError::HolderId)?;

        let addr_count = cur.u8()? as usize;
        if addr_count > MAX_ADDRS {
            return Err(AnnounceDecodeError::TooMany {
                field: "addr",
                count: addr_count,
                max: MAX_ADDRS,
            });
        }
        let mut addrs = Vec::with_capacity(addr_count);
        for _ in 0..addr_count {
            let len = cur.u8()? as usize;
            let bytes = cur.take(len)?;
            addrs.push(
                std::str::from_utf8(bytes)
                    .map_err(|_| AnnounceDecodeError::Utf8 { field: "addr" })?
                    .to_string(),
            );
        }

        let hash_count = cur.u16()? as usize;
        if hash_count > MAX_HASHES {
            return Err(AnnounceDecodeError::TooMany {
                field: "hash",
                count: hash_count,
                max: MAX_HASHES,
            });
        }
        let mut hashes = Vec::with_capacity(hash_count);
        for _ in 0..hash_count {
            let len = cur.u16()? as usize;
            let bytes = cur.take(len)?;
            let hex = std::str::from_utf8(bytes)
                .map_err(|_| AnnounceDecodeError::Utf8 { field: "hash" })?;
            hashes.push(BlobHash::from_hex(hex));
        }

        Ok(BlobAnnounce {
            holder,
            addrs,
            hashes,
            ttl,
        })
    }
}

/// Minimal forward-only byte cursor with bounds-checked reads.
struct Cursor<'a> {
    buf: &'a [u8],
    pos: usize,
}

impl<'a> Cursor<'a> {
    fn new(buf: &'a [u8]) -> Self {
        Cursor { buf, pos: 0 }
    }

    fn take(&mut self, n: usize) -> Result<&'a [u8], AnnounceDecodeError> {
        if self.pos + n > self.buf.len() {
            return Err(AnnounceDecodeError::Truncated {
                at: self.pos,
                need: n,
                have: self.buf.len() - self.pos,
            });
        }
        let out = &self.buf[self.pos..self.pos + n];
        self.pos += n;
        Ok(out)
    }

    fn u8(&mut self) -> Result<u8, AnnounceDecodeError> {
        Ok(self.take(1)?[0])
    }

    fn u16(&mut self) -> Result<u16, AnnounceDecodeError> {
        let b = self.take(2)?;
        Ok(u16::from_le_bytes([b[0], b[1]]))
    }

    fn array32(&mut self) -> Result<[u8; 32], AnnounceDecodeError> {
        let b = self.take(32)?;
        let mut out = [0u8; 32];
        out.copy_from_slice(b);
        Ok(out)
    }
}

/// Cap on a single announce frame (1 MiB). Bounds the receive-side allocation
/// against a malformed or hostile peer; well above any legitimate announce
/// (`MAX_HASHES` × a hex hash + overhead).
const MAX_ANNOUNCE_FRAME: usize = 1 << 20;

/// Per-step I/O timeout on the receive side. Without it, a peer that opens the
/// ALPN stream and then stalls (never sends the length prefix, or sends a
/// length then withholds the body) would park the handler task indefinitely —
/// a trivial resource-exhaustion vector for any formation member.
const ANNOUNCE_IO_TIMEOUT: Duration = Duration::from_secs(10);

/// How much of an announce to trust, given the peer that actually delivered it.
///
/// Announces are only formation-gated at the QUIC layer; the `holder` field is
/// otherwise attacker-chosen, so a compromised member could otherwise forge
/// holdings and inject dialing addresses for arbitrary third-party endpoints
/// (ADR-0006 defense-in-depth; peat-node#170). This caps the blast radius:
///
/// - **First-party** (`holder == from`): trusted fully, including the carried
///   addresses — we are talking directly to the holder.
/// - **Relayed** (`holder != from`, `ttl < DEFAULT`): the holding is recorded
///   (useful for discovery) but the addresses are NOT trusted — a relay can't
///   introduce third-party dialing hints. The holder's real address is learned
///   from the holder's own first-party announce when it reaches a direct peer.
/// - **Forged origin** (`holder != from`, `ttl == DEFAULT`): an originator may
///   only announce itself, so a full-TTL announce for someone else is rejected.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum AnnounceTrust {
    /// Drop entirely — forged origin claim.
    Reject,
    /// Record holdings and learn addresses.
    HoldingsAndAddrs,
    /// Record holdings only; ignore the carried addresses.
    HoldingsOnly,
}

/// Classify an announce by who delivered it. See [`AnnounceTrust`].
pub(crate) fn classify_announce(holder: EndpointId, from: EndpointId, ttl: u8) -> AnnounceTrust {
    if holder == from {
        AnnounceTrust::HoldingsAndAddrs
    } else if ttl == DEFAULT_ANNOUNCE_TTL {
        AnnounceTrust::Reject
    } else {
        AnnounceTrust::HoldingsOnly
    }
}

/// Shared slot wiring the [`BlobAnnounceHandler`] to its blob store.
///
/// The handler is registered into the store's iroh `Router` *before* the store
/// exists, so it can't hold an `Arc<NetworkedIrohBlobStore>`. Construction code
/// hands it this `Arc<OnceLock<…>>` and fills it with a `Weak` ref once the
/// store is built (same cycle-break as the sync handler). `Weak` so the handler
/// never keeps the store alive.
pub(crate) type BlobStoreSlot = Arc<OnceLock<Weak<NetworkedIrohBlobStore>>>;

/// Iroh `ProtocolHandler` for `peat/blob-announce/1`.
///
/// On an inbound announce it applies the holding to the store's provider index
/// (`NetworkedIrohBlobStore::apply_announce`), which also drives bounded
/// re-broadcast of newly-learned holdings.
pub(crate) struct BlobAnnounceHandler {
    store: BlobStoreSlot,
}

impl std::fmt::Debug for BlobAnnounceHandler {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("BlobAnnounceHandler").finish()
    }
}

impl BlobAnnounceHandler {
    /// Build a handler bound to `store`, a slot the construction code fills with
    /// a `Weak<NetworkedIrohBlobStore>` once the store exists.
    pub(crate) fn new(store: BlobStoreSlot) -> Self {
        Self { store }
    }
}

impl iroh::protocol::ProtocolHandler for BlobAnnounceHandler {
    async fn accept(&self, connection: Connection) -> Result<(), AcceptError> {
        let from = connection.remote_id();
        if let Err(e) = handle_announce(&connection, &self.store).await {
            warn!(peer = %from.fmt_short(), error = %e, "blob-announce protocol error");
        }
        Ok(())
    }
}

/// Read one announce frame, apply it, and ack. The announcer opens the bi
/// stream (mirrors the client below); we accept it, read a length-prefixed
/// frame, ack a single byte (so the announcer knows it landed before dropping
/// the connection), then apply.
async fn handle_announce(connection: &Connection, store: &BlobStoreSlot) -> anyhow::Result<()> {
    let from = connection.remote_id();
    let (mut send, mut recv) = timeout(ANNOUNCE_IO_TIMEOUT, connection.accept_bi())
        .await
        .map_err(|_| anyhow::anyhow!("timed out waiting for announce stream"))??;

    let mut len_buf = [0u8; 4];
    timeout(ANNOUNCE_IO_TIMEOUT, recv.read_exact(&mut len_buf))
        .await
        .map_err(|_| anyhow::anyhow!("timed out reading announce length"))??;
    let len = u32::from_le_bytes(len_buf) as usize;
    if len > MAX_ANNOUNCE_FRAME {
        anyhow::bail!("announce frame too large: {len} bytes (max {MAX_ANNOUNCE_FRAME})");
    }
    let mut buf = vec![0u8; len];
    timeout(ANNOUNCE_IO_TIMEOUT, recv.read_exact(&mut buf))
        .await
        .map_err(|_| anyhow::anyhow!("timed out reading announce body"))??;

    // Ack before applying so the sender can drop the connection promptly.
    send.write_all(&[1u8]).await?;
    send.finish()?;

    let announce = BlobAnnounce::decode(&buf)?;
    if let Some(store) = store.get().and_then(Weak::upgrade) {
        store.apply_announce(announce, from).await;
    }
    Ok(())
}

/// Client: send `announce` to `peer` over a fresh `peat/blob-announce/1`
/// connection and wait for the one-byte ack. Errors if the peer doesn't speak
/// the ALPN (old node) or is unreachable — callers treat that as best-effort.
pub async fn send_announce(
    endpoint: &Endpoint,
    peer: EndpointId,
    announce: &BlobAnnounce,
) -> anyhow::Result<()> {
    let connection = endpoint.connect(peer, CAP_BLOB_ANNOUNCE_ALPN).await?;
    let (mut send, mut recv) = connection.open_bi().await?;

    let bytes = announce.encode();
    send.write_all(&(bytes.len() as u32).to_le_bytes()).await?;
    send.write_all(&bytes).await?;
    send.finish()?;

    // Best-effort ack — delivery is the goal, not the ack itself.
    let mut ack = [0u8; 1];
    let _ = recv.read_exact(&mut ack).await;
    connection.close(0u32.into(), b"done");
    Ok(())
}

/// Fire-and-forget: announce `announce` to every peer in `peers`, each on its
/// own task so the caller (e.g. `fetch_blob`, or a re-broadcast) is never
/// blocked on the network. Failures (old node / unreachable) are logged at
/// debug and otherwise ignored — graceful degradation.
pub(crate) fn spawn_announce(endpoint: Endpoint, peers: Vec<EndpointId>, announce: BlobAnnounce) {
    for peer in peers {
        let endpoint = endpoint.clone();
        let announce = announce.clone();
        tokio::spawn(async move {
            if let Err(e) = send_announce(&endpoint, peer, &announce).await {
                debug!(peer = %peer.fmt_short(), error = %e, "blob-announce send failed");
            }
        });
    }
}

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

    fn holder() -> EndpointId {
        // A deterministic, valid EndpointId for round-trip tests.
        let secret = iroh::SecretKey::from_bytes(&[7u8; 32]);
        secret.public()
    }

    fn other() -> EndpointId {
        iroh::SecretKey::from_bytes(&[9u8; 32]).public()
    }

    #[test]
    fn classify_first_party_trusts_addrs_at_any_ttl() {
        let h = holder();
        // holder == from → trust everything, regardless of TTL.
        assert_eq!(
            classify_announce(h, h, DEFAULT_ANNOUNCE_TTL),
            AnnounceTrust::HoldingsAndAddrs
        );
        assert_eq!(classify_announce(h, h, 1), AnnounceTrust::HoldingsAndAddrs);
    }

    #[test]
    fn classify_rejects_forged_origin() {
        // Full-TTL (originated) announce claiming a holder other than the
        // sender is a forgery.
        assert_eq!(
            classify_announce(holder(), other(), DEFAULT_ANNOUNCE_TTL),
            AnnounceTrust::Reject
        );
    }

    #[test]
    fn classify_relayed_records_holdings_without_addrs() {
        // Relayed (ttl < DEFAULT) third-party holding: discover it, but don't
        // trust the relay's third-party addresses.
        assert_eq!(
            classify_announce(holder(), other(), DEFAULT_ANNOUNCE_TTL - 1),
            AnnounceTrust::HoldingsOnly
        );
    }

    #[test]
    fn round_trips_full_frame() {
        let a = BlobAnnounce {
            holder: holder(),
            addrs: vec!["10.0.0.10:51071".to_string(), "[::1]:51071".to_string()],
            hashes: vec![
                BlobHash::from_hex("deadbeef"),
                BlobHash::from_hex("cafef00d"),
            ],
            ttl: DEFAULT_ANNOUNCE_TTL,
        };
        let decoded = BlobAnnounce::decode(&a.encode()).expect("decode");
        assert_eq!(a, decoded);
    }

    #[test]
    fn round_trips_empty_addrs_and_hashes() {
        let a = BlobAnnounce {
            holder: holder(),
            addrs: vec![],
            hashes: vec![],
            ttl: 0,
        };
        assert_eq!(a, BlobAnnounce::decode(&a.encode()).expect("decode"));
    }

    #[test]
    fn rejects_wrong_version() {
        let mut bytes = BlobAnnounce {
            holder: holder(),
            addrs: vec![],
            hashes: vec![],
            ttl: 1,
        }
        .encode();
        bytes[0] = 99;
        assert_eq!(
            BlobAnnounce::decode(&bytes),
            Err(AnnounceDecodeError::Version(99))
        );
    }

    #[test]
    fn rejects_truncated_frame() {
        let bytes = BlobAnnounce {
            holder: holder(),
            addrs: vec!["10.0.0.10:51071".to_string()],
            hashes: vec![BlobHash::from_hex("deadbeef")],
            ttl: 2,
        }
        .encode();
        // Drop the last byte — the final hash string is now short.
        let truncated = &bytes[..bytes.len() - 1];
        assert!(matches!(
            BlobAnnounce::decode(truncated),
            Err(AnnounceDecodeError::Truncated { .. })
        ));
    }

    #[test]
    fn relayed_decrements_ttl_and_stops_at_zero() {
        let a = BlobAnnounce {
            holder: holder(),
            addrs: vec![],
            hashes: vec![BlobHash::from_hex("deadbeef")],
            ttl: 1,
        };
        let once = a.relayed().expect("ttl 1 -> relay");
        assert_eq!(once.ttl, 0);
        assert_eq!(once.hashes, a.hashes);
        assert!(once.relayed().is_none(), "ttl 0 must not relay");
    }
}
