//! P2P sync transport trait + filesystem-shared default impl.
//!
//! The trait abstracts how an actor's operations reach other peers. The
//! default [`FilesystemSync`] uses a shared directory — every actor writes
//! its announcements as files under `<base_dir>/<peer_id>/` and reads
//! peer subdirectories to discover others. It is intentionally minimal:
//! useful for development and single-host multi-actor pilots, but NOT a
//! real P2P solution. Trigger-based replacement (Iroh, Hypercore, Radicle,
//! NATS) is documented per ADR-027 / D18.

use std::path::PathBuf;

/// Failures emitted by sync-backend operations.
#[derive(Debug, thiserror::Error)]
pub enum SyncError {
    /// Disk IO failure with operation context.
    #[error("io in {context}: {source}")]
    Io {
        context: String,
        #[source]
        source: std::io::Error,
    },

    /// The requested op id was not found on any reachable peer.
    #[error("op '{0}' not available from any peer")]
    OpNotAvailable(String),
}

/// Abstraction over a P2P sync transport. Implementations announce
/// applied operations to peers, expose the current peer set, fetch op
/// bodies from peers on demand, and subscribe to a specific peer's
/// announcement stream.
pub trait SyncBackend {
    /// Announce that the local actor has applied `op_id`.
    ///
    /// # Errors
    ///
    /// Backend-specific. The filesystem default returns IO errors; remote
    /// transports may also return network failures.
    fn announce(&mut self, op_id: &str) -> Result<(), SyncError>;

    /// Currently-known peer ids (anyone the backend believes is reachable).
    fn discover_peers(&self) -> Vec<String>;

    /// Fetch the op body for `op_id` from peers. Returns `Ok(None)` when
    /// the op was found nowhere; returns `Err` on transport failure.
    ///
    /// # Errors
    ///
    /// Backend-specific transport failures.
    fn fetch_op(&self, op_id: &str) -> Result<Option<Vec<u8>>, SyncError>;

    /// Begin tracking announcements from `peer_id`. Idempotent.
    ///
    /// # Errors
    ///
    /// Backend-specific; the filesystem default returns IO errors when
    /// the peer directory cannot be probed.
    fn subscribe_peer(&mut self, peer_id: &str) -> Result<(), SyncError>;

    /// Op ids `peer_id` has announced — the enumeration a consumer needs to
    /// know what to [`fetch_op`](Self::fetch_op) and ingest. Returns an
    /// empty vec when the peer has announced nothing.
    ///
    /// # Errors
    ///
    /// Backend-specific transport failures while listing announcements.
    fn announced_ops(&self, peer_id: &str) -> Result<Vec<String>, SyncError>;
}

/// Filesystem-shared sync backend — default impl with zero external deps.
///
/// Each peer is a subdirectory under `base_dir`; the local peer's
/// announcements are written as `<base_dir>/<peer_id>/<op_id>.announce`.
pub struct FilesystemSync {
    base_dir: PathBuf,
    peer_id: String,
    subscribed: Vec<String>,
}

impl FilesystemSync {
    /// Construct a filesystem-shared sync rooted at `base_dir` for the
    /// local actor identified by `peer_id`. The local peer directory is
    /// created eagerly so subsequent `announce` calls cannot race against
    /// directory creation.
    ///
    /// # Errors
    ///
    /// IO error when the local peer directory cannot be created.
    pub fn new(
        base_dir: impl Into<PathBuf>,
        peer_id: impl Into<String>,
    ) -> Result<Self, SyncError> {
        let base_dir = base_dir.into();
        let peer_id = peer_id.into();
        let local = base_dir.join(&peer_id);
        std::fs::create_dir_all(&local).map_err(|e| SyncError::Io {
            context: format!("creating local peer dir {}", local.display()),
            source: e,
        })?;
        Ok(Self {
            base_dir,
            peer_id,
            subscribed: Vec::new(),
        })
    }

    fn announce_path(&self, op_id: &str) -> PathBuf {
        self.base_dir
            .join(&self.peer_id)
            .join(format!("{op_id}.announce"))
    }

    fn peer_dir(&self, peer_id: &str) -> PathBuf {
        self.base_dir.join(peer_id)
    }

    /// Write an op body under the local peer directory so that subsequent
    /// `fetch_op` calls can retrieve it. Storage is intentionally simple —
    /// one file per op, named by op id.
    ///
    /// # Errors
    ///
    /// IO error on write.
    pub fn publish_body(&self, op_id: &str, body: &[u8]) -> Result<(), SyncError> {
        let path = self
            .base_dir
            .join(&self.peer_id)
            .join(format!("{op_id}.body"));
        std::fs::write(&path, body).map_err(|e| SyncError::Io {
            context: format!("writing body to {}", path.display()),
            source: e,
        })
    }
}

