use std::collections::BTreeSet;

use crate::{
    app::{AppResult, AppaService, MAX_CONCURRENT_PEER_REQUESTS, SyncContext},
    app::{
        manifest::{ManifestBuild, build_manifest_with_stages, report_local_changes},
        materialize::{
            RemoteManifestResult, apply_remote_manifest, recover_pending_materialization,
        },
    },
    domain::{AuditEvent, AuditEventKind, FolderRoster, manifest_root_hash},
    iroh::{FolderSession, MAX_AUDIT_EVENTS_PER_RESPONSE, NodeHost, PeerSummaryResponse},
    storage::FolderConfig,
};
use futures_util::{StreamExt, stream};

impl AppaService {
    #[cfg(test)]
    pub(super) async fn sync_with_node(
        &self,
        folder: &FolderConfig,
        node: &NodeHost,
    ) -> AppResult<usize> {
        self.sync_with_node_using_local_scan(folder, node, true)
            .await
    }

    pub(super) async fn sync_with_node_using_local_scan(
        &self,
        folder: &FolderConfig,
        node: &NodeHost,
        should_scan_local_files: bool,
    ) -> AppResult<usize> {
        let context = SyncContext {
            folder,
            node,
            state_store: &self.state_store,
            now: time::OffsetDateTime::now_utc(),
        };
        recover_pending_materialization(&context).await?;
        self.save_lan_discovered_peers(folder, node)?;
        self.save_discovered_peers(folder, node)?;
        let local_roster = self.load_roster(folder.id)?;
        let can_publish_changes = folder.mode.can_send()
            && local_roster.can_publish(&node.endpoint_address().id.to_string());
        self.register_sync_folder(
            folder,
            node,
            &local_roster,
            self.state_store.load_manifest(folder.id)?,
        )
        .await?;
        let manifest_build = if should_scan_local_files && can_publish_changes {
            build_manifest_with_stages(&context, |manifest| async move {
                self.publish_staged_manifest(folder, node, manifest).await
            })
            .await?
        } else {
            ManifestBuild {
                manifest: self.state_store.load_manifest(folder.id)?,
                changed_paths: Vec::new(),
            }
        };
        report_local_changes(&manifest_build.changed_paths);
        let has_local_changes = !manifest_build.changed_paths.is_empty();
        let mut local_manifest = manifest_build.manifest;
        self.register_sync_folder(folder, node, &local_roster, local_manifest.clone())
            .await?;
        self.persist_published_manifest(node, &local_manifest)
            .await?;
        if has_local_changes && can_publish_changes {
            let report = node
                .announce_manifest(
                    self.state_store.peers(folder.id)?,
                    folder.id,
                    folder.capability.clone(),
                    manifest_root_hash(&local_manifest)?,
                )
                .await;
            tracing::debug!(folder = %folder.name, announced_count = report.announced_count, unavailable_count = report.unavailable_count, "Announced local manifest changes");
        }
        let mut synchronized_count = 0;
        let mut attempted_peer_ids = BTreeSet::new();
        loop {
            let peers: Vec<_> = self
                .state_store
                .peers(folder.id)?
                .into_iter()
                .filter(|peer| attempted_peer_ids.insert(peer.id))
                .collect();
            if peers.is_empty() {
                break;
            }
            let peer_summaries = request_peer_summaries(
                node,
                peers,
                folder.id,
                folder.capability.clone(),
                self.state_store.join_invite(folder.id)?,
            )
            .await;
            for (peer, response) in peer_summaries {
                let Some(remote_result) = self
                    .synchronize_peer(&context, &local_manifest, peer, response)
                    .await?
                else {
                    continue;
                };
                local_manifest = remote_result.manifest;
                synchronized_count += remote_result.synchronized_file_count;
                self.state_store.save_manifest(&local_manifest)?;
                node.publish_manifest(local_manifest.clone()).await?;
            }
        }
        self.save_discovered_peers(folder, node)?;
        self.save_lan_discovered_peers(folder, node)?;
        self.state_store.record_successful_sync(folder.id)?;
        Ok(synchronized_count)
    }

    async fn publish_staged_manifest(
        &self,
        folder: &FolderConfig,
        node: &NodeHost,
        manifest: crate::domain::Manifest,
    ) -> AppResult<()> {
        self.persist_published_manifest(node, &manifest).await?;
        let report = node
            .announce_manifest(
                self.state_store.peers(folder.id)?,
                folder.id,
                folder.capability.clone(),
                manifest_root_hash(&manifest)?,
            )
            .await;
        tracing::debug!(folder = %folder.name, announced_count = report.announced_count, unavailable_count = report.unavailable_count, "Announced staged local manifest changes");
        Ok(())
    }