impl SyncBackend for FilesystemSync {
    fn announce(&mut self, op_id: &str) -> Result<(), SyncError> {
        let path = self.announce_path(op_id);
        std::fs::write(&path, b"").map_err(|e| SyncError::Io {
            context: format!("writing announce {}", path.display()),
            source: e,
        })
    }

    fn discover_peers(&self) -> Vec<String> {
        let Ok(entries) = std::fs::read_dir(&self.base_dir) else {
            return Vec::new();
        };
        let mut peers: Vec<String> = entries
            .filter_map(|e| e.ok())
            .filter(|e| e.file_type().map(|t| t.is_dir()).unwrap_or(false))
            .filter_map(|e| e.file_name().into_string().ok())
            .filter(|name| name != &self.peer_id)
            .collect();
        peers.sort();
        peers
    }

    fn fetch_op(&self, op_id: &str) -> Result<Option<Vec<u8>>, SyncError> {
        let body_name = format!("{op_id}.body");
        // Look in local peer dir first (so an actor can fetch its own
        // bodies), then in subscribed peers, then in any peer dir.
        let candidates: Vec<&str> = std::iter::once(self.peer_id.as_str())
            .chain(self.subscribed.iter().map(String::as_str))
            .collect();
        for peer in candidates {
            let path = self.peer_dir(peer).join(&body_name);
            if path.exists() {
                let bytes = std::fs::read(&path).map_err(|e| SyncError::Io {
                    context: format!("reading op body {}", path.display()),
                    source: e,
                })?;
                return Ok(Some(bytes));
            }
        }
        // Last-resort scan of every peer dir.
        for peer in self.discover_peers() {
            let path = self.peer_dir(&peer).join(&body_name);
            if path.exists() {
                let bytes = std::fs::read(&path).map_err(|e| SyncError::Io {
                    context: format!("reading op body {}", path.display()),
                    source: e,
                })?;
                return Ok(Some(bytes));
            }
        }
        Ok(None)
    }

    fn announced_ops(&self, peer_id: &str) -> Result<Vec<String>, SyncError> {
        let dir = self.peer_dir(peer_id);
        let entries = match std::fs::read_dir(&dir) {
            Ok(e) => e,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
            Err(e) => {
                return Err(SyncError::Io {
                    context: format!("listing announcements in {}", dir.display()),
                    source: e,
                })
            }
        };
        let mut ids: Vec<String> = entries
            .filter_map(|e| e.ok())
            .filter_map(|e| e.file_name().into_string().ok())
            .filter_map(|name| name.strip_suffix(".announce").map(str::to_owned))
            .collect();
        ids.sort();
        Ok(ids)
    }

    fn subscribe_peer(&mut self, peer_id: &str) -> Result<(), SyncError> {
        let dir = self.peer_dir(peer_id);
        if !dir.exists() {
            return Err(SyncError::Io {
                context: format!("subscribing to {}", dir.display()),
                source: std::io::Error::new(
                    std::io::ErrorKind::NotFound,
                    format!("peer directory {} does not exist", dir.display()),
                ),
            });
        }
        if !self.subscribed.iter().any(|p| p == peer_id) {
            self.subscribed.push(peer_id.to_owned());
        }
        Ok(())
    }
}

#[cfg(test)]
#[test]
fn filesystem_default() {
    let tmp = tempfile::tempdir().expect("tempdir");
    let base = tmp.path();

    let mut alice = FilesystemSync::new(base, "alice").expect("init alice");
    let mut bob = FilesystemSync::new(base, "bob").expect("init bob");

    // Each side discovers the other.
    let mut alice_peers = alice.discover_peers();
    alice_peers.sort();
    assert_eq!(alice_peers, vec!["bob".to_string()]);
    let mut bob_peers = bob.discover_peers();
    bob_peers.sort();
    assert_eq!(bob_peers, vec!["alice".to_string()]);

    // Alice publishes an op body and announces it.
    let op_id = "op-0001";
    let body = b"hello from alice";
    alice.publish_body(op_id, body).expect("publish ok");
    alice.announce(op_id).expect("announce ok");

    // Bob subscribes to alice and fetches the body.
    bob.subscribe_peer("alice").expect("subscribe ok");
    let fetched = bob.fetch_op(op_id).expect("fetch ok").expect("op present");
    assert_eq!(fetched, body);

    // Bob fetching an unknown op returns Ok(None).
    let missing = bob.fetch_op("op-not-here").expect("fetch ok");
    assert!(missing.is_none());

    // Subscribing to a non-existent peer fails clearly.
    let err = bob
        .subscribe_peer("nobody")
        .expect_err("subscribe must fail");
    assert!(matches!(err, SyncError::Io { .. }));
}