    async fn register_sync_folder(
        &self,
        folder: &FolderConfig,
        node: &NodeHost,
        roster: &FolderRoster,
        manifest: crate::domain::Manifest,
    ) -> AppResult<()> {
        node.register_folder(
            FolderSession {
                folder_id: folder.id,
                capability: folder.capability.clone(),
                active_member_ids: roster.active_device_ids(),
                roster: roster.clone(),
                peer_endpoints: self.state_store.peers(folder.id)?,
                audit_events: self.state_store.audit_events(folder.id)?,
            },
            manifest,
        )
        .await
    }

    async fn persist_published_manifest(
        &self,
        node: &NodeHost,
        manifest: &crate::domain::Manifest,
    ) -> AppResult<()> {
        let audit_event = self.manifest_audit_event(manifest, time::OffsetDateTime::now_utc())?;
        self.state_store
            .save_manifest_with_audit(manifest, &audit_event)?;
        node.publish_audit_events(
            manifest.folder_id,
            self.state_store.audit_events(manifest.folder_id)?,
        )
        .await?;
        node.publish_manifest(manifest.clone()).await
    }

    fn manifest_audit_event(
        &self,
        manifest: &crate::domain::Manifest,
        now: time::OffsetDateTime,
    ) -> AppResult<AuditEvent> {
        let identity = self.paths.load_identity()?;
        let roster = self.load_roster(manifest.folder_id)?;
        let author_device_id = identity.public().to_string();
        if !roster.can_publish(&author_device_id) {
            anyhow::bail!("a non-writing member cannot sign a manifest audit event");
        }
        let (sequence, parent_hash) = self
            .state_store
            .next_audit_sequence_and_parent(manifest.folder_id, &author_device_id)?;
        let event = AuditEvent::create(
            manifest.folder_id,
            author_device_id,
            sequence,
            parent_hash,
            AuditEventKind::ManifestCommitted,
            Some(manifest_root_hash(manifest)?),
            roster.hash()?,
            roster.clone(),
            now,
        );
        event.sign(&identity)
    }

    pub(super) fn save_discovered_peers(
        &self,
        folder: &FolderConfig,
        node: &NodeHost,
    ) -> AppResult<()> {
        for peer in node.take_discovered_peers() {
            if peer.folder_id == folder.id {
                self.state_store.save_peer(folder.id, &peer.endpoint)?;
                self.enroll_discovered_member(folder.id, &peer.endpoint.id.to_string())?;
            }
        }
        Ok(())
    }

    fn save_lan_discovered_peers(&self, folder: &FolderConfig, node: &NodeHost) -> AppResult<()> {
        let active_member_ids = self.active_member_ids(folder.id)?;
        for peer in node.lan_discovered_peers() {
            let device_id = peer.id.to_string();
            if active_member_ids.contains(&device_id) {
                tracing::debug!(folder = %folder.name, peer = %device_id, "Refreshed trusted local peer route");
                self.state_store.save_peer(folder.id, &peer)?;
            } else {
                tracing::debug!(folder = %folder.name, peer = %device_id, "Ignoring untrusted local peer");
            }
        }
        Ok(())
    }

    async fn save_peer_response(
        &self,
        folder: &FolderConfig,
        node: &NodeHost,
        peer: &iroh::EndpointAddr,
        roster: &FolderRoster,
        peer_endpoints: Vec<iroh::EndpointAddr>,
        capability: &str,
    ) -> AppResult<()> {
        if roster.folder_id != folder.id {
            anyhow::bail!("peer returned a roster for another folder");
        }
        if roster.capability != capability {
            anyhow::bail!("peer capability is not bound to its signed roster");
        }
        let current_roster = self.load_roster(folder.id)?;
        if !current_roster.contains_member(&peer.id.to_string()) {
            anyhow::bail!("peer is not an active member of the local roster");
        }
        if roster.owner_device_id != current_roster.owner_device_id {
            anyhow::bail!("peer returned a roster with another owner");
        }
        if roster.epoch < current_roster.epoch {
            self.state_store.save_peer(folder.id, peer)?;
            self.save_roster_peers(folder.id, &current_roster, peer_endpoints)?;
            return self.publish_peer_endpoints(folder.id, node).await;
        }
        if roster.epoch == current_roster.epoch && roster != &current_roster {
            anyhow::bail!("peer returned a conflicting folder roster epoch");
        }
        self.state_store.save_peer(folder.id, peer)?;
        let accepted_new_roster = self.state_store.save_roster_if_newer(roster)?;
        self.save_roster_peers(folder.id, roster, peer_endpoints)?;
        self.publish_peer_endpoints(folder.id, node).await?;
        if accepted_new_roster && capability != folder.capability {
            self.state_store.update_capability(folder.id, capability)?;
            tracing::info!(folder = %folder.name, "Updated folder capability after member revocation");
        }
        Ok(())
    }

    async fn publish_peer_endpoints(
        &self,
        folder_id: crate::domain::FolderId,
        node: &NodeHost,
    ) -> AppResult<()> {
        node.publish_peer_endpoints(folder_id, self.state_store.peers(folder_id)?)
            .await
    }

    fn save_roster_peers(
        &self,
        folder_id: crate::domain::FolderId,
        roster: &FolderRoster,
        peer_endpoints: Vec<iroh::EndpointAddr>,
    ) -> AppResult<()> {
        for peer in peer_endpoints {
            if roster.contains_member(&peer.id.to_string()) {
                self.state_store.save_peer(folder_id, &peer)?;
            }
        }
        Ok(())
    }

    async fn synchronize_peer(
        &self,
        context: &SyncContext<'_>,
        local_manifest: &crate::domain::Manifest,
        peer: iroh::EndpointAddr,
        response: anyhow::Result<PeerSummaryResponse>,
    ) -> AppResult<Option<RemoteManifestResult>> {
        let folder = context.folder;
        let PeerSummaryResponse {
            summary,
            roster,
            peer_endpoints,
            capability,
        } = match response {
            Ok(response) => response,
            Err(error) => {
                tracing::debug!(folder = %folder.name, peer = %peer.id, %error, "peer is unavailable; continuing with other peers");
                return Ok(None);
            }
        };
        if summary.folder_id != folder.id {
            tracing::warn!(folder = %folder.name, peer = %peer.id, "peer returned a summary for another folder");
            return Ok(None);
        }
        self.save_peer_response(
            folder,
            context.node,
            &peer,
            &roster,
            peer_endpoints,
            &capability,
        )
        .await?;
        self.receive_peer_audit_events(context, &peer).await?;
        if summary.root_hash == manifest_root_hash(local_manifest)? {
            tracing::debug!(folder = %folder.name, peer = %peer.id, "Folder roots match; skipping manifest transfer");
            return Ok(None);
        }
        let remote_manifest = match context
            .node
            .request_manifest_delta(
                peer.clone(),
                folder.id,
                folder.capability.clone(),
                self.state_store.join_invite(folder.id)?,
                local_manifest,
            )
            .await
        {
            Ok(manifest) => manifest,
            Err(error) => {
                tracing::debug!(folder = %folder.name, peer = %peer.id, %error, "peer did not provide its changed entries");
                return Ok(None);
            }
        };
        if remote_manifest.folder_id != folder.id {
            tracing::warn!(folder = %folder.name, peer = %peer.id, "peer returned a manifest for another folder");
            return Ok(None);
        }
        if !folder.mode.can_receive() {
            tracing::debug!(folder = %folder.name, "Ignoring remote changes for send-only folder");
            return Ok(None);
        }
        match apply_remote_manifest(context, local_manifest, &remote_manifest, peer.clone()).await {
            Ok(result) => Ok(Some(result)),
            Err(error) => {
                tracing::warn!(folder = %folder.name, peer = %peer.id, %error, "could not apply peer manifest");
                Ok(None)
            }
        }
    }

    async fn receive_peer_audit_events(
        &self,
        context: &SyncContext<'_>,
        peer: &iroh::EndpointAddr,
    ) -> AppResult<()> {
        loop {
            let events = match context
                .node
                .request_audit_events(
                    peer.clone(),
                    context.folder.id,
                    context.folder.capability.clone(),
                    self.state_store.join_invite(context.folder.id)?,
                    self.state_store
                        .audit_heads(context.folder.id)?
                        .into_iter()
                        .map(|head| (head.author_device_id, head.sequence))
                        .collect(),
                )
                .await
            {
                Ok(events) => events,
                Err(error) => {
                    tracing::debug!(folder = %context.folder.name, peer = %peer.id, %error, "peer did not provide audit events");
                    return Ok(());
                }
            };
            let is_last_page = events.len() < MAX_AUDIT_EVENTS_PER_RESPONSE;
            for event in events {
                if event.folder_id != context.folder.id {
                    anyhow::bail!("peer returned an audit event for another folder");
                }
                if let Err(error) = self.state_store.append_audit_event(&event) {
                    let detail = format!(
                        "rejected audit event from {}: {error}",
                        event.author_device_id
                    );
                    self.state_store.record_audit_fault(
                        context.folder.id,
                        &event.author_device_id,
                        &detail,
                    )?;
                    anyhow::bail!(detail);
                }
            }
            if is_last_page {
                break;
            }
        }
        context
            .node
            .publish_audit_events(
                context.folder.id,
                self.state_store.audit_events(context.folder.id)?,
            )
            .await
    }
}

async fn request_peer_summaries(
    node: &NodeHost,
    peers: Vec<iroh::EndpointAddr>,
    folder_id: crate::domain::FolderId,
    capability: String,
    invite: Option<crate::protocol::Invite>,
) -> Vec<(iroh::EndpointAddr, anyhow::Result<PeerSummaryResponse>)> {
    stream::iter(peers)
        .map(|peer| {
            let request_capability = capability.clone();
            let request_invite = invite.clone();
            async move {
                let response = node
                    .request_manifest_summary(
                        peer.clone(),
                        folder_id,
                        request_capability,
                        request_invite,
                    )
                    .await;
                (peer, response)
            }
        })
        .buffer_unordered(MAX_CONCURRENT_PEER_REQUESTS)
        .collect()
        .await
}